Event-Driven Architecture with Aurora DSQL Change Data Capture
Inspired by the DynamoDB feature of streaming change events to consumers via DynamoDB Streams the news of Aurora DSQL now generally supporting Change Data Capture (CDC) via Kinesis is an exciting one as implementing Event-Driven Architectures using a native feature of an SQL database reduces complexity on the application side and streamlines consumers by adopting a standardised pattern, in this case a Kinesis Data Stream.
The Scenario
As a proof of concept we are going to look at a relatively simple scenario of a Request Processor that accepts messages containing a URL and stores the status code of the HTTP response when calling that URL in a table. The process is initiated by a Create Requests Lambda that writes “Request Created” events into a database table, then picked up by the Dispatcher Lambda via the CDC Kinesis Stream, dispatched to a SQS Queue and finally picked up by the Process Requests Lambda that processes the requests (by calling the URL) and stores the status code as a “Request Processed” event in the table again. As the structure of the table is relatively simple we are going to use a Single Table design instead of one table per event type which would require some JOIN’s for reading the actual status of an event.
The Infrastructure
Now we need to create some infrastructure for the above, starting with the Aurora DSQL cluster and the Kinesis Data Stream that will be used for CDC (all using TypeScript CDK in the following):
const cluster = new Cluster(this, 'Cluster', {
removalPolicy: RemovalPolicy.DESTROY,
});
const stream = new Stream(this, 'Stream', {
streamName: 'Stream',
removalPolicy: RemovalPolicy.DESTROY,
});next we need an IAM role that Aurora DSQL is going to use to write to Kinesis:
const cdcRole = new Role(this, 'CDCRole', {
assumedBy: new ServicePrincipal('dsql'),
inlinePolicies: {
Default: new PolicyDocument({
statements: [
new PolicyStatement({
actions: [
'kinesis:DescribeStreamSummary',
'kinesis:ListShards',
'kinesis:PutRecord',
'kinesis:PutRecords',
],
resources: [
stream.streamArn
]
})
]
})
},
roleName: 'CDCRole'
});and to create the CDC Stream we currently need to use an AWS Custom Resource as CloudFormation is currently not supporting it, and for this we are referencing the cluster, the stream and the role created above:
new AwsCustomResource(this, 'CreateCDCStream', {
onCreate: {
service: 'dsql',
action: 'createStream',
parameters: {
clusterIdentifier: cluster.clusterId,
targetDefinition: {
kinesis: {
streamArn: stream.streamArn,
roleArn: cdcRole.roleArn,
}
},
ordering: 'UNORDERED',
format: 'JSON',
},
physicalResourceId: PhysicalResourceId.of('create-cdc-stream')
},
policy: AwsCustomResourcePolicy.fromStatements([
new PolicyStatement({
actions: ['dsql:CreateStream'],
resources: [cluster.clusterArn]
}),
new PolicyStatement({
actions: ['iam:PassRole'],
resources: [cdcRole.roleArn]
})
])
});Finally we need to create the three Lambdas, together with Kinesis and SQS Event Sources to invoke the Lambdas:
const queue = new Queue(this, 'RequestQueue', {
queueName: 'RequestQueue',
});
const createRequestLambda = new RustFunction(this, 'CreateRequest', {
entry: 'create-request',
environment: {
'CLUSTER_ENDPOINT': `${cluster.clusterId}.dsql.${this.region}.on.aws`,
},
functionName: 'CreateRequest',
logGroup: new LogGroup(this, 'CreateRequestLogGroup', {
logGroupName: '/aws/lambda/CreateRequest',
removalPolicy: RemovalPolicy.DESTROY,
}),
});
cluster.grantConnectAdmin(createRequestLambda);
const processRequestLambda = new RustFunction(this, 'ProcessRequest', {
entry: 'process-request',
environment: {
'CLUSTER_ENDPOINT': `${cluster.clusterId}.dsql.${this.region}.on.aws`,
},
functionName: 'ProcessRequest',
logGroup: new LogGroup(this, 'ProcessRequestLogGroup', {
logGroupName: '/aws/lambda/ProcessRequest',
removalPolicy: RemovalPolicy.DESTROY,
}),
});
cluster.grantConnectAdmin(processRequestLambda);
const sqsEventSource = new SqsEventSource(queue, {
batchSize: 1,
});
processRequestLambda.addEventSource(sqsEventSource)
const dispatcherLambda = new RustFunction(this, 'Dispatcher', {
entry: 'dispatcher',
environment: {
'QUEUE_URL': queue.queueUrl,
},
functionName: 'Dispatcher',
logGroup: new LogGroup(this, 'DispatcherLogGroup', {
logGroupName: '/aws/lambda/Dispatcher',
removalPolicy: RemovalPolicy.DESTROY,
}),
});
queue.grantSendMessages(dispatcherLambda);
const kinesisEventSource = new KinesisEventSource(stream, {
startingPosition: StartingPosition.LATEST,
});
dispatcherLambda.addEventSource(kinesisEventSource);We are using the Aurora DSQL Admin user for this for simplicities sake, for a more fine grained setup of Aurora DSQL see https://medium.com/@johannesfloriangeiger/amazon-aurora-dsql-with-lambdas-revisited-and-this-time-with-rust-ece95306b902.
Get Johannes Geiger’s stories in your inbox
Join Medium for free to get updates from this writer.
After having deployed everything we are logging into Aurora DSQL via the AWS Console and create the table structure:
CREATE TABLE public.requests (id UUID, request_type VARCHAR(255), url VARCHAR(255), status_code VARCHAR(255), PRIMARY KEY(id, request_type))And when invoking the Create Request Lambda with a URL we can query said table and see the “Request Created” and shortly after the “Request Processed” events:
SELECT * FROM public.requests;Summary
The integration between Aurora DSQL and Kinesis works seamless, and consuming Kinesis with Lambdas is similar. If Aurora DSQL would now also support setting up the CDC Stream via CloudFormation and we wouldn't have to use the AWS SDK via an AWS Custom Resource for this it would be even better.
Appendix
create-request/src/main.rs
use aurora_dsql_sqlx_connector::{retry_on_occ, OCCRetryConfig};
use lambda_runtime::{run, service_fn, Error, LambdaEvent};
use serde::Deserialize;
use sqlx::PgPool;
use std::env;
use uuid::Uuid;
struct Handler<'a> {
pool: &'a PgPool,
}
#[derive(Deserialize, Debug)]
struct CreateRequest {
url: String,
}
impl<'a> Handler<'a> {
async fn handle(&self, lambda_event: LambdaEvent<CreateRequest>) -> Result<(), Error> {
let occ_config = OCCRetryConfig::default();
retry_on_occ(&occ_config, || async {
let mut transaction = self.pool.begin().await?;
sqlx::query("INSERT INTO public.requests(id, request_type, url) VALUES ($1, $2, $3);")
.bind(Uuid::new_v4())
.bind("REQUEST_CREATED")
.bind(lambda_event.payload.url.clone())
.execute(&mut *transaction)
.await?;
transaction.commit().await?;
Ok(())
})
.await?;
Ok(())
}
}
#[tokio::main]
async fn main() -> Result<(), Error> {
let cluster_endpoint = env::var("CLUSTER_ENDPOINT")?;
let connection_string = format!("postgres://admin@{}/postgres", cluster_endpoint);
let pool = aurora_dsql_sqlx_connector::pool::connect(connection_string.as_str()).await?;
let handler = Handler { pool: &pool };
run(service_fn(|lambda_event: LambdaEvent<CreateRequest>| {
handler.handle(lambda_event)
}))
.await?;
pool.close().await;
Ok(())
}dispatcher/src/main.rs
use aws_config::{BehaviorVersion, load_defaults};
use aws_lambda_events::kinesis::KinesisEvent;
use aws_sdk_sqs::{Client, types::builders::MessageAttributeValueBuilder};
use lambda_runtime::{Error, LambdaEvent, run, service_fn};
use serde::Deserialize;
use std::env;
struct Handler<'a> {
queue_url: String,
client: &'a Client,
}
#[derive(Deserialize, Debug)]
struct After {
id: String,
request_type: String,
url: String,
}
#[derive(Deserialize, Debug)]
struct CDCMessage {
op: String,
after: Option<After>,
}
const OPERATION_CREATE: &'static str = "c";
impl<'a> Handler<'a> {
async fn handle(&self, lambda_event: LambdaEvent<KinesisEvent>) -> Result<(), Error> {
for kinesis_event_record in lambda_event.payload.records {
let cdc_message: CDCMessage =
serde_json::from_slice(kinesis_event_record.kinesis.data.0.as_slice())?;
let operation = cdc_message.op;
let create = operation == OPERATION_CREATE;
if create {
let after = cdc_message.after.unwrap();
let id = uuid::Uuid::parse_str(after.id.as_str()).unwrap();
let request_type = after.request_type;
println!("Request received: {} ({}).", request_type, id);
match request_type.as_str() {
"REQUEST_CREATED" => {
println!("Request created.");
self.client
.send_message()
.queue_url(self.queue_url.clone())
.message_attributes(
"id",
MessageAttributeValueBuilder::default()
.string_value(id)
.data_type("String".to_string())
.build()?,
)
.message_body(after.url.as_str())
.send()
.await?;
}
"REQUEST_PROCESSED" => {
println!("Request processed.");
}
_ => {
println!("Unknown request type: {}", request_type);
}
}
println!("Request handled ({}).", id);
} else {
println!("Unknown operation: {}", operation);
}
}
Ok(())
}
}
#[tokio::main]
async fn main() -> Result<(), Error> {
let queue_url = env::var("QUEUE_URL")?;
let sdk_config = load_defaults(BehaviorVersion::latest()).await;
let client = Client::new(&sdk_config);
let handler = Handler {
queue_url,
client: &client,
};
run(service_fn(|lambda_event: LambdaEvent<KinesisEvent>| {
handler.handle(lambda_event)
}))
.await?;
Ok(())
}process-request/src/main.rs
use aurora_dsql_sqlx_connector::{OCCRetryConfig, retry_on_occ};
use aws_lambda_events::sqs::SqsEvent;
use lambda_runtime::{Error, LambdaEvent, run, service_fn};
use sqlx::PgPool;
use std::env;
struct Handler<'a> {
pool: &'a PgPool,
}
impl<'a> Handler<'a> {
async fn handle(&self, lambda_event: LambdaEvent<SqsEvent>) -> Result<(), Error> {
for sqs_message in lambda_event.payload.records {
let url = sqs_message.body.unwrap();
let id = sqlx::types::Uuid::parse_str(
sqs_message
.message_attributes
.get("id")
.unwrap()
.string_value
.clone()
.unwrap()
.as_str(),
)?;
let status_code = ureq::get(url.as_str()).call()?.status();
let occ_config = OCCRetryConfig::default();
retry_on_occ(&occ_config, || async {
let mut transaction = self.pool.begin().await?;
sqlx::query("INSERT INTO public.requests(id, request_type, url, status_code) VALUES ($1, $2, $3, $4);")
.bind(id)
.bind("REQUEST_PROCESSED")
.bind(url.clone())
.bind(status_code.as_str())
.execute(&mut *transaction)
.await?;
transaction.commit().await?;
Ok(())
}).await?;
}
Ok(())
}
}
#[tokio::main]
async fn main() -> Result<(), Error> {
let cluster_endpoint = env::var("CLUSTER_ENDPOINT")?;
let connection_string = format!("postgres://admin@{}/postgres", cluster_endpoint);
let pool = aurora_dsql_sqlx_connector::pool::connect(connection_string.as_str()).await?;
let handler = Handler { pool: &pool };
run(service_fn(|lambda_event: LambdaEvent<SqsEvent>| {
handler.handle(lambda_event)
}))
.await?;
pool.close().await;
Ok(())
}
