add DynamoDB

packages/@aws-cdk/aws-pipes-sources-alpha/README.md

@@ -72,7 +72,33 @@ declare const sourceStream: kinesis.Stream;
 declare const targetQueue: sqs.Queue;
 
 const pipeSource = new sources.KinesisSource(sourceStream, {
- startingPosition: sources.StartingPosition.LATEST,
+ startingPosition: sources.KinesisStartingPosition.LATEST,
+});
+
+const pipe = new pipes.Pipe(this, 'Pipe', {
+ source: pipeSource,
+ target: new SomeTarget(targetQueue)
+});
+```
+
+### Amazon DynamoDB
+
+A DynamoDB stream can be used as a source for a pipe. The stream will be polled for new messages and the messages will be sent to the pipe.
+
+```ts
+import * as ddb from 'aws-cdk-lib/aws-dynamodb';
+
+const table = new ddb.TableV2(stack, 'MyTable', {
+ partitionKey: {
+ name: 'id',
+ type: ddb.AttributeType.STRING,
+ },
+ dynamoStream: ddb.StreamViewType.NEW_IMAGE,
+});
+declare const targetQueue: sqs.Queue;
+
+const pipeSource = new sources.DynamoDBSource(table, {
+ startingPosition: sources.DynamoDBStartingPosition.LATEST,
 });
 
 const pipe = new pipes.Pipe(this, 'Pipe', {

packages/@aws-cdk/aws-pipes-sources-alpha/lib/dynamodb.ts

@@ -0,0 +1,158 @@
+import { IPipe, ISource, SourceConfig } from '@aws-cdk/aws-pipes-alpha';
+import { Duration } from 'aws-cdk-lib';
+import { IRole } from 'aws-cdk-lib/aws-iam';
+import { ITableV2 } from 'aws-cdk-lib/aws-dynamodb';
+import { DeadLetterConfigParameters } from './deadLetterConfig';
+import { DynamoDBStartingPosition, OnPartialBatchItemFailure } from './enums';
+
+/**
+ * Parameters for the DynamoDB source.
+ */
+export interface DynamoDBSourceParameters {
+ /**
+ * The maximum number of records to include in each batch.
+ *
+ * @see https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-pipes-pipe-pipesourcedynamodbstreamparameters.html#cfn-pipes-pipe-pipesourcedynamodbstreamparameters-batchsize
+ * @default 1
+ */
+ readonly batchSize?: number;
+
+ /**
+ * Define the target queue to send dead-letter queue events to.
+ *
+ * @see https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-pipes-pipe-pipesourcedynamodbstreamparameters.html#cfn-pipes-pipe-pipesourcedynamodbstreamparameters-deadletterconfig
+ * @default no dead letter queue
+ */
+ readonly deadLetterConfig?: DeadLetterConfigParameters;
+
+ /**
+ * The maximum length of a time to wait for events.
+ *
+ * @see https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-pipes-pipe-pipesourcedynamodbstreamparameters.html#cfn-pipes-pipe-pipesourcedynamodbstreamparameters-maximumbatchingwindowinseconds
+ * @default no batching window
+ */
+ readonly maximumBatchingWindow?: Duration;
+
+ /**
+ * (Streams only) Discard records older than the specified age. The default value is -1, which sets the maximum age to infinite. When the value is set to infinite, EventBridge never discards old records.
+ *
+ * Leave undefined to set the maximum record age to infinite.
+ *
+ * @see https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-pipes-pipe-pipesourcedynamodbstreamparameters.html#cfn-pipes-pipe-pipesourcedynamodbstreamparameters-maximumrecordageinseconds
+ * @default -1 (infinite)
+ */
+ readonly maximumRecordAge?: Duration;
+
+ /**
+ * (Streams only) Discard records after the specified number of retries. The default value is -1, which sets the maximum number of retries to infinite. When MaximumRetryAttempts is infinite, EventBridge retries failed records until the record expires in the event source.
+ *
+ * @see https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-pipes-pipe-pipesourcedynamodbstreamparameters.html#cfn-pipes-pipe-pipesourcedynamodbstreamparameters-maximumretryattempts
+ * @default -1 (infinite)
+ */
+ readonly maximumRetryAttempts?: number;
+
+ /**
+ * (Streams only) Define how to handle item process failures. AUTOMATIC_BISECT halves each batch and retry each half until all the records are processed or there is one failed message left in the batch.
+ *
+ * @see https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-pipes-pipe-pipesourcedynamodbstreamparameters.html#cfn-pipes-pipe-pipesourcedynamodbstreamparameters-onpartialbatchitemfailure
+ * @default off
+ */
+ readonly onPartialBatchItemFailure?: OnPartialBatchItemFailure;
+
+ /**
+ * (Streams only) The number of batches to process concurrently from each shard. The default value is 1.
+ *
+ * @see https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-pipes-pipe-pipesourcedynamodbstreamparameters.html#cfn-pipes-pipe-pipesourcedynamodbstreamparameters-parallelizationfactor
+ * @default 1
+ */
+ readonly parallelizationFactor?: number;
+
+ /**
+ * (Streams only) The position in a stream from which to start reading.
+ *
+ * @see https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-pipes-pipe-pipesourcedynamodbstreamparameters.html#cfn-pipes-pipe-pipesourcedynamodbstreamparameters-startingposition
+ */
+ readonly startingPosition: DynamoDBStartingPosition;
+}
+
+/**
+ * A source that reads from an DynamoDB stream.
+ */
+export class DynamoDBSource implements ISource {
+ private readonly table: ITableV2;
+ readonly sourceArn;
+ private sourceParameters;
+
+ private batchSize;
+ private maximumBatchingWindowInSeconds;
+ private maximumRecordAgeInSeconds;
+ private maximumRetryAttempts;
+ private parallelizationFactor;
+
+ constructor(table: ITableV2, parameters: DynamoDBSourceParameters) {
+ this.table = table;
+
+ if (table.tableStreamArn === undefined) {
+ throw new Error('Table does not have a stream defined, cannot create pipes source');
+ }
+
+ this.sourceArn = table.tableStreamArn;
+ this.sourceParameters = parameters;
+
+ this.batchSize = this.sourceParameters.batchSize;
+ this.maximumBatchingWindowInSeconds = this.sourceParameters.maximumBatchingWindow?.toSeconds();
+ this.maximumRecordAgeInSeconds = this.sourceParameters.maximumRecordAge?.toSeconds();
+ this.maximumRetryAttempts = this.sourceParameters.maximumRetryAttempts;
+ this.parallelizationFactor = this.sourceParameters.parallelizationFactor;
+
+ if (this.batchSize !== undefined) {
+ if (this.batchSize < 1 || this.batchSize > 10000) {
+ throw new Error(`Batch size must be between 1 and 10000, received ${this.batchSize}`);
+ }
+ }
+ if (this.maximumBatchingWindowInSeconds !== undefined) {
+ // only need to check upper bound since Duration amounts cannot be negative
+ if (this.maximumBatchingWindowInSeconds > 300) {
+ throw new Error(`Maximum batching window must be between 0 and 300, received ${this.maximumBatchingWindowInSeconds}`);
+ }
+ }
+ if (this.maximumRecordAgeInSeconds !== undefined) {
+ // only need to check upper bound since Duration amounts cannot be negative
+ if (this.maximumRecordAgeInSeconds > 604800) {
+ throw new Error(`Maximum record age in seconds must be between -1 and 604800, received ${this.maximumRecordAgeInSeconds}`);
+ }
+ }
+ if (this.maximumRetryAttempts !== undefined) {
+ if (this.maximumRetryAttempts < -1 || this.maximumRetryAttempts > 10000) {
+ throw new Error(`Maximum retry attempts must be between -1 and 10000, received ${this.maximumRetryAttempts}`);
+ }
+ }
+ if (this.parallelizationFactor !== undefined) {
+ if (this.parallelizationFactor < 1 || this.parallelizationFactor > 10) {
+ throw new Error(`Parallelization factor must be between 1 and 10, received ${this.parallelizationFactor}`);
+ }
+ }
+ }
+
+ bind(_pipe: IPipe): SourceConfig {
+ return {
+ sourceParameters: {
+ dynamoDbStreamParameters: {
+ batchSize: this.batchSize,
+ deadLetterConfig: this.sourceParameters.deadLetterConfig,
+ maximumBatchingWindowInSeconds: this.maximumBatchingWindowInSeconds,
+ maximumRecordAgeInSeconds: this.maximumRecordAgeInSeconds,
+ maximumRetryAttempts: this.maximumRetryAttempts,
+ onPartialBatchItemFailure: this.sourceParameters.onPartialBatchItemFailure,
+ parallelizationFactor: this.sourceParameters.parallelizationFactor,
+ startingPosition: this.sourceParameters.startingPosition,
+ },
+ },
+ };
+ }
+
+ grantRead(grantee: IRole): void {
+ this.table.grantStreamRead(grantee);
+ }
+}
+

packages/@aws-cdk/aws-pipes-sources-alpha/lib/enums.ts

@@ -0,0 +1,41 @@
+/**
+ * Define how to handle item process failures.
+ */
+export enum OnPartialBatchItemFailure {
+ /**
+ * AUTOMATIC_BISECT
+ */
+ AUTOMATIC_BISECT = 'AUTOMATIC_BISECT',
+}
+
+/**
+ * The position in a Kinesis stream from which to start reading.
+ */
+export enum KinesisStartingPosition {
+ /**
+ * TRIM_HORIZON
+ */
+ TRIM_HORIZON = 'TRIM_HORIZON',
+ /**
+ * LATEST
+ */
+ LATEST = 'LATEST',
+ /**
+ * AT_TIMESTAMP
+ */
+ AT_TIMESTAMP = 'AT_TIMESTAMP',
+}
+
+/**
+ * The position in a DynamoDB stream from which to start reading.
+ */
+export enum DynamoDBStartingPosition {
+ /**
+ * TRIM_HORIZON
+ */
+ TRIM_HORIZON = 'TRIM_HORIZON',
+ /**
+ * LATEST
+ */
+ LATEST = 'LATEST',
+}

packages/@aws-cdk/aws-pipes-sources-alpha/lib/index.ts

@@ -1,3 +1,5 @@
 export * from './sqs';
 export * from './kinesis';
+export * from './dynamodb';
+export * from './enums';
 export * from './deadLetterConfig';

packages/@aws-cdk/aws-pipes-sources-alpha/lib/kinesis.ts

@@ -3,6 +3,7 @@ import { Duration } from 'aws-cdk-lib';
 import { IRole } from 'aws-cdk-lib/aws-iam';
 import { IStream } from 'aws-cdk-lib/aws-kinesis';
 import { DeadLetterConfigParameters } from './deadLetterConfig';
+import { KinesisStartingPosition, OnPartialBatchItemFailure } from './enums';
 
 /**
  * Parameters for the Kinesis source.
@@ -71,7 +72,7 @@ export interface KinesisSourceParameters {
  *
  * @see https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-properties-pipes-pipe-pipesourcekinesisstreamparameters.html#cfn-pipes-pipe-pipesourcekinesisstreamparameters-startingposition
  */
- readonly startingPosition: StartingPosition;
+ readonly startingPosition: KinesisStartingPosition;
 
  /**
  * With StartingPosition set to AT_TIMESTAMP, the time from which to start reading, in Unix time seconds.
@@ -82,41 +83,14 @@ export interface KinesisSourceParameters {
  readonly startingPositionTimestamp?: string;
 }
 
-/**
- * Define how to handle item process failures.
- */
-export enum OnPartialBatchItemFailure {
- /**
- * AUTOMATIC_BISECT
- */
- AUTOMATIC_BISECT = 'AUTOMATIC_BISECT',
-}
-
-/**
- * The position in a stream from which to start reading.
- */
-export enum StartingPosition {
- /**
- * TRIM_HORIZON
- */
- TRIM_HORIZON = 'TRIM_HORIZON',
- /**
- * LATEST
- */
- LATEST = 'LATEST',
- /**
- * AT_TIMESTAMP
- */
- AT_TIMESTAMP = 'AT_TIMESTAMP',
-}
-
 /**
  * A source that reads from Kinesis.
  */
 export class KinesisSource implements ISource {
  private readonly stream: IStream;
  readonly sourceArn;
  private sourceParameters;
+
  private batchSize;
  private maximumBatchingWindowInSeconds;
  private maximumRecordAgeInSeconds;
@@ -127,6 +101,7 @@ export class KinesisSource implements ISource {
  this.stream = stream;
  this.sourceArn = stream.streamArn;
  this.sourceParameters = parameters;
+
  this.batchSize = this.sourceParameters.batchSize;
  this.maximumBatchingWindowInSeconds = this.sourceParameters.maximumBatchingWindow?.toSeconds();
  this.maximumRecordAgeInSeconds = this.sourceParameters.maximumRecordAge?.toSeconds();

packages/@aws-cdk/aws-pipes-sources-alpha/test/__snapshots__/dynamodb.test.ts.snap

@@ -0,0 +1,68 @@
+// Jest Snapshot v1, https://goo.gl/fbAQLP
+
+exports[`dynamodb source should grant pipe role read access 1`] = `
+{
+ "MyPipeRoleCBC8E9AB": {
+ "Properties": {
+ "AssumeRolePolicyDocument": {
+ "Statement": [
+ {
+ "Action": "sts:AssumeRole",
+ "Effect": "Allow",
+ "Principal": {
+ "Service": "pipes.amazonaws.com",
+ },
+ },
+ ],
+ "Version": "2012-10-17",
+ },
+ },
+ "Type": "AWS::IAM::Role",
+ },
+}
+`;
+
+exports[`dynamodb source should grant pipe role read access 2`] = `
+{
+ "MyPipeRoleDefaultPolicy31387C20": {
+ "Properties": {
+ "PolicyDocument": {
+ "Statement": [
+ {
+ "Action": "dynamodb:ListStreams",
+ "Effect": "Allow",
+ "Resource": {
+ "Fn::GetAtt": [
+ "MyTable794EDED1",
+ "StreamArn",
+ ],
+ },
+ },
+ {
+ "Action": [
+ "dynamodb:DescribeStream",
+ "dynamodb:GetRecords",
+ "dynamodb:GetShardIterator",
+ ],
+ "Effect": "Allow",
+ "Resource": {
+ "Fn::GetAtt": [
+ "MyTable794EDED1",
+ "StreamArn",
+ ],
+ },
+ },
+ ],
+ "Version": "2012-10-17",
+ },
+ "PolicyName": "MyPipeRoleDefaultPolicy31387C20",
+ "Roles": [
+ {
+ "Ref": "MyPipeRoleCBC8E9AB",
+ },
+ ],
+ },
+ "Type": "AWS::IAM::Policy",
+ },
+}
+`;