diff --git a/dsql-cdc-eventbridge-fanout-cdk/README.md b/dsql-cdc-eventbridge-fanout-cdk/README.md new file mode 100644 index 000000000..0c03a529e --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/README.md @@ -0,0 +1,148 @@ +# Amazon Aurora DSQL CDC to Amazon EventBridge Fan-out + +This pattern deploys an event-driven fan-out architecture that captures real-time database changes from Amazon Aurora DSQL using Change Data Capture (CDC), streams them through Amazon Kinesis Data Streams, processes them with AWS Lambda, and routes them to multiple consumers via Amazon EventBridge. + +Important: This pattern is fundamentally different from the existing `apigw-lambda-dsql` pattern (basic CRUD) and the standalone [sample-aurora-dsql-cdc-demo](https://github.com/aws-samples/sample-aurora-dsql-cdc-demo) (RAG/semantic search chatbot). This pattern demonstrates **event-driven microservices fan-out** — using Amazon Aurora DSQL CDC as a change event source to drive multiple independent downstream workflows through Amazon EventBridge content-based routing. + +Learn more about this pattern at Serverless Land Patterns: https://serverlessland.com/patterns/dsql-cdc-eventbridge-fanout-cdk + +## Architecture + +``` +┌──────────────┐ ┌─────────────────────┐ ┌──────────────────────┐ ┌─────────────────────┐ +│ Amazon Aurora │────▶│ Amazon Kinesis Data │────▶│ AWS Lambda │────▶│ Amazon EventBridge │ +│ DSQL (CDC) │ │ Streams │ │ (CDC Processor) │ │ (Custom Bus) │ +└──────────────┘ └─────────────────────┘ └──────────────────────┘ └──────────┬──────────┘ + │ + ┌────────────────────────────────────────────────────────────┼───────────────────┐ + │ │ │ + ▼ ▼ ▼ + ┌──────────────────┐ ┌─────────────────┐ ┌──────────────┐ + │ Amazon SQS │ │ AWS Step │ │ Amazon SNS │ + │ (Audit Queue) │ │ Functions │ │ (Alerts) │ + │ ALL events │ │ INSERT events │ │ DELETE events │ + └──────────────────┘ └─────────────────┘ └──────────────┘ +``` + +**How it works:** + +1. Amazon Aurora DSQL captures every committed row-level change (INSERT, UPDATE, DELETE) and delivers it as a structured JSON record to Amazon Kinesis Data Streams +2. AWS Lambda consumes the Amazon Kinesis stream, parses the CDC payload, classifies the operation type, and publishes typed events to an Amazon EventBridge custom event bus +3. Amazon EventBridge routes events to three independent consumers based on content: + - **ALL changes** → Amazon SQS audit queue (compliance/audit trail) + - **INSERT events** → AWS Step Functions workflow (data validation) + - **DELETE events** → Amazon SNS topic (real-time alerting) + +## Requirements + +- [AWS CDK v2](https://docs.aws.amazon.com/cdk/v2/guide/getting_started.html) installed and configured +- [Node.js 20+](https://nodejs.org/) with npm +- AWS account [bootstrapped for CDK](https://docs.aws.amazon.com/cdk/v2/guide/bootstrapping.html) +- An existing Amazon Aurora DSQL cluster (see Deployment section) +- Python 3.12 (for AWS Lambda functions) + +## Deployment + +### Step 1: Create an Amazon Aurora DSQL cluster (if you don't have one) + +```bash +aws dsql create-cluster --region us-east-1 +``` + +Note the `identifier` from the response — you'll need it in Step 3. + +### Step 2: Install dependencies and synthesize + +```bash +cd dsql-cdc-eventbridge-fanout-cdk/cdk +npm install +npx cdk synth +``` + +### Step 3: Deploy the stack + +```bash +npx cdk deploy --parameters DsqlClusterId= +``` + +Replace `` with your Amazon Aurora DSQL cluster identifier. + +### Step 4: Insert data to trigger CDC events + +Connect to your Amazon Aurora DSQL cluster and insert data: + +```sql +CREATE TABLE orders ( + id UUID DEFAULT gen_random_uuid() PRIMARY KEY, + customer_name TEXT NOT NULL, + amount DECIMAL(10,2), + created_at TIMESTAMP DEFAULT now() +); + +INSERT INTO orders (customer_name, amount) VALUES ('Acme Corp', 1250.00); +INSERT INTO orders (customer_name, amount) VALUES ('Globex Inc', 780.50); +DELETE FROM orders WHERE customer_name = 'Globex Inc'; +``` + +## Testing + +After deploying, verify the fan-out is working: + +### 1. Check Amazon SQS audit queue (receives ALL events) + +```bash +aws sqs receive-message \ + --queue-url $(aws cloudformation describe-stacks \ + --stack-name DsqlCdcEventbridgeFanoutStack \ + --query 'Stacks[0].Outputs[?OutputKey==`AuditQueueUrl`].OutputValue' \ + --output text) \ + --max-number-of-messages 5 +``` + +### 2. Check AWS Step Functions executions (INSERT events only) + +```bash +aws stepfunctions list-executions \ + --state-machine-arn $(aws cloudformation describe-stacks \ + --stack-name DsqlCdcEventbridgeFanoutStack \ + --query 'Stacks[0].Outputs[?OutputKey==`ValidationStateMachineArn`].OutputValue' \ + --output text) \ + --max-results 5 +``` + +### 3. Subscribe to Amazon SNS topic to receive DELETE alerts + +```bash +aws sns subscribe \ + --topic-arn $(aws cloudformation describe-stacks \ + --stack-name DsqlCdcEventbridgeFanoutStack \ + --query 'Stacks[0].Outputs[?OutputKey==`AlertTopicArn`].OutputValue' \ + --output text) \ + --protocol email \ + --notification-endpoint your-email@example.com +``` + +## Cleanup + +> **Warning:** This will delete all resources including the Amazon SQS queues and their messages. The Amazon Aurora DSQL cluster is NOT deleted (it was created externally). + +```bash +cd dsql-cdc-eventbridge-fanout-cdk/cdk +npx cdk destroy +``` + +## Services Used + +| Service | Role | +|---------|------| +| Amazon Aurora DSQL | Source database with CDC enabled | +| Amazon Kinesis Data Streams | Receives CDC event stream from Amazon Aurora DSQL | +| AWS Lambda | Processes CDC events, classifies operations, publishes to Amazon EventBridge | +| Amazon EventBridge | Content-based routing of CDC events to multiple consumers | +| Amazon SQS | Audit queue for all change events (compliance) | +| AWS Step Functions | Data validation workflow for INSERT events | +| Amazon SNS | Real-time alerting for DELETE events | + +---- +Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +SPDX-License-Identifier: MIT-0 diff --git a/dsql-cdc-eventbridge-fanout-cdk/cdk/.gitignore b/dsql-cdc-eventbridge-fanout-cdk/cdk/.gitignore new file mode 100644 index 000000000..1f0ea0352 --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/cdk/.gitignore @@ -0,0 +1,6 @@ +node_modules +cdk.out +cdk.context.json +build +*.js +*.d.ts diff --git a/dsql-cdc-eventbridge-fanout-cdk/cdk/bin/app.ts b/dsql-cdc-eventbridge-fanout-cdk/cdk/bin/app.ts new file mode 100644 index 000000000..ea4b329a0 --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/cdk/bin/app.ts @@ -0,0 +1,12 @@ +#!/usr/bin/env node +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT-0 (2026) + +import 'source-map-support/register'; +import * as cdk from 'aws-cdk-lib'; +import { DsqlCdcEventbridgeFanoutStack } from '../lib/dsql-cdc-eventbridge-fanout-stack'; + +const app = new cdk.App(); +new DsqlCdcEventbridgeFanoutStack(app, 'DsqlCdcEventbridgeFanoutStack', { + description: 'Amazon Aurora DSQL CDC to Amazon EventBridge fan-out pattern (uksb-1tupboc57)', +}); diff --git a/dsql-cdc-eventbridge-fanout-cdk/cdk/cdk.json b/dsql-cdc-eventbridge-fanout-cdk/cdk/cdk.json new file mode 100644 index 000000000..a6700a2ff --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/cdk/cdk.json @@ -0,0 +1,3 @@ +{ + "app": "npx ts-node --prefer-ts-exts bin/app.ts" +} diff --git a/dsql-cdc-eventbridge-fanout-cdk/cdk/lib/dsql-cdc-eventbridge-fanout-stack.ts b/dsql-cdc-eventbridge-fanout-cdk/cdk/lib/dsql-cdc-eventbridge-fanout-stack.ts new file mode 100644 index 000000000..e78cdb14b --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/cdk/lib/dsql-cdc-eventbridge-fanout-stack.ts @@ -0,0 +1,285 @@ +// Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +// SPDX-License-Identifier: MIT-0 (2026) + +import * as cdk from 'aws-cdk-lib'; +import * as kinesis from 'aws-cdk-lib/aws-kinesis'; +import * as lambda from 'aws-cdk-lib/aws-lambda'; +import * as iam from 'aws-cdk-lib/aws-iam'; +import * as events from 'aws-cdk-lib/aws-events'; +import * as targets from 'aws-cdk-lib/aws-events-targets'; +import * as sqs from 'aws-cdk-lib/aws-sqs'; +import * as sns from 'aws-cdk-lib/aws-sns'; +import * as sfn from 'aws-cdk-lib/aws-stepfunctions'; +import * as sfnTasks from 'aws-cdk-lib/aws-stepfunctions-tasks'; +import * as logs from 'aws-cdk-lib/aws-logs'; +import * as cr from 'aws-cdk-lib/custom-resources'; +import * as kinesisEvtSrc from 'aws-cdk-lib/aws-lambda-event-sources'; +import { Construct } from 'constructs'; +import * as path from 'path'; + +export class DsqlCdcEventbridgeFanoutStack extends cdk.Stack { + constructor(scope: Construct, id: string, props?: cdk.StackProps) { + super(scope, id, props); + + const region = cdk.Stack.of(this).region; + const account = cdk.Stack.of(this).account; + + // --- Context parameters --- + const dsqlClusterId = new cdk.CfnParameter(this, 'DsqlClusterId', { + type: 'String', + description: 'Amazon Aurora DSQL cluster identifier', + }); + + // ========================================================= + // 1. Amazon Kinesis Data Stream (CDC target) + // ========================================================= + const cdcStream = new kinesis.Stream(this, 'CdcStream', { + streamName: `dsql-cdc-${cdk.Names.uniqueId(this).slice(-8).toLowerCase()}`, + shardCount: 1, + retentionPeriod: cdk.Duration.hours(24), + encryption: kinesis.StreamEncryption.MANAGED, + }); + + // ========================================================= + // 2. IAM Role for DSQL to write to Kinesis + // ========================================================= + const dsqlCdcRole = new iam.Role(this, 'DsqlCdcRole', { + assumedBy: new iam.ServicePrincipal('dsql.amazonaws.com'), + description: 'Allows Amazon Aurora DSQL CDC to write to Amazon Kinesis', + inlinePolicies: { + KinesisWrite: new iam.PolicyDocument({ + statements: [ + new iam.PolicyStatement({ + effect: iam.Effect.ALLOW, + actions: [ + 'kinesis:PutRecord', + 'kinesis:PutRecords', + 'kinesis:DescribeStream', + ], + resources: [cdcStream.streamArn], + }), + ], + }), + }, + }); + + // ========================================================= + // 3. Custom Resource: Create/Delete DSQL CDC Stream + // (No CFN resource type for DSQL streams yet — use SDK) + // ========================================================= + const streamManagerFn = new lambda.Function(this, 'CdcStreamManagerFn', { + runtime: lambda.Runtime.PYTHON_3_12, + handler: 'handler.on_event', + code: lambda.Code.fromAsset(path.join(__dirname, '../../lambdas/cdc-stream-manager'), { + bundling: { + image: lambda.Runtime.PYTHON_3_12.bundlingImage, + command: [ + 'bash', '-c', + 'pip install -r requirements.txt -t /asset-output && cp handler.py /asset-output/', + ], + }, + }), + timeout: cdk.Duration.minutes(5), + memorySize: 256, + description: 'Custom Resource: manages Amazon Aurora DSQL CDC stream lifecycle', + environment: { + CLUSTER_ID: dsqlClusterId.valueAsString, + KINESIS_STREAM_ARN: cdcStream.streamArn, + CDC_ROLE_ARN: dsqlCdcRole.roleArn, + }, + }); + + streamManagerFn.addToRolePolicy(new iam.PolicyStatement({ + effect: iam.Effect.ALLOW, + actions: [ + 'dsql:CreateStream', + 'dsql:DeleteStream', + 'dsql:GetStream', + 'dsql:ListStreams', + ], + resources: [ + `arn:aws:dsql:${region}:${account}:cluster/${dsqlClusterId.valueAsString}`, + `arn:aws:dsql:${region}:${account}:cluster/${dsqlClusterId.valueAsString}/stream/*`, + ], + })); + + streamManagerFn.addToRolePolicy(new iam.PolicyStatement({ + effect: iam.Effect.ALLOW, + actions: ['iam:PassRole'], + resources: [dsqlCdcRole.roleArn], + conditions: { + StringEquals: { 'iam:PassedToService': 'dsql.amazonaws.com' }, + }, + })); + + const cdcStreamCr = new cr.Provider(this, 'CdcStreamProvider', { + onEventHandler: streamManagerFn, + }); + + const cdcStreamResource = new cdk.CustomResource(this, 'DsqlCdcStream', { + serviceToken: cdcStreamCr.serviceToken, + properties: { + ClusterId: dsqlClusterId.valueAsString, + KinesisStreamArn: cdcStream.streamArn, + RoleArn: dsqlCdcRole.roleArn, + }, + }); + + // ========================================================= + // 4. Amazon EventBridge Custom Event Bus + // ========================================================= + const cdcEventBus = new events.EventBus(this, 'CdcEventBus', { + eventBusName: 'dsql-cdc-events', + }); + + // ========================================================= + // 5. AWS Lambda: CDC Processor (Kinesis → EventBridge) + // ========================================================= + const cdcProcessorFn = new lambda.Function(this, 'CdcProcessorFn', { + runtime: lambda.Runtime.PYTHON_3_12, + handler: 'handler.lambda_handler', + code: lambda.Code.fromAsset(path.join(__dirname, '../../lambdas/cdc-processor')), + timeout: cdk.Duration.seconds(60), + memorySize: 256, + description: 'Processes Amazon Aurora DSQL CDC events and fans out to Amazon EventBridge', + environment: { + EVENT_BUS_NAME: cdcEventBus.eventBusName, + }, + }); + + cdcProcessorFn.addToRolePolicy(new iam.PolicyStatement({ + effect: iam.Effect.ALLOW, + actions: ['events:PutEvents'], + resources: [cdcEventBus.eventBusArn], + })); + + cdcProcessorFn.addEventSource(new kinesisEvtSrc.KinesisEventSource(cdcStream, { + startingPosition: lambda.StartingPosition.TRIM_HORIZON, + batchSize: 100, + maxBatchingWindow: cdk.Duration.seconds(5), + retryAttempts: 3, + bisectBatchOnError: true, + })); + + // ========================================================= + // 6. Fan-out Target A: Amazon SQS (Audit Queue) + // ========================================================= + const auditDlq = new sqs.Queue(this, 'AuditDlq', { + queueName: 'dsql-cdc-audit-dlq', + retentionPeriod: cdk.Duration.days(14), + encryption: sqs.QueueEncryption.SQS_MANAGED, + }); + + const auditQueue = new sqs.Queue(this, 'AuditQueue', { + queueName: 'dsql-cdc-audit', + visibilityTimeout: cdk.Duration.seconds(30), + encryption: sqs.QueueEncryption.SQS_MANAGED, + deadLetterQueue: { queue: auditDlq, maxReceiveCount: 3 }, + }); + + new events.Rule(this, 'AuditRule', { + eventBus: cdcEventBus, + ruleName: 'route-all-changes-to-audit', + description: 'Route ALL CDC events to Amazon SQS audit queue', + eventPattern: { + source: ['dsql.cdc'], + }, + targets: [new targets.SqsQueue(auditQueue)], + }); + + // ========================================================= + // 7. Fan-out Target B: AWS Step Functions (Data Validation) + // ========================================================= + const validatePass = new sfn.Pass(this, 'ValidationPassed', { + result: sfn.Result.fromObject({ validated: true }), + }); + + const validateFail = new sfn.Pass(this, 'ValidationFailed', { + result: sfn.Result.fromObject({ validated: false }), + }); + + const validationChoice = new sfn.Choice(this, 'HasRequiredFields') + .when( + sfn.Condition.isPresent('$.detail.newImage'), + validatePass, + ) + .otherwise(validateFail); + + const validationStateMachine = new sfn.StateMachine(this, 'ValidationWorkflow', { + definitionBody: sfn.DefinitionBody.fromChainable(validationChoice), + timeout: cdk.Duration.minutes(5), + tracingEnabled: true, + logs: { + destination: new logs.LogGroup(this, 'ValidationLogGroup', { + logGroupName: '/aws/stepfunctions/dsql-cdc-validation', + retention: logs.RetentionDays.ONE_WEEK, + removalPolicy: cdk.RemovalPolicy.DESTROY, + }), + level: sfn.LogLevel.ERROR, + }, + }); + + new events.Rule(this, 'ValidationRule', { + eventBus: cdcEventBus, + ruleName: 'route-inserts-to-validation', + description: 'Route INSERT events to AWS Step Functions for validation', + eventPattern: { + source: ['dsql.cdc'], + detailType: ['INSERT'], + }, + targets: [new targets.SfnStateMachine(validationStateMachine)], + }); + + // ========================================================= + // 8. Fan-out Target C: Amazon SNS (Real-time Alerts) + // ========================================================= + const alertTopic = new sns.Topic(this, 'AlertTopic', { + topicName: 'dsql-cdc-delete-alerts', + displayName: 'Amazon Aurora DSQL CDC - DELETE Alerts', + }); + + new events.Rule(this, 'DeleteAlertRule', { + eventBus: cdcEventBus, + ruleName: 'route-deletes-to-alert', + description: 'Route DELETE events to Amazon SNS for alerting', + eventPattern: { + source: ['dsql.cdc'], + detailType: ['DELETE'], + }, + targets: [new targets.SnsTopic(alertTopic)], + }); + + // ========================================================= + // Outputs + // ========================================================= + new cdk.CfnOutput(this, 'KinesisStreamArn', { + value: cdcStream.streamArn, + description: 'Amazon Kinesis Data Stream ARN (CDC target)', + }); + + new cdk.CfnOutput(this, 'EventBusArn', { + value: cdcEventBus.eventBusArn, + description: 'Amazon EventBridge custom event bus ARN', + }); + + new cdk.CfnOutput(this, 'AuditQueueUrl', { + value: auditQueue.queueUrl, + description: 'Amazon SQS audit queue URL', + }); + + new cdk.CfnOutput(this, 'ValidationStateMachineArn', { + value: validationStateMachine.stateMachineArn, + description: 'AWS Step Functions validation workflow ARN', + }); + + new cdk.CfnOutput(this, 'AlertTopicArn', { + value: alertTopic.topicArn, + description: 'Amazon SNS alert topic ARN', + }); + + new cdk.CfnOutput(this, 'CdcStreamId', { + value: cdcStreamResource.getAttString('StreamId'), + description: 'Amazon Aurora DSQL CDC stream identifier', + }); + } +} diff --git a/dsql-cdc-eventbridge-fanout-cdk/cdk/package.json b/dsql-cdc-eventbridge-fanout-cdk/cdk/package.json new file mode 100644 index 000000000..6af041f45 --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/cdk/package.json @@ -0,0 +1,20 @@ +{ + "name": "dsql-cdc-eventbridge-fanout-cdk", + "version": "1.0.0", + "bin": { "app": "bin/app.ts" }, + "scripts": { + "build": "tsc", + "synth": "cdk synth", + "deploy": "cdk deploy", + "destroy": "cdk destroy" + }, + "dependencies": { + "aws-cdk-lib": "2.185.0", + "constructs": "^10.0.0", + "source-map-support": "^0.5.21" + }, + "devDependencies": { + "typescript": "~5.4.0", + "ts-node": "^10.9.0" + } +} diff --git a/dsql-cdc-eventbridge-fanout-cdk/cdk/tsconfig.json b/dsql-cdc-eventbridge-fanout-cdk/cdk/tsconfig.json new file mode 100644 index 000000000..7ddcfe705 --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/cdk/tsconfig.json @@ -0,0 +1,24 @@ +{ + "compilerOptions": { + "target": "ES2020", + "module": "commonjs", + "lib": ["es2020"], + "declaration": true, + "strict": true, + "noImplicitAny": true, + "strictNullChecks": true, + "noImplicitThis": true, + "alwaysStrict": true, + "noUnusedLocals": false, + "noUnusedParameters": false, + "noImplicitReturns": true, + "noFallthroughCasesInSwitch": false, + "inlineSourceMap": true, + "inlineSources": true, + "experimentalDecorators": true, + "strictPropertyInitialization": false, + "outDir": "./build", + "rootDir": "." + }, + "exclude": ["node_modules", "build"] +} diff --git a/dsql-cdc-eventbridge-fanout-cdk/example-pattern.json b/dsql-cdc-eventbridge-fanout-cdk/example-pattern.json new file mode 100644 index 000000000..4e7fbc86c --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/example-pattern.json @@ -0,0 +1,113 @@ +{ + "title": "Amazon Aurora DSQL CDC to Amazon EventBridge Fan-out", + "description": "Stream real-time database changes from Amazon Aurora DSQL via CDC to Amazon EventBridge for content-based routing to multiple consumers", + "language": "TypeScript", + "level": "300", + "framework": "AWS CDK", + "introBox": { + "headline": "How it works", + "text": [ + "Amazon Aurora DSQL captures committed row-level changes (INSERT, UPDATE, DELETE) and streams them as JSON records to Amazon Kinesis Data Streams.", + "AWS Lambda consumes the stream, classifies each CDC event by operation type, and publishes typed events to an Amazon EventBridge custom event bus.", + "Amazon EventBridge routes events to three independent consumers: Amazon SQS (audit trail for ALL events), AWS Step Functions (validation workflow for INSERTs), and Amazon SNS (alerting for DELETEs)." + ] + }, + "gitHub": { + "template": { + "repoURL": "https://github.com/aws-samples/serverless-patterns/tree/main/dsql-cdc-eventbridge-fanout-cdk", + "templateURL": "serverless-patterns/dsql-cdc-eventbridge-fanout-cdk", + "projectFolder": "dsql-cdc-eventbridge-fanout-cdk", + "templateFile": "cdk/lib/dsql-cdc-eventbridge-fanout-stack.ts" + } + }, + "resources": { + "bullets": [ + { + "text": "Amazon Aurora DSQL CDC documentation", + "link": "https://docs.aws.amazon.com/aurora-dsql/latest/userguide/cdc-streams.html" + }, + { + "text": "Amazon EventBridge content-based filtering", + "link": "https://docs.aws.amazon.com/eventbridge/latest/userguide/eb-event-patterns.html" + }, + { + "text": "AWS CDK Custom Resources", + "link": "https://docs.aws.amazon.com/cdk/api/v2/docs/aws-cdk-lib.custom_resources-readme.html" + } + ] + }, + "deploy": { + "text": [ + "cd dsql-cdc-eventbridge-fanout-cdk/cdk", + "npm install", + "npx cdk deploy --parameters DsqlClusterId=your-cluster-id" + ] + }, + "testing": { + "text": [ + "Insert data into your Amazon Aurora DSQL cluster to generate CDC events.", + "Check the Amazon SQS audit queue for all change events.", + "Check AWS Step Functions executions for INSERT validation workflows.", + "Subscribe to the Amazon SNS topic to receive DELETE alerts." + ] + }, + "cleanup": { + "text": [ + "npx cdk destroy" + ] + }, + "authors": [ + { + "name": "Nithin Chandran R", + "image": "", + "bio": "Technical Account Manager at AWS", + "linkedin": "nithin-chandran-r", + "twitter": "" + } + ], + "patternArch": { + "icon1": { + "x": 10, + "y": 50, + "service": "dsql", + "label": "Amazon Aurora DSQL" + }, + "icon2": { + "x": 30, + "y": 50, + "service": "kinesis", + "label": "Amazon Kinesis" + }, + "icon3": { + "x": 50, + "y": 50, + "service": "lambda", + "label": "AWS Lambda" + }, + "icon4": { + "x": 70, + "y": 50, + "service": "eventbridge", + "label": "Amazon EventBridge" + }, + "icon5": { + "x": 90, + "y": 25, + "service": "sqs", + "label": "Amazon SQS" + }, + "icon6": { + "x": 90, + "y": 50, + "service": "sfn", + "label": "AWS Step Functions" + }, + "icon7": { + "x": 90, + "y": 75, + "service": "sns", + "label": "Amazon SNS" + } + }, + "patternType": "Serverless" +} diff --git a/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-processor/handler.py b/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-processor/handler.py new file mode 100644 index 000000000..8d93c6959 --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-processor/handler.py @@ -0,0 +1,83 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: MIT-0 (2026) +"""CDC Processor: Parse Amazon Aurora DSQL CDC events from Amazon Kinesis +and fan out to Amazon EventBridge with typed detail-type routing.""" + +import base64 +import json +import os +from datetime import datetime, timezone + +import boto3 + +EVENTS_CLIENT = boto3.client('events') +EVENT_BUS_NAME = os.environ['EVENT_BUS_NAME'] +SOURCE = 'dsql.cdc' + + +def lambda_handler(event, context): + """Process Amazon Kinesis records containing Amazon Aurora DSQL CDC events.""" + records = event.get('Records', []) + if not records: + return {'statusCode': 200, 'processed': 0} + + # DSQL CDC JSON format: + # {"type":"full","op":"c|u|d","before":null|{...},"after":{...}|null,"source":{...}} + OP_MAP = {'c': 'INSERT', 'u': 'UPDATE', 'd': 'DELETE'} + + entries = [] + for record in records: + try: + payload = base64.b64decode(record['kinesis']['data']) + cdc_event = json.loads(payload) + + # Map DSQL CDC op codes to human-readable operation types + raw_op = cdc_event.get('op', '').lower() + operation = OP_MAP.get(raw_op, 'UNKNOWN') + table_name = cdc_event.get('source', {}).get('table', 'unknown') + + # Use operation as EventBridge detail-type for content-based routing + detail_type = operation + + entry = { + 'Source': SOURCE, + 'DetailType': detail_type, + 'Detail': json.dumps({ + 'tableName': table_name, + 'operation': operation, + 'newImage': cdc_event.get('after'), + 'oldImage': cdc_event.get('before'), + 'timestamp': cdc_event.get('source', {}).get('ts_ms', + datetime.now(timezone.utc).isoformat()), + 'sequenceNumber': record['kinesis'].get('sequenceNumber'), + }), + 'EventBusName': EVENT_BUS_NAME, + } + entries.append(entry) + + except (json.JSONDecodeError, KeyError) as e: + print(f'Failed to parse CDC record: {e}') + continue + + # Publish in batches of 10 (EventBridge PutEvents limit) + failed_count = 0 + for i in range(0, len(entries), 10): + batch = entries[i:i + 10] + try: + response = EVENTS_CLIENT.put_events(Entries=batch) + failed_count += response.get('FailedEntryCount', 0) + except Exception as e: + print(f'EventBridge PutEvents error: {e}') + failed_count += len(batch) + + print(f'Processed {len(records)} records, published {len(entries)} events, ' + f'{failed_count} failures') + + if failed_count > 0: + raise Exception(f'{failed_count} events failed to publish to Amazon EventBridge') + + return { + 'statusCode': 200, + 'processed': len(records), + 'published': len(entries), + } diff --git a/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/.gitignore b/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/.gitignore new file mode 100644 index 000000000..61f07b025 --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/.gitignore @@ -0,0 +1,10 @@ +# Vendored dependencies (bundled at deploy time, not committed) +boto3/ +botocore/ +s3transfer/ +urllib3/ +jmespath/ +dateutil/ +six.py +*.dist-info/ +bin/ diff --git a/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/handler.py b/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/handler.py new file mode 100644 index 000000000..4e0d0e9b1 --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/handler.py @@ -0,0 +1,108 @@ +# Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. +# SPDX-License-Identifier: MIT-0 (2026) +"""Custom Resource: Create and delete Amazon Aurora DSQL CDC streams.""" + +import json +import os +import time + +import boto3 + +DSQL_CLIENT = boto3.client('dsql') + + +def on_event(event, context): + """Handle CloudFormation Custom Resource lifecycle events.""" + request_type = event['RequestType'] + properties = event['ResourceProperties'] + + cluster_id = properties.get('ClusterId', os.environ.get('CLUSTER_ID', '')) + kinesis_arn = properties.get('KinesisStreamArn', os.environ.get('KINESIS_STREAM_ARN', '')) + role_arn = properties.get('RoleArn', os.environ.get('CDC_ROLE_ARN', '')) + + if request_type == 'Create': + return create_stream(cluster_id, kinesis_arn, role_arn) + elif request_type == 'Update': + # CDC streams are immutable — delete old, create new + old_stream_id = event.get('PhysicalResourceId', '') + if old_stream_id: + delete_stream(cluster_id, old_stream_id) + return create_stream(cluster_id, kinesis_arn, role_arn) + elif request_type == 'Delete': + stream_id = event.get('PhysicalResourceId', '') + if stream_id and stream_id != 'NONE': + delete_stream(cluster_id, stream_id) + return {'PhysicalResourceId': stream_id or 'NONE'} + + +def create_stream(cluster_id: str, kinesis_arn: str, role_arn: str) -> dict: + """Create a DSQL CDC stream targeting Amazon Kinesis.""" + try: + response = DSQL_CLIENT.create_stream( + clusterIdentifier=cluster_id, + targetDefinition={ + 'kinesis': { + 'streamArn': kinesis_arn, + 'roleArn': role_arn, + } + }, + ordering='UNORDERED', + format='JSON', + ) + + stream_id = response['streamIdentifier'] + print(f'Created CDC stream: {stream_id} (status: {response["status"]})') + + # Wait for stream to become ACTIVE (max 60s) + wait_for_active(cluster_id, stream_id) + + return { + 'PhysicalResourceId': stream_id, + 'Data': { + 'StreamId': stream_id, + 'StreamArn': response.get('arn', ''), + 'Status': 'ACTIVE', + }, + } + + except Exception as e: + print(f'Failed to create CDC stream: {e}') + raise + + +def delete_stream(cluster_id: str, stream_id: str) -> None: + """Delete a DSQL CDC stream.""" + try: + DSQL_CLIENT.delete_stream( + clusterIdentifier=cluster_id, + streamIdentifier=stream_id, + ) + print(f'Deleted CDC stream: {stream_id}') + except DSQL_CLIENT.exceptions.ResourceNotFoundException: + print(f'Stream {stream_id} already deleted') + except Exception as e: + print(f'Failed to delete CDC stream {stream_id}: {e}') + raise + + +def wait_for_active(cluster_id: str, stream_id: str, max_wait: int = 60) -> None: + """Poll until stream status is ACTIVE.""" + waited = 0 + while waited < max_wait: + try: + response = DSQL_CLIENT.get_stream( + clusterIdentifier=cluster_id, + streamIdentifier=stream_id, + ) + status = response.get('status', '') + if status == 'ACTIVE': + return + if status == 'FAILED': + raise Exception(f'CDC stream {stream_id} entered FAILED state') + except Exception as e: + if 'ResourceNotFound' not in str(e): + raise + time.sleep(5) + waited += 5 + + print(f'Warning: stream {stream_id} not ACTIVE after {max_wait}s, proceeding') diff --git a/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/requirements.txt b/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/requirements.txt new file mode 100644 index 000000000..c26e4b579 --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/requirements.txt @@ -0,0 +1 @@ +boto3>=1.43.36