Merge pull request aws-samples#1611 from raffaeu/raffaeu-feature-sqs-stepfunctions-message-aggregator

Author: boyney123

sqs-lambda-dynamodb-message-aggregator-cdk/README.md

@@ -0,0 +1,89 @@
+# SQS to Lambda and DynamoDB (Message Aggregator pattern)
+
+This pattern create two SQS queue, one for receiving the splitted messages and one for the final aggregation and a lambda function and a DynamoDB Table. The pattern is an implementation of the Integration pattern: "Message Aggregator" available here: [https://www.enterpriseintegrationpatterns.com/patterns/messaging/Aggregator.html](https://www.enterpriseintegrationpatterns.com/patterns/messaging/Aggregator.html)
+
+Learn more about this pattern at Serverless Land Patterns: [https://serverlessland.com/patterns/sqs-lambda-dynamodb-message-aggregator-cdk](https://serverlessland.com/patterns/sqs-lambda-dynamodb-message-aggregator-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. You are responsible for any AWS costs incurred. No warranty is implied in this example.
+
+## Requirements
+
+* [Create an AWS account](https://portal.aws.amazon.com/gp/aws/developer/registration/index.html) if you do not already have one and log in. The IAM user that you use must have sufficient permissions to make necessary AWS service calls and manage AWS resources.
+* [AWS CLI](https://docs.aws.amazon.com/cli/latest/userguide/install-cliv2.html) installed and configured
+* [Git Installed](https://git-scm.com/book/en/v2/Getting-Started-Installing-Git)
+* [AWS CDK](https://docs.aws.amazon.com/cdk/v2/guide/getting_started.html) (AWS SAMCDK) installed
+
+## Deployment Instructions
+
+1. Create a new directory, navigate to that directory in a terminal and clone the GitHub repository:
+    ```bash 
+    $: git clone https://github.com/aws-samples/serverless-patterns
+    ```
+1. Change directory to the pattern directory:
+    ```bash
+    $: cd sqs-lambda-dynamodb-message-aggregator-cdk/src
+    ```
+1. From the command line, use AWS SAM to deploy the AWS resources for the pattern as specified in the template.yml file:
+    ```bash
+    $: cdk deploy
+    ```
+1. Note the outputs from the CDK deployment process. These contain the resource names and/or ARNs which are used for testing.
+
+## How it works
+
+The pattern is composed by an SQS Queue which is capable to receive portion of messages. Messages can be aggregated together by providing, for each message, two distinct attributes. `CorrelationId` will inform the AWS Lambda function how to aggregate multiple messages and `Total` will inform the AWS Lambda about the total amount of messages that should be expected for one single aggregation.
+
+Messages are tracked inside a DynamoDB table. When all messages with the same `CorrelationId` are received, the message is aggregated, forwarded to a destination SQS Queue and the record is deleted from DynamoDB, to avoid unecessary costs.
+
+## Testing
+
+Inside the Folder `msg` there are 3 distinct messages and the relative attribute file. The script `test.sh` will send all three messages to the source SqS queue.
+
+```bash
+$: ./msg/test.sh
+Provide the name of the SQS queue to send messages to
+https://sqs.us-east-1.amazonaws.com/[your account]/[your account]-src-queue
+{
+    "MD5OfMessageBody": "8be1b3b225c0f5113fa4c37f421667d2",
+    "MD5OfMessageAttributes": "a512f8c4c91e1bd27eaedaa9916243fe",
+    "MessageId": "35527c0c-3dcc-4ec7-8afa-b997019d0549"
+}
+{
+    "MD5OfMessageBody": "4018bae8eed190d3c907720c3f3cdf1c",
+    "MD5OfMessageAttributes": "a512f8c4c91e1bd27eaedaa9916243fe",
+    "MessageId": "dda99dbf-2f97-4e88-ab28-68011b707a9a"
+}
+{
+    "MD5OfMessageBody": "a9d29a076761cb5a4a1f955814338fcb",
+    "MD5OfMessageAttributes": "a512f8c4c91e1bd27eaedaa9916243fe",
+    "MessageId": "33e07f8d-5c37-4ffa-aaef-b8f90b9b7646"
+}
+```
+
+The script will wait 5 seconds then try to pull the destination queue:
+
+```bash
+Provide the name of the SQS queue to receive the aggregated message
+https://sqs.us-east-1.amazonaws.com/[your account]/[your account]-dest-queue
+{
+    "Messages": [
+        {
+            "MessageId": "52505cb6-7013-4f0f-a6f7-4d4f2f5e223d",
+            "ReceiptHandle": "AQEB8E0k8R6Qn4BdU8fxWM93NTavPuy6+0lbtg7avweEoMWgETQar1puOAROTB5XZuiqSyaEi7pRH6qaBAgfEuzBLX9aC4gGxRPBXrYyu3byntEk2k6T68q2sLjrhlHpa5dbUuuJGk5ahluQkbMaLqtq3jMRHfVp8QLrzAGdBqZzIGLTJAI9kfD1H38anwnlWF1fSMsY/jC21spEeq8zM72w4Lc83qMynrsPJMCysrzoyG7/PVXYnf8OWHj+am9FPzGPSq0YE+1nW1ATJKEcug2DXJgVdbWpXe+uJ5osOHfEk+e/pTf0uHd2ZL1trr1ICwU4F4Af81tZH8GYP4S7aJ8EMLZycyZDxXBuoHIj2wh86E0gy4Ke0vygbskyYkLAZXSqmiwP87oCIYUcxuITgiA3Hg==",
+            "MD5OfBody": "0871a008c15d4c8ccede318f8b942ee5",
+            "Body": "{\"VAT\":\"7.7\",\"Street\":\"Louisiana Road 11\",\"Bank\":\"New York State Bank\",\"TotalAmount\":\"123.00 USD\",\"OrderId\":\"123ABC\",\"State\":\"NY\",\"Transaction\":\"123ABC\",\"Date\":\"2023-01-01\",\"Time\":\"08:00:00\"}"
+        }
+    ]
+}
+```
+
+## Cleanup
+ 
+1. Delete the stack
+    ```bash
+    $: cdk destroy
+    ```
+----
+Copyright 2023 Amazon.com, Inc. or its affiliates. All Rights Reserved.
+
+SPDX-License-Identifier: MIT-0

sqs-lambda-dynamodb-message-aggregator-cdk/example-pattern.json

@@ -0,0 +1,55 @@
+{
+  "title": "Message Aggregator with SQS, Lambda and DynamoDB",
+  "description": "Creates a Lambda function that keep tracks of single messages using DynamoDB and forward a single aggregate message to a different SQS queue.",
+  "language": "TypeScript",
+  "level": "200",
+  "framework": "CDK",
+  "introBox": {
+    "headline": "How it works",
+    "text": [
+      "This CDK project demonstrate how to create a Message Aggregator pattern using Amazon SQS, AWS Lambda and Amazon DynamoDB.",
+      "The solution creates a source and a destination SQS Queue and a Lambda function that keeps track of the messages by leveraging a DynamoDB Table.",
+      "This architecture is an implementation of the Integration pattern: Message Aggregator explained in the Integration Patterns book."
+    ]
+  },
+  "gitHub": {
+    "template": {
+      "repoURL": "https://github.com/aws-samples/serverless-patterns/tree/main/sqs-lambda-dynamodb-message-aggregator-cdk",
+      "templateURL": "serverless-patterns/sqs-lambda-dynamodb-message-aggregator-cdk",
+      "projectFolder": "sqs-lambda-dynamodb-message-aggregator-cdk",
+      "templateFile": "sqs-lambda-dynamodb-message-aggregator-cdk/src/lib/src-stack.ts"
+    }
+  },
+  "resources": {
+    "bullets": [
+      {
+        "text": "Message Aggregator Pattern",
+        "link": "https://www.enterpriseintegrationpatterns.com/patterns/messaging/Aggregator.html"
+      }
+    ]
+  },
+  "deploy": {
+    "text": [
+      "cdk deploy"
+    ]
+  },
+  "testing": {
+    "text": [
+      "See the GitHub repo for detailed testing instructions."
+    ]
+  },
+  "cleanup": {
+    "text": [
+      "Delete the stack: <code>cdk destroy</code>."
+    ]
+  },
+  "authors": [
+    {
+      "name": "Raffaele Garofalo (Raf)",
+      "image": "https://avatars.githubusercontent.com/raffaeu",
+      "bio": "Raffaele Garofalo (Raf) is a Senior Solutions Architect at AWS and member of the AWS Serverless TFC (Technical Field Community).",
+      "linkedin": "raffaeu",
+      "twitter": "raffaeu"
+    }
+  ]
+}

sqs-lambda-dynamodb-message-aggregator-cdk/src/.gitignore

@@ -0,0 +1,9 @@
+*.js
+!jest.config.js
+*.d.ts
+node_modules
+
+# CDK asset staging directory
+.cdk.staging
+cdk.out
+!code/*.js

sqs-lambda-dynamodb-message-aggregator-cdk/src/.npmignore

@@ -0,0 +1,6 @@
+*.ts
+!*.d.ts
+
+# CDK asset staging directory
+.cdk.staging
+cdk.out

sqs-lambda-dynamodb-message-aggregator-cdk/src/bin/src.ts

@@ -0,0 +1,21 @@
+#!/usr/bin/env node
+import 'source-map-support/register';
+import * as cdk from 'aws-cdk-lib';
+import { SrcStack } from '../lib/src-stack';
+
+const app = new cdk.App();
+new SrcStack(app, 'SrcStack', {
+  /* If you don't specify 'env', this stack will be environment-agnostic.
+   * Account/Region-dependent features and context lookups will not work,
+   * but a single synthesized template can be deployed anywhere. */
+
+  /* Uncomment the next line to specialize this stack for the AWS Account
+   * and Region that are implied by the current CLI configuration. */
+  // env: { account: process.env.CDK_DEFAULT_ACCOUNT, region: process.env.CDK_DEFAULT_REGION },
+
+  /* Uncomment the next line if you know exactly what Account and Region you
+   * want to deploy the stack to. */
+  // env: { account: '123456789012', region: 'us-east-1' },
+
+  /* For more information, see https://docs.aws.amazon.com/cdk/latest/guide/environments.html */
+});

sqs-lambda-dynamodb-message-aggregator-cdk/src/cdk.json

@@ -0,0 +1,55 @@
+{
+  "app": "npx ts-node --prefer-ts-exts bin/src.ts",
+  "watch": {
+    "include": [
+      "**"
+    ],
+    "exclude": [
+      "README.md",
+      "cdk*.json",
+      "**/*.d.ts",
+      "**/*.js",
+      "tsconfig.json",
+      "package*.json",
+      "yarn.lock",
+      "node_modules",
+      "test"
+    ]
+  },
+  "context": {
+    "@aws-cdk/aws-lambda:recognizeLayerVersion": true,
+    "@aws-cdk/core:checkSecretUsage": true,
+    "@aws-cdk/core:target-partitions": [
+      "aws",
+      "aws-cn"
+    ],
+    "@aws-cdk-containers/ecs-service-extensions:enableDefaultLogDriver": true,
+    "@aws-cdk/aws-ec2:uniqueImdsv2TemplateName": true,
+    "@aws-cdk/aws-ecs:arnFormatIncludesClusterName": true,
+    "@aws-cdk/aws-iam:minimizePolicies": true,
+    "@aws-cdk/core:validateSnapshotRemovalPolicy": true,
+    "@aws-cdk/aws-codepipeline:crossAccountKeyAliasStackSafeResourceName": true,
+    "@aws-cdk/aws-s3:createDefaultLoggingPolicy": true,
+    "@aws-cdk/aws-sns-subscriptions:restrictSqsDescryption": true,
+    "@aws-cdk/aws-apigateway:disableCloudWatchRole": true,
+    "@aws-cdk/core:enablePartitionLiterals": true,
+    "@aws-cdk/aws-events:eventsTargetQueueSameAccount": true,
+    "@aws-cdk/aws-iam:standardizedServicePrincipals": true,
+    "@aws-cdk/aws-ecs:disableExplicitDeploymentControllerForCircuitBreaker": true,
+    "@aws-cdk/aws-iam:importedRoleStackSafeDefaultPolicyName": true,
+    "@aws-cdk/aws-s3:serverAccessLogsUseBucketPolicy": true,
+    "@aws-cdk/aws-route53-patters:useCertificate": true,
+    "@aws-cdk/customresources:installLatestAwsSdkDefault": false,
+    "@aws-cdk/aws-rds:databaseProxyUniqueResourceName": true,
+    "@aws-cdk/aws-codedeploy:removeAlarmsFromDeploymentGroup": true,
+    "@aws-cdk/aws-apigateway:authorizerChangeDeploymentLogicalId": true,
+    "@aws-cdk/aws-ec2:launchTemplateDefaultUserData": true,
+    "@aws-cdk/aws-secretsmanager:useAttachedSecretResourcePolicyForSecretTargetAttachments": true,
+    "@aws-cdk/aws-redshift:columnId": true,
+    "@aws-cdk/aws-stepfunctions-tasks:enableEmrServicePolicyV2": true,
+    "@aws-cdk/aws-ec2:restrictDefaultSecurityGroup": true,
+    "@aws-cdk/aws-apigateway:requestValidatorUniqueId": true,
+    "@aws-cdk/aws-kms:aliasNameRef": true,
+    "@aws-cdk/core:includePrefixInUniqueNameGeneration": true
+  }
+}

sqs-lambda-dynamodb-message-aggregator-cdk/src/code/handler.js

@@ -0,0 +1,66 @@
+const AWS = require('aws-sdk');
+
+exports.handler = async (event, context, callback) => {
+
+    // retrieve dynamo name from environment variable
+    const dynamoDB = new AWS.DynamoDB.DocumentClient();
+    const queue = new AWS.SQS();
+
+    // loop through all sqs records
+    for (const record of event.Records) {
+
+        // create dynamo record
+        const correlationId = record.messageAttributes.CorrelationId.stringValue;
+        const total = Number(record.messageAttributes.Total.stringValue);
+        const body = JSON.parse(record.body);
+        let item = {
+            id: correlationId,
+            body: body,
+            count: 0
+        };
+
+        // check if item exists in dynamo
+        const dynamoRecord = await dynamoDB.get({
+            TableName: process.env.DYNAMODB_TABLE_NAME,
+            Key: {
+                'id': correlationId
+            }
+        }).promise();
+        // if item exists, update item
+        if (dynamoRecord.Item) {
+            item.body = Object.assign(dynamoRecord.Item.body, item.body);
+            item.count = dynamoRecord.Item.count + 1;
+        } else {
+            // if item doesn't exist, create item
+            item.body = item.body;
+            item.count = 1;
+        }
+        // put item in dynamo
+        const result = await dynamoDB.put({
+            TableName: process.env.DYNAMODB_TABLE_NAME,
+            Item: item,
+            ReturnValues: 'ALL_OLD'
+        }).promise();
+
+        // if item is last, trigger aggregation
+        // and delete item from Dynamo
+        if (item.count === total) {
+            await queue.sendMessage({
+                QueueUrl: process.env.DESTINATION_QUEUE_URL,
+                MessageBody: JSON.stringify(item.body)
+            }).promise();
+            await dynamoDB.delete({
+                TableName: process.env.DYNAMODB_TABLE_NAME,
+                Key: {
+                    'id': correlationId
+                }
+            }).promise();
+        }
+    }
+
+    //complete
+    callback(null, {
+        statusCode: '200',
+        body: JSON.stringify({ 'status': 'complete' })
+    });
+};

sqs-lambda-dynamodb-message-aggregator-cdk/src/jest.config.js

@@ -0,0 +1,8 @@
+module.exports = {
+  testEnvironment: 'node',
+  roots: ['<rootDir>/test'],
+  testMatch: ['**/*.test.ts'],
+  transform: {
+    '^.+\\.tsx?$': 'ts-jest'
+  }
+};

sqs-lambda-dynamodb-message-aggregator-cdk/src/lib/src-stack.ts

@@ -0,0 +1,50 @@
+import * as cdk from 'aws-cdk-lib';
+import { Construct } from 'constructs';
+
+export class SrcStack extends cdk.Stack {
+  constructor(scope: Construct, id: string, props?: cdk.StackProps) {
+    super(scope, id, props);
+
+    // get account id
+    const account = cdk.Stack.of(this).account;
+
+    // create an sqs queue starting with account id
+    const srcSqs = new cdk.aws_sqs.Queue(this, 'SrcQueue', {
+      queueName: `${account}-src-queue`,
+      removalPolicy: cdk.RemovalPolicy.DESTROY,
+    });
+
+    // create the destination queue
+    const destSqs = new cdk.aws_sqs.Queue(this, 'DestQueue', {
+      queueName: `${account}-dest-queue`,
+      removalPolicy: cdk.RemovalPolicy.DESTROY,
+    });
+
+    // create a dynamodb table to aggregate messages
+    const srcTable = new cdk.aws_dynamodb.Table(this, 'SrcTable', {
+      partitionKey: { name: 'id', type: cdk.aws_dynamodb.AttributeType.STRING },
+      removalPolicy: cdk.RemovalPolicy.DESTROY,
+      tableName: `${account}-events-table`
+    });
+
+    // create a lambda using the file code/lambda.handler.js
+    const srcLambda = new cdk.aws_lambda.Function(this, 'SrcLambda', {
+      code: cdk.aws_lambda.Code.fromAsset('code'),
+      handler: 'handler.handler',
+      runtime: cdk.aws_lambda.Runtime.NODEJS_14_X,
+      environment: {
+        DYNAMODB_TABLE_NAME: srcTable.tableName,
+        DESTINATION_QUEUE_URL: destSqs.queueUrl,
+      },
+      functionName: `${account}-aggregator-lambda`
+    });
+
+    // grant permissions
+    srcSqs.grantConsumeMessages(srcLambda);
+    srcTable.grantReadWriteData(srcLambda);
+    destSqs.grantSendMessages(srcLambda);
+
+    // register src queue as a trigger of the lambda
+    srcLambda.addEventSource(new cdk.aws_lambda_event_sources.SqsEventSource(srcSqs, { batchSize: 10 }));
+  }
+}

sqs-lambda-dynamodb-message-aggregator-cdk/src/msg/attributes-01.json

@@ -0,0 +1,10 @@
+{
+    "CorrelationId": {
+        "DataType": "String",
+        "StringValue": "01"
+    },
+    "Total": {
+        "DataType": "String",
+        "StringValue": "3"
+    }
+}