From 7ba814c2efba739f32491c430671e68bcd234d47 Mon Sep 17 00:00:00 2001 From: Nithin Chandran Rajashankar Date: Mon, 13 Jul 2026 06:06:45 +0000 Subject: [PATCH 1/3] feat(dsql-cdc-eventbridge-fanout-cdk): DSQL CDC to EventBridge fan-out Event-driven microservices pattern using Amazon Aurora DSQL CDC (GA July 2026) as a change event source. CDC events flow through Amazon Kinesis to AWS Lambda, which classifies operations and publishes typed events to Amazon EventBridge. Three independent consumers demonstrate content-based routing: Amazon SQS (audit), AWS Step Functions (validation), Amazon SNS (alerting). 7 AWS services, 29 CFN resources. Fundamentally different from existing apigw-lambda-dsql (basic CRUD) and sample-aurora-dsql-cdc-demo (RAG/semantic search). This pattern shows DSQL as an event source for microservices fan-out via EventBridge. --- dsql-cdc-eventbridge-fanout-cdk/README.md | 148 ++++++++++ .../cdk/.gitignore | 6 + .../cdk/bin/app.ts | 12 + dsql-cdc-eventbridge-fanout-cdk/cdk/cdk.json | 3 + .../lib/dsql-cdc-eventbridge-fanout-stack.ts | 278 ++++++++++++++++++ .../cdk/package.json | 20 ++ .../cdk/tsconfig.json | 24 ++ .../example-pattern.json | 113 +++++++ .../lambdas/cdc-processor/handler.py | 78 +++++ .../lambdas/cdc-stream-manager/handler.py | 108 +++++++ 10 files changed, 790 insertions(+) create mode 100644 dsql-cdc-eventbridge-fanout-cdk/README.md create mode 100644 dsql-cdc-eventbridge-fanout-cdk/cdk/.gitignore create mode 100644 dsql-cdc-eventbridge-fanout-cdk/cdk/bin/app.ts create mode 100644 dsql-cdc-eventbridge-fanout-cdk/cdk/cdk.json create mode 100644 dsql-cdc-eventbridge-fanout-cdk/cdk/lib/dsql-cdc-eventbridge-fanout-stack.ts create mode 100644 dsql-cdc-eventbridge-fanout-cdk/cdk/package.json create mode 100644 dsql-cdc-eventbridge-fanout-cdk/cdk/tsconfig.json create mode 100644 dsql-cdc-eventbridge-fanout-cdk/example-pattern.json create mode 100644 dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-processor/handler.py create mode 100644 dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/handler.py diff --git a/dsql-cdc-eventbridge-fanout-cdk/README.md b/dsql-cdc-eventbridge-fanout-cdk/README.md new file mode 100644 index 0000000000..0c03a529ed --- /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 0000000000..1f0ea03523 --- /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 0000000000..ea4b329a03 --- /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 0000000000..a6700a2ff4 --- /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 0000000000..d12cc6ee30 --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/cdk/lib/dsql-cdc-eventbridge-fanout-stack.ts @@ -0,0 +1,278 @@ +// 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 snsSubs from 'aws-cdk-lib/aws-sns-subscriptions'; +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')), + 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 0000000000..6af041f45b --- /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 0000000000..7ddcfe705e --- /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 0000000000..4e7fbc86c0 --- /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 0000000000..43b6e878f5 --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-processor/handler.py @@ -0,0 +1,78 @@ +# 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} + + entries = [] + for record in records: + try: + payload = base64.b64decode(record['kinesis']['data']) + cdc_event = json.loads(payload) + + # DSQL CDC JSON format contains: operation, tableName, newImage, oldImage + operation = cdc_event.get('operation', 'UNKNOWN').upper() + table_name = cdc_event.get('tableName', 'unknown') + + # Map to EventBridge detail-type for routing + detail_type = operation # INSERT, UPDATE, DELETE + + entry = { + 'Source': SOURCE, + 'DetailType': detail_type, + 'Detail': json.dumps({ + 'tableName': table_name, + 'operation': operation, + 'newImage': cdc_event.get('newImage'), + 'oldImage': cdc_event.get('oldImage'), + 'timestamp': cdc_event.get('commitTimestamp', + 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/handler.py b/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/handler.py new file mode 100644 index 0000000000..4e0d0e9b11 --- /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') From 603977a8dc3f74507cb52b61e58d2caf31088429 Mon Sep 17 00:00:00 2001 From: Nithin Chandran Rajashankar Date: Mon, 13 Jul 2026 06:31:52 +0000 Subject: [PATCH 2/3] fix(dsql-cdc-eventbridge-fanout-cdk): Map DSQL CDC op codes and bundle boto3 DSQL CDC uses Debezium-style format: op='c'/'u'/'d' with before/after fields. Map op codes to human-readable types for EventBridge routing. Bundle boto3>=1.43.36 via CDK Docker bundling (Lambda runtime ships older version without dsql.create_stream support). Deployed and tested end-to-end: - 3 records processed, 3 events published, 0 failures - Amazon SQS audit: receives all 3 event types (INSERT, UPDATE, DELETE) - AWS Step Functions: 1 INSERT validation SUCCEEDED - Amazon SNS: DELETE event routed to alerting topic --- .../lib/dsql-cdc-eventbridge-fanout-stack.ts | 11 ++++++++-- .../lambdas/cdc-processor/handler.py | 21 ++++++++++++------- .../lambdas/cdc-stream-manager/.gitignore | 10 +++++++++ .../cdc-stream-manager/requirements.txt | 1 + 4 files changed, 33 insertions(+), 10 deletions(-) create mode 100644 dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/.gitignore create mode 100644 dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/requirements.txt 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 index d12cc6ee30..e78cdb14ba 100644 --- 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 @@ -9,7 +9,6 @@ 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 snsSubs from 'aws-cdk-lib/aws-sns-subscriptions'; 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'; @@ -71,7 +70,15 @@ export class DsqlCdcEventbridgeFanoutStack extends cdk.Stack { 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')), + 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', diff --git a/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-processor/handler.py b/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-processor/handler.py index 43b6e878f5..8d93c69596 100644 --- a/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-processor/handler.py +++ b/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-processor/handler.py @@ -21,18 +21,23 @@ def lambda_handler(event, context): 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) - # DSQL CDC JSON format contains: operation, tableName, newImage, oldImage - operation = cdc_event.get('operation', 'UNKNOWN').upper() - table_name = cdc_event.get('tableName', 'unknown') + # 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') - # Map to EventBridge detail-type for routing - detail_type = operation # INSERT, UPDATE, DELETE + # Use operation as EventBridge detail-type for content-based routing + detail_type = operation entry = { 'Source': SOURCE, @@ -40,9 +45,9 @@ def lambda_handler(event, context): 'Detail': json.dumps({ 'tableName': table_name, 'operation': operation, - 'newImage': cdc_event.get('newImage'), - 'oldImage': cdc_event.get('oldImage'), - 'timestamp': cdc_event.get('commitTimestamp', + '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'), }), 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 0000000000..61f07b0254 --- /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/requirements.txt b/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/requirements.txt new file mode 100644 index 0000000000..c26e4b5794 --- /dev/null +++ b/dsql-cdc-eventbridge-fanout-cdk/lambdas/cdc-stream-manager/requirements.txt @@ -0,0 +1 @@ +boto3>=1.43.36 From cfcc35d1f46e26a60d672a7975c61722b4dce357 Mon Sep 17 00:00:00 2001 From: Nithin Chandran R Date: Tue, 11 Aug 2026 17:50:37 +0000 Subject: [PATCH 3/3] fix(dsql-cdc-eventbridge-fanout-cdk): Simplify to 4 core services Remove fan-out targets (SQS, SFN, SNS) per reviewer feedback. Pattern now focuses on the core DSQL CDC pipeline: Amazon Aurora DSQL -> Amazon Kinesis -> AWS Lambda -> Amazon EventBridge. Users add their own rules and targets to the custom event bus for downstream routing. --- dsql-cdc-eventbridge-fanout-cdk/README.md | 125 ++++++------------ .../lib/dsql-cdc-eventbridge-fanout-stack.ts | 118 +---------------- .../example-pattern.json | 54 +++----- 3 files changed, 70 insertions(+), 227 deletions(-) diff --git a/dsql-cdc-eventbridge-fanout-cdk/README.md b/dsql-cdc-eventbridge-fanout-cdk/README.md index 0c03a529ed..e1af47107c 100644 --- a/dsql-cdc-eventbridge-fanout-cdk/README.md +++ b/dsql-cdc-eventbridge-fanout-cdk/README.md @@ -1,75 +1,61 @@ -# Amazon Aurora DSQL CDC to Amazon EventBridge Fan-out +# Amazon Aurora DSQL CDC to Amazon EventBridge -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. +This pattern deploys an event-driven pipeline 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 publishes typed events to an Amazon EventBridge custom event bus for downstream consumption. Learn more about this pattern at Serverless Land Patterns: https://serverlessland.com/patterns/dsql-cdc-eventbridge-fanout-cdk +Important: this application uses various AWS services and there are costs associated with these services after the Free Tier usage - please see the [AWS Pricing page](https://aws.amazon.com/pricing/) for details. + ## 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 │ - └──────────────────┘ └─────────────────┘ └──────────────┘ +┌──────────────────┐ ┌─────────────────────┐ ┌──────────────────────┐ ┌─────────────────────┐ +│ Amazon Aurora │────▶│ Amazon Kinesis Data │────▶│ AWS Lambda │────▶│ Amazon EventBridge │ +│ DSQL (CDC) │ │ Streams │ │ (CDC Processor) │ │ (Custom Bus) │ +└──────────────────┘ └─────────────────────┘ └──────────────────────┘ └─────────────────────┘ ``` **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) +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 (Debezium-style op codes), classifies the operation type, and publishes typed events to an Amazon EventBridge custom event bus. +3. Amazon EventBridge receives events with source `dsql.cdc` and detail-type `INSERT`, `UPDATE`, or `DELETE`. Add your own rules and targets to route events to any downstream consumer. ## 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) +- An existing Amazon Aurora DSQL cluster - Python 3.12 (for AWS Lambda functions) ## Deployment -### Step 1: Create an Amazon Aurora DSQL cluster (if you don't have one) +1. Create an Amazon Aurora DSQL cluster (if you don't have one): -```bash -aws dsql create-cluster --region us-east-1 -``` + ```bash + aws dsql create-cluster --region us-east-1 + ``` -Note the `identifier` from the response — you'll need it in Step 3. + Note the `identifier` from the response. -### Step 2: Install dependencies and synthesize +2. Install dependencies and build: -```bash -cd dsql-cdc-eventbridge-fanout-cdk/cdk -npm install -npx cdk synth -``` - -### Step 3: Deploy the stack + ```bash + cd dsql-cdc-eventbridge-fanout-cdk/cdk + npm install + npm run build + ``` -```bash -npx cdk deploy --parameters DsqlClusterId= -``` +3. Deploy the stack: -Replace `` with your Amazon Aurora DSQL cluster identifier. + ```bash + npx cdk deploy --parameters DsqlClusterId= + ``` -### Step 4: Insert data to trigger CDC events +## Testing -Connect to your Amazon Aurora DSQL cluster and insert data: +After deploying, insert data into your Amazon Aurora DSQL cluster to trigger CDC events: ```sql CREATE TABLE orders ( @@ -80,54 +66,32 @@ CREATE TABLE orders ( ); 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'; +UPDATE orders SET amount = 1500.00 WHERE customer_name = 'Acme Corp'; +DELETE FROM orders WHERE customer_name = 'Acme Corp'; ``` -## Testing - -After deploying, verify the fan-out is working: - -### 1. Check Amazon SQS audit queue (receives ALL events) +Then verify events arrive on the custom event bus by adding a temporary rule: ```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 +aws events put-rule \ + --name test-cdc-rule \ + --event-bus-name dsql-cdc-events \ + --event-pattern '{"source": ["dsql.cdc"]}' ``` -### 2. Check AWS Step Functions executions (INSERT events only) +## Extending This Pattern -```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 +Add Amazon EventBridge rules and targets to route CDC events to any consumer: -```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 -``` +- Route ALL changes to Amazon SQS for audit +- Route INSERT events to AWS Step Functions for validation +- Route DELETE events to Amazon SNS for alerting ## 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). +> **Warning:** This will delete all resources. The Amazon Aurora DSQL cluster is NOT deleted (it was created externally). ```bash -cd dsql-cdc-eventbridge-fanout-cdk/cdk npx cdk destroy ``` @@ -138,10 +102,7 @@ npx cdk destroy | 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 | +| Amazon EventBridge | Custom event bus for content-based routing of CDC events | ---- Copyright Amazon.com, Inc. or its affiliates. All Rights Reserved. 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 index e78cdb14ba..824d290024 100644 --- 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 @@ -6,12 +6,6 @@ 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'; @@ -41,7 +35,7 @@ export class DsqlCdcEventbridgeFanoutStack extends cdk.Stack { }); // ========================================================= - // 2. IAM Role for DSQL to write to Kinesis + // 2. IAM Role for Amazon Aurora DSQL to write to Amazon Kinesis // ========================================================= const dsqlCdcRole = new iam.Role(this, 'DsqlCdcRole', { assumedBy: new iam.ServicePrincipal('dsql.amazonaws.com'), @@ -133,7 +127,7 @@ export class DsqlCdcEventbridgeFanoutStack extends cdk.Stack { }); // ========================================================= - // 5. AWS Lambda: CDC Processor (Kinesis → EventBridge) + // 5. AWS Lambda: CDC Processor (Amazon Kinesis → Amazon EventBridge) // ========================================================= const cdcProcessorFn = new lambda.Function(this, 'CdcProcessorFn', { runtime: lambda.Runtime.PYTHON_3_12, @@ -141,7 +135,7 @@ export class DsqlCdcEventbridgeFanoutStack extends cdk.Stack { 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', + description: 'Processes Amazon Aurora DSQL CDC events and publishes to Amazon EventBridge', environment: { EVENT_BUS_NAME: cdcEventBus.eventBusName, }, @@ -161,94 +155,6 @@ export class DsqlCdcEventbridgeFanoutStack extends cdk.Stack { 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 // ========================================================= @@ -259,22 +165,12 @@ export class DsqlCdcEventbridgeFanoutStack extends cdk.Stack { 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', + description: 'Amazon EventBridge custom event bus ARN — add rules and targets to consume CDC events', }); - new cdk.CfnOutput(this, 'AlertTopicArn', { - value: alertTopic.topicArn, - description: 'Amazon SNS alert topic ARN', + new cdk.CfnOutput(this, 'CdcProcessorFunctionName', { + value: cdcProcessorFn.functionName, + description: 'AWS Lambda CDC processor function name', }); new cdk.CfnOutput(this, 'CdcStreamId', { diff --git a/dsql-cdc-eventbridge-fanout-cdk/example-pattern.json b/dsql-cdc-eventbridge-fanout-cdk/example-pattern.json index 4e7fbc86c0..1e29402adc 100644 --- a/dsql-cdc-eventbridge-fanout-cdk/example-pattern.json +++ b/dsql-cdc-eventbridge-fanout-cdk/example-pattern.json @@ -1,6 +1,6 @@ { - "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", + "title": "Amazon Aurora DSQL CDC to Amazon EventBridge", + "description": "Stream real-time database changes from Amazon Aurora DSQL via CDC through Amazon Kinesis and AWS Lambda to Amazon EventBridge", "language": "TypeScript", "level": "300", "framework": "AWS CDK", @@ -9,7 +9,7 @@ "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)." + "Add Amazon EventBridge rules and targets to route events to any downstream consumer such as Amazon SQS, AWS Step Functions, or Amazon SNS." ] }, "gitHub": { @@ -29,10 +29,6 @@ { "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" } ] }, @@ -40,15 +36,14 @@ "text": [ "cd dsql-cdc-eventbridge-fanout-cdk/cdk", "npm install", + "npm run build", "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." + "Add an Amazon EventBridge rule on the dsql-cdc-events bus to verify events arrive." ] }, "cleanup": { @@ -59,55 +54,46 @@ "authors": [ { "name": "Nithin Chandran R", - "image": "", "bio": "Technical Account Manager at AWS", - "linkedin": "nithin-chandran-r", - "twitter": "" + "linkedin": "nithin-chandran-r" } ], "patternArch": { "icon1": { - "x": 10, + "x": 15, "y": 50, "service": "dsql", "label": "Amazon Aurora DSQL" }, "icon2": { - "x": 30, + "x": 38, "y": 50, "service": "kinesis", "label": "Amazon Kinesis" }, "icon3": { - "x": 50, + "x": 62, "y": 50, "service": "lambda", "label": "AWS Lambda" }, "icon4": { - "x": 70, + "x": 85, "y": 50, "service": "eventbridge", "label": "Amazon EventBridge" }, - "icon5": { - "x": 90, - "y": 25, - "service": "sqs", - "label": "Amazon SQS" + "line1": { + "from": "icon1", + "to": "icon2" }, - "icon6": { - "x": 90, - "y": 50, - "service": "sfn", - "label": "AWS Step Functions" + "line2": { + "from": "icon2", + "to": "icon3" }, - "icon7": { - "x": 90, - "y": 75, - "service": "sns", - "label": "Amazon SNS" + "line3": { + "from": "icon3", + "to": "icon4" } - }, - "patternType": "Serverless" + } }