Carabas/DMS: Improve CloudFormation stack

- Configure ReplicationType to use `full-load-and-cdc`.
- Configure ReplicationSettings to use `EnableBeforeImage`.
- Add RDSParameterGroup to configure pgaudit, pglogical, and
  pg_stat_statements plugins.
- Configure DMS source endpoint (PostgreSQL) to use pglogical.
- Configure DMS target endpoint (Kinesis) to include all optional
  details: ControlDetails, PartitionValue, TransactionDetails,
  NullAndEmpty, TableAlterOperations, IncludeSchemaTable
- Add `RDSInstanceArn` output variable.
- Add `ReplicationArn` output variable.

Author: amotl

doc/carabas/research.md

@@ -34,3 +34,11 @@
 
 ## DMS
 - https://stackoverflow.com/questions/77995867/dynamic-tables-via-dms-kinesis-iceberg-transactional-data-lake
+- https://aws.amazon.com/blogs/database/tune-replication-performance-with-aws-dms-for-an-amazon-kinesis-data-streams-target-endpoint-part-3/
+- https://www.cockroachlabs.com/docs/stable/aws-dms
+
+## wal2json
+- https://hevodata.com/learn/pg-logical/
+- https://aws.amazon.com/blogs/database/stream-changes-from-amazon-rds-for-postgresql-using-amazon-kinesis-data-streams-and-aws-lambda/
+- https://github.com/eulerto/wal2json
+- https://docs.aws.amazon.com/AmazonRDS/latest/PostgreSQLReleaseNotes/postgresql-extensions.html#postgresql-extensions-15x

examples/aws/rds_postgresql_kinesis_lambda_oci_cratedb.py

@@ -19,6 +19,9 @@ def main():
     - CrateDB Cloud
 
     Prerequisites: Register an OCI repository.
