Carabas: Kinesis, DynamoDB, DMS - CDC record processing in different flavours

.github/workflows/tests.yml

@@ -33,6 +33,7 @@ jobs:
     env:
       OS: ${{ matrix.os }}
       PYTHON: ${{ matrix.python-version }}
+      TC_KEEPALIVE: true
 
     name: Python ${{ matrix.python-version }} on OS ${{ matrix.os }}
     steps:
@@ -56,7 +57,7 @@ jobs:
         pip install "setuptools>=64" --upgrade
 
         # Install package in editable mode.
-        pip install --use-pep517 --prefer-binary --editable=.[test,develop]
+        pip install --use-pep517 --prefer-binary --editable=.[all,test,develop]
 
     - name: Run linter and software tests
       run: |

CHANGES.md

@@ -1,6 +1,7 @@
 # Changelog
 
 ## in progress
+- Carabas: A subsystem to divert workloads to other people’s computers
 
 ## 2024-07-10 v0.0.2
 - Initial working version, supporting MQTT, AMQP, and SQLAlchemy/CrateDB

doc/backlog.rst

@@ -40,6 +40,8 @@ Iteration 2
 - [o] Examples: Add ``appsink`` example
 - [o] Improve inline docs
 - [o] Release 0.1.0
+- [o] CSV: https://github.com/alan-turing-institute/CleverCSV
+- [o] Excel & ODF: https://github.com/dimastbk/python-calamine
 
 
 ***********

doc/carabas/backlog.md

@@ -0,0 +1,21 @@
+# Carabas Backlog
+
+## Iteration +1
+- [x] Improve type mapping
+- [x] Generalize CDC event -> SQL translator
+- [ ] Only optionally display debug output of Docker build process,
+  [ ] when using `--verbose`.
+- [ ] Bring back "Zip" use, for interactive hacking
+- [ ] Distill into a Lambda variant
+- [ ] Automation!
+  - [ ] DDL: CREATE TABLE <tablename> (data OBJECT(DYNAMIC));
+  - [ ] Wrap KCL launcher into manager component
+
+## Iteration +2
+- [ ] Performance improvements (simdjson?)
+- [ ] Use SQLAlchemy for generating and submitting SQL statement
+- [ ] Improve efficiency by using bulk operations when applicable
+- [ ] is in UPDATE_ROLLBACK_COMPLETE_CLEANUP_IN_PROGRESS state and can not be updated
+- [ ] is in ROLLBACK_COMPLETE state and can not be updated.
+- [ ] Cannot create a publicly accessible DBInstance. The specified VPC has no
+  internet gateway attached.Update the VPC and then try again

doc/carabas/dms/handbook.md

@@ -0,0 +1,79 @@
+(aws-dms-handbook)=
+# AWS DMS Handbook
+
+A few useful AWSCLI commands to check the status of the DMS engine and
+relevant pipeline elements. You can also use the AWS Web Console to
+inspect and commandeer the same details.
+
+
+## Status Checks
+Display ARNs of all replication instances.
+```shell
+aws dms describe-replication-instances | jq -r '.ReplicationInstances[].ReplicationInstanceArn'
+```
+Display replication endpoints and relevant connection settings.
+```shell
+aws dms describe-endpoints
+```
+Invoke connection test on given DMS endpoint.
+```shell
+aws dms test-connection \
+  --replication-instance-arn arn:aws:dms:eu-central-1:831394476016:rep:JD2LL6OM35BJZNKZIRSOE2FXIY \
+  --endpoint-arn arn:aws:dms:eu-central-1:831394476016:endpoint:3IVDGL6E4RDNBF2LFBYF6DYV3Y
+```
+Display connection test results.
+```shell
+aws dms describe-connections
+```
+
+
+## Operations
+Enumerate all configured replication tasks with compact output.
+```shell
+aws dms describe-replication-tasks | \
+  jq '.ReplicationTasks[] | {ReplicationTaskIdentifier, ReplicationTaskArn, MigrationType, StartReplicationType, Status, StopReason, FailureMessages, ProvisionData}'
+```
+Start replication task with given ARN.
+```shell
+aws dms start-replication-task \
+  --start-replication-task-type start-replication --replication-task-arn \
+  arn:aws:dms:eu-central-1:831394476016:task:7QBLNBTPCNDEBG7CHI3WA73YFA
+```
+Stop replication task with given ARN.
+```shell
+aws dms stop-replication-task --replication-task-arn \
+  arn:aws:dms:eu-central-1:831394476016:task:7QBLNBTPCNDEBG7CHI3WA73YFA
+```
+
+
+## Logging
+To see detailed progress about the replication process, use CloudWatch to
+inspect corresponding log output.
+
+Enumerate all log groups.
+```shell
+aws logs describe-log-groups
+```
+
+Get log output history.
+```shell
+aws logs get-log-events \
+  --log-group-name dms-tasks-testdrive-dms-instance \
+  --log-stream-name dms-task-7QBLNBTPCNDEBG7CHI3WA73YFA | jq .events[].message
+```
+
+Start watching the log output using the `start-live-tail` CloudWatch operation.
+```shell
+aws logs start-live-tail --log-group-identifiers \
+    arn:aws:logs:eu-central-1:831394476016:log-group:/aws/rds/instance/testdrive-dms-postgresql-dev-db/postgresql \
+    arn:aws:logs:eu-central-1:831394476016:log-group:dms-tasks-testdrive-dms-instance
+```
+
+
+## CloudFormation
+When the CloudFormation deployment is stuck, or if you want to start from scratch,
+those commands are useful.
+```shell
+aws cloudformation continue-update-rollback --stack-name testdrive-dms-postgresql-dev
+aws cloudformation delete-stack --stack-name testdrive-dms-postgresql-dev
+```