+
+    Resources:
+    - https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/USER_VPC.WorkingWithRDSInstanceinaVPC.html
     """
 
     # Build and publish OCI image that includes the AWS Lambda function.
@@ -70,34 +73,13 @@ def main():
     psql_command = (
         f'psql "postgresql://{stack.db_username}:{stack.db_password}@{PublicDbEndpoint}:{PublicDbPort}/postgres"'
     )
+
+    print("Result of CloudFormation deployment:")
     print(psql_command)
 
+    print("RDS Instance ARN:", stack.get_output_value(stack._bsm, "RDSInstanceArn"))
     print("Stream ARN:", stack.get_output_value(stack._bsm, "StreamArn"))
-
-    """
-    aws dms describe-replications
-    aws dms start-replication \
-        --start-replication-type=start-replication \
-        --replication-config-arn arn:aws:dms:eu-central-1:931394475905:replication-config:LB2JAGY7XFB7PA7HEX3MI36CUA
-
-    aws logs describe-log-groups
-    aws logs start-live-tail --log-group-identifiers \
-        arn:aws:logs:eu-central-1:931394475905:log-group:/aws/rds/instance/testdrive-dms-postgresql-dev-db/postgresql \
-        arn:aws:logs:eu-central-1:931394475905:log-group:dms-serverless-replication-LB2JAGY7XFB7PA7HEX3MI36CUA
-
-    aws cloudformation continue-update-rollback --stack-name testdrive-dms-postgresql-dev
-    aws cloudformation delete-stack --stack-name testdrive-dms-postgresql-dev
-    """
-    """
-    - https://docs.aws.amazon.com/dms/latest/APIReference/API_StartReplication.html#DMS-StartReplication-request-StartReplicationType
-    - https://docs.aws.amazon.com/cli/latest/reference/dms/start-replication-task.html
-
-    Possible values:
-
-    - start-replication
-    - resume-processing
-    - reload-target
-    """
+    print("Replication ARN:", stack.get_output_value(stack._bsm, "ReplicationArn"))
 
 
 if __name__ == "__main__":

lorrystream/carabas/aws/stack/dms.py

@@ -1,3 +1,4 @@
+import json
 import typing as t
 
 import attr
@@ -199,6 +200,28 @@ def database(self):
         )
         group.add(self._db_security_group)
 
+        # aws rds describe-db-parameter-groups
+        # aws rds describe-db-parameters --db-parameter-group-name default.postgres15
+        db_parameter_group = rds.DBParameterGroup(
+            "RDSPostgreSQLParameterGroup",
+            rp_Family="postgres15",
+            rp_Description="DMS parameter group for postgres15",
+            p_DBParameterGroupName="dms-postgres15",
+            # aws rds describe-db-parameters --db-parameter-group-name default.postgres15
+            p_Parameters={
+                "log_connections": True,
+                # https://docs.aws.amazon.com/AmazonRDS/latest/UserGuide/Appendix.PostgreSQL.CommonDBATasks.pgaudit.html
+                "pgaudit.log": "all",
+                "pgaudit.log_statement_once": True,
+                # `rds.logical_replication is a cluster level setting, not db instance setting?
+                # https://stackoverflow.com/a/66252465
+                "rds.logical_replication": True,
+                # TODO: wal2json?
+                "shared_preload_libraries": "pgaudit,pglogical,pg_stat_statements",
+            },
+        )
+        group.add(db_parameter_group)
+
         db = rds.DBInstance(
             "RDSPostgreSQL",
             p_DBInstanceClass="db.t3.micro",
@@ -208,6 +231,7 @@ def database(self):
             # The current default engine version for AWS DMS is 3.5.2.
             # https://docs.aws.amazon.com/dms/latest/userguide/CHAP_ReleaseNotes.html
             p_EngineVersion="15",
+            p_DBParameterGroupName="dms-postgres15",
             # The parameter AllocatedStorage must be provided and must not be null.
             # Invalid storage size for engine name postgres and storage type gp2: 1
             p_AllocatedStorage="5",
@@ -235,11 +259,17 @@ def database(self):
                 Name=cf.Sub.from_params(f"{self.env_name}-db"),
                 Description=cf.Sub.from_params(f"The DB instance for {self.env_name}"),
             ),
-            ra_DependsOn=[self._db_security_group, self._db_subnet_group],
+            ra_DependsOn=[db_parameter_group, self._db_security_group, self._db_subnet_group],
         )
         self._db = db
         group.add(db)
 
+        rds_arn = cf.Output(
+            "RDSInstanceArn",
+            Value=db.rv_DBInstanceArn,
+        )
+        group.add(rds_arn)
+
         public_endpoint = cf.Output(
             "PublicDbEndpoint",
             Value=db.rv_EndpointAddress,
@@ -406,6 +436,9 @@ def dms(self):
         )
         group.add(vpc_endpoint_stream)
 
+        # https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Source.PostgreSQL.html#CHAP_Source.PostgreSQL.Advanced
+        # https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Source.PostgreSQL.html#CHAP_Source.PostgreSQL.RDSPostgreSQL
+        # https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Source.PostgreSQL.html#CHAP_Source.PostgreSQL.ConnectionAttrib
         source_endpoint = dms.Endpoint(  # type: ignore[call-arg,misc]
             "DMSSourceEndpoint",
             rp_EndpointType="source",
@@ -417,6 +450,12 @@ def dms(self):
             p_Username=self.db_username,
             p_Password=self.db_password,
             p_DatabaseName="postgres",
+            p_ExtraConnectionAttributes=json.dumps(
+                {
+                    "CaptureDdls": True,
+                    "PluginName": "pglogical",
+                }
+            ),
             p_EndpointIdentifier=f"{self.env_name}-endpoint-source",
             ra_DependsOn=[self._db],
         )
@@ -427,6 +466,12 @@ def dms(self):
             p_KinesisSettings=dms.PropEndpointKinesisSettings(
                 p_StreamArn=self._stream.rv_Arn,
                 p_MessageFormat="json-unformatted",
+                p_IncludeControlDetails=True,
+                p_IncludePartitionValue=True,
+                p_IncludeTransactionDetails=True,
+                p_IncludeNullAndEmpty=True,
+                p_IncludeTableAlterOperations=True,
+                p_PartitionIncludeSchemaTable=True,
                 # The parameter ServiceAccessRoleArn must be provided and must not be blank.
                 p_ServiceAccessRoleArn=dms_target_access_role.rv_Arn,
             ),
@@ -437,6 +482,7 @@ def dms(self):
         group.add(target_endpoint)
 
         # FIXME: Currently hard-coded to table `public.foo`.
+        # https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Target.Kinesis.html
         map_to_kinesis = {
             "rules": [
                 {
@@ -466,7 +512,7 @@ def dms(self):
             "DMSReplicationConfig",
             rp_ReplicationConfigIdentifier=f"{self.env_name}-dms-serverless",
             # p_ResourceIdentifier=f"{self.env_name}-dms-serverless-resource",  # noqa: ERA001
-            rp_ReplicationType="full-load",
+            rp_ReplicationType="full-load-and-cdc",
             rp_SourceEndpointArn=source_endpoint.ref(),
             rp_TargetEndpointArn=target_endpoint.ref(),
             rp_ComputeConfig=dms.PropReplicationConfigComputeConfig(
@@ -478,6 +524,12 @@ def dms(self):
             ),
             rp_TableMappings=map_to_kinesis,
             p_ReplicationSettings={
+                # https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Tasks.CustomizingTasks.TaskSettings.BeforeImage.html
+                "BeforeImageSettings": {
+                    "EnableBeforeImage": True,
+                    "FieldName": "before-image",
+                    "ColumnFilter": "pk-only",
+                },
                 # https://docs.aws.amazon.com/dms/latest/userguide/CHAP_Tasks.CustomizingTasks.TaskSettings.Logging.html
                 "Logging": {
                     "EnableLogging": True,
@@ -507,7 +559,7 @@ def dms(self):
                         # {"Id": "VALIDATOR", "Severity": "LOGGER_SEVERITY_DETAILED_DEBUG"},  # noqa: ERA001
                         {"Id": "VALIDATOR_EXT", "Severity": "LOGGER_SEVERITY_DETAILED_DEBUG"},
                     ],
-                }
+                },
             },
             ra_DependsOn=[
                 dms_replication_subnet_group,
@@ -521,6 +573,12 @@ def dms(self):
         )
         group.add(serverless_replication)
 
+        replication_arn = cf.Output(
+            "ReplicationArn",
+            Value=serverless_replication.rv_ReplicationConfigArn,
+        )
+        group.add(replication_arn)
+
         return self.add(group)
 
     @property