doc/carabas/dms/index.md

@@ -0,0 +1,135 @@
+(aws-dms)=
+# Pipelines with AWS DMS
+
+_AWS DMS to Kinesis to CrateDB._
+
+## What's Inside
+- [Working with AWS DMS tasks]
+- [Using Amazon Kinesis Data Streams as a target for AWS Database Migration Service]
+- An IaC driver program based on [AWS CloudFormation] technologies using the
+  [cottonformation] Python API. It can be used to set up infrastructure on AWS
+  without much ado.
+- DMS: Full load and CDC
+- DMS Source: RDS PostgreSQL
+- DMS Target: Amazon Kinesis
+- CDC Target: CrateDB Cloud
+
+
+## AWS Infrastructure Setup
+The following walkthrough describes a full deployment of AWS DMS including
+relevant outbound data processors for demonstration purposes.
+
+In order to run it in production, you are welcome to derive from it and tweak
+it for your own purposes. YMMV. If you need support, don't hesitate to ask for
+help.
+
+### Install
+Install LorryStream.
+```shell
+pip install --upgrade 'lorrystream[carabas]'
+```
+Acquire IaC driver program.
+```shell
+wget https://github.com/daq-tools/lorrystream/raw/main/examples/aws/rds_postgresql_kinesis_lambda_oci_cratedb.py
+```
+
+### Configure
+Please configure endpoint and replication settings within the source code
+of the IaC program you just acquired, and presented next.
+
+### Deploy
+First, prepare an AWS ECR repository for publishing the OCI image including your
+downstream processor element that is consuming the replication data stream from
+Amazon Kinesis, and runs it into CrateDB. To learn about how this works, please
+visit the documentation section about the [](project:#ecr-repository).
+
+Configure CrateDB database sink address.
+```shell
+export SINK_SQLALCHEMY_URL='crate://admin:dZ..qB@example.eks1.eu-west-1.aws.cratedb.net:4200/?ssl=true'
+```
+
+Invoke the IaC driver program in order to deploy relevant resources on AWS.
+```shell
+python examples/aws/rds_postgresql_kinesis_lambda_oci_cratedb.py
+```
+
+After deployment succeeded, you will be presented a corresponding response including
+relevant information about entrypoints to the software stack you've just created.
+```text
+Result of CloudFormation deployment:
+psql command: psql "postgresql://dynapipe:secret11@testdrive-dms-postgresql-dev-db.czylftvqn1ed.eu-central-1.rds.amazonaws.com:5432/postgres"
+RDS Instance ARN: arn:aws:rds:eu-central-1:831394476016:db:testdrive-dms-postgresql-dev-db
+Stream ARN: arn:aws:kinesis:eu-central-1:831394476016:stream/testdrive-dms-postgresql-dev-stream
+Replication ARN: arn:aws:dms:eu-central-1:831394476016:replication-config:EAM3JEHXGBGZBPN5PLON7NPDEE
+```
+
+:::{note}
+Please note this is a demonstration stack, deviating from typical real-world situations.
+
+- Contrary to this stack, which includes an RDS PostgreSQL instance, a database instance
+  will already be up and running, so the remaining task is to just configure the Kinesis
+  Data Stream and consume it.
+
+- Contrary to this stack, which uses AWS Lambda to host the downstream processor element,
+  when aiming for better cost-effectiveness, you will run corresponding code on a dedicated
+  computing environment.
+:::
+
+
+## Operations
+Please consult the [](project:#aws-dms-handbook) to learn about commands
+suitable for operating the AWS DMS engine.
+
+:::{toctree}
+:hidden:
+
+handbook
+:::
+
+
+
+## Usage
+
+### DMS
+AWS DMS provides `full-load` and `full-load-and-cdc` migration types.
+For a `full-load-and-cdc` task, AWS DMS migrates table data, and then applies
+data changes that occur on the source, automatically establishing continuous
+replication.
+
+When starting a replication task using [StartReplicationTask], you can use those
+possible values for `--start-replication-task-type`, see also [start-replication-task]:
+
+:start-replication:
+    The only valid value for the first run of the task when the migration type is
+    `full-load` or `full-load-and-cdc`
+
+:resume-processing:
+    Not applicable for any full-load task, because you can't resume partially loaded
+    tables during the full load phase. Use it to replicate the changes from the last
+    stop position.
+
+:reload-target:
+    For a `full-load-and-cdc` task, load all the tables again, and start capturing
+    source changes.
+
+
+## Migration by DMS Source
+This section enumerates specific information to consider when aiming to use DMS
+for your database as a source element.
+
+:::{toctree}
+:maxdepth: 2
+
+postgresql
+mysql
+:::
+
+
+
+[AWS CloudFormation]: https://en.wikipedia.org/wiki/AWS_CloudFormation
+[cottonformation]: https://pypi.org/project/cottonformation/
+[StartReplicationTask]: https://docs.aws.amazon.com/dms/latest/APIReference/API_StartReplicationTask.html
+[start-replication-task]: https://docs.aws.amazon.com/cli/latest/reference/dms/start-replication-task.html
+[Using Amazon Kinesis Data Streams as a target for AWS Database Migration Service]: https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Target.Kinesis.html
+[Using object mapping to migrate data to a Kinesis data stream]: https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Target.Kinesis.html#CHAP_Target.Kinesis.ObjectMapping
+[Working with AWS DMS tasks]: https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Tasks.html

doc/carabas/dms/mysql.md

@@ -0,0 +1,4 @@
+(aws-dms-mysql)=
+# AWS DMS with MySQL/MariaDB source
+
+WIP.

doc/carabas/dms/postgresql.md

@@ -0,0 +1,57 @@
+(aws-dms-postgresql)=
+# AWS DMS with PostgreSQL source
+
+## What's Inside
+- [Using a PostgreSQL database as an AWS DMS source]
+
+### Prerequisites
+First of all, activate the `pglocical` extension on your RDS PostgreSQL instance.
+```sql
+CREATE EXTENSION pglogical;
+SELECT * FROM pg_catalog.pg_extension WHERE extname='pglogical';
+```
+
+```sql
+SHOW shared_preload_libraries;
+SELECT name, setting FROM pg_settings WHERE name in ('rds.logical_replication','shared_preload_libraries');
+```
+
+
+### Data in Source
+After that, connect to RDS PostgreSQL, and provision a little bunch of data.
+```sql
+DROP TABLE IF EXISTS foo CASCADE;
+CREATE TABLE foo (id INT PRIMARY KEY, name TEXT, age INT, attributes JSONB);
+INSERT INTO foo (id, name, age, attributes) VALUES (42, 'John', 30, '{"foo": "bar"}');
+INSERT INTO foo (id, name, age, attributes) VALUES (43, 'Jane', 31, '{"baz": "qux"}');
+```
+
+### Data in Target
+```sql
+cr> SELECT * FROM public.foo;
+```
+```postgresql
++---------------------------------------------------------------------+
+| data                                                                |
++---------------------------------------------------------------------+
+| {"age": 30, "attributes": {"foo": "bar"}, "id": 42, "name": "John"} |
+| {"age": 31, "attributes": {"baz": "qux"}, "id": 43, "name": "Jane"} |
++---------------------------------------------------------------------+
+```
+
+
+
+```sql
+UPDATE foo SET age=32 WHERE name='Jane';
+UPDATE foo SET age=33 WHERE id=43;
+UPDATE foo SET age=33 WHERE attributes->>'foo'='bar';
+UPDATE foo SET attributes = jsonb_set(attributes, '{last_name}', '"Doe"', true) WHERE name='John';
+```
+```sql
+DELETE FROM foo WHERE name='Jane';
+DELETE FROM foo WHERE name='John';
+```
+
+
+
+[Using a PostgreSQL database as an AWS DMS source]: https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Source.PostgreSQL.html

doc/carabas/index.md

@@ -0,0 +1,53 @@
+# Carabas
+
+A subsystem to divert workloads to other people's computers.
+Workloads can be whole pipelines or elements of pipelines.
+Provides blended computing environments on your fingertips.
+
+## Etymology
+- [Marquis von Carabas]
+- [Die Meisterkatze oder der gestiefelte Kater]
+- [Le Maître chat ou le Chat botté]
+- [Puss in Boots]
+
+## What's Inside
+
+### Kinesis KCL v2
+:::{toctree}
+:maxdepth: 2
+:glob:
+kcl/kinesis
+:::
+
+### DynamoDB -> Kinesis KCL v2
+:::{toctree}
+:maxdepth: 2
+:glob:
+kcl/dynamodb*
+:::
+
+### DMS -> Kinesis
+:::{toctree}
+:maxdepth: 2
+dms/index
+:::
+
+### Kinesis -> Lambda
+:::{toctree}
+:maxdepth: 2
+lambda/index
+:::
+
+
+## Development
+:::{toctree}
+:maxdepth: 2
+backlog
+research
+:::
+
+
+[Die Meisterkatze oder der gestiefelte Kater]: https://de.frwiki.wiki/wiki/Le_Ma%C3%AEtre_chat_ou_le_Chat_bott%C3%A9
+[Le Maître chat ou le Chat botté]: https://fr.wikipedia.org/wiki/Le_Ma%C3%AEtre_chat_ou_le_Chat_bott%C3%A9
+[Marquis von Carabas]: https://de.frwiki.wiki/wiki/Marquis_de_Carabas
+[Puss in Boots]: https://en.wikipedia.org/wiki/Puss_in_Boots