Merge pull request #7 from nideveloper/master

Adding in The Scalable Webhook Pattern

Author: cdk-patterns

README.md

@@ -20,6 +20,9 @@ These patterns are from https://www.jeremydaly.com/serverless-microservice-patte
 #### [The Simple Webservice](/the-simple-webservice/README.md)
 ![Architecture](https://raw.githubusercontent.com/cdk-patterns/serverless/master/the-simple-webservice/img/architecture.png)
 
+#### [The Scalable Webhook](/the-scalable-webhook/README.md)
+![Architecture](https://raw.githubusercontent.com/cdk-patterns/serverless/master/the-scalable-webhook/img/arch.png)
+
 ## Pattern Usage
 
 All Patterns (unless otherwise stated in their readme) should support the same commands so you can just run:

package-lock.json

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

the-scalable-webhook/.gitignore

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

the-scalable-webhook/.npmignore

@@ -0,0 +1,29 @@
+# The Scalable Webhook
+
+This is an example CDK stack to deploy The Scalable Webhook stack described by Jeremy Daly here - https://www.jeremydaly.com/serverless-microservice-patterns-for-aws/#scalablewebhook
+
+You would use this pattern when you have a non serverless resource like an RDS DB in direct contact with a serverless resource like a lambda. You need to make
+sure that your serverless resource doesn't scale up to an amount that it DOS attacks your non serverless resource.
+
+This is done by putting a queue between them and having a lambda with a throttled concurrency policy pull items off the queue and communicate with your 
+serverless resource at a rate it can handle.
+
+![Architecture](https://raw.githubusercontent.com/cdk-patterns/serverless/master/the-scalable-webhook/img/arch.png)
+
+<strong>NOTE:</strong> For this pattern in the cdk deployable construct I have swapped RDS for DynamoDB. <br /><br />Why? Because it is significantly cheaper/faster for developers to deploy and maintain, I also don't think we lose the essence of the pattern with this swap given we still do the pub/sub deduplication via SQS/Lambda and throttle the subscription lambda. RDS also introduces extra complexity in that it needs to be deployed in a VPC. I am slightly worried developers would get distracted by the extra RDS logic when the main point is the pattern. A real life implementation of this pattern could use RDS MySQL or it could be a call to an on-prem mainframe, the main purpose of the pattern is the throttling to not overload the scale-limited resource.
+
+## How to test pattern
+
+When you deploy this you will have an API Gateway where any url is routed through to the publish lambda. If you modify the url from / to say /hello this url will be sent as a message via sqs to a lambda
+which will insert "hello from /hello" into dynamodb as a message. You can track the progress of your message at every stage through cloudwatch as logs are printed, you can view the contents of
+dynamo in the console and the contents of sqs in the console. You should also notice that SQS can include duplicate messages but in those instances you don't get two identical records in DynamoDB as 
+we used an id we generated in the message as the key
+
+## Useful commands
+
+ * `npm run build`   compile typescript to js
+ * `npm run watch`   watch for changes and compile
+ * `npm run test`    perform the jest unit tests
+ * `npm run deploy`      deploy this stack to your default AWS account/region
+ * `cdk diff`        compare deployed stack with current state
+ * `cdk synth`       emits the synthesized CloudFormation template

the-scalable-webhook/README.md

@@ -0,0 +1,7 @@
+#!/usr/bin/env node
+import 'source-map-support/register';
+import * as cdk from '@aws-cdk/core';
+import { TheScalableWebhookStack } from '../lib/the-scalable-webhook-stack';
+
+const app = new cdk.App();
+new TheScalableWebhookStack(app, 'TheScalableWebhookStack');

the-scalable-webhook/bin/the-scalable-webhook.ts

@@ -0,0 +1,4 @@
+{
+  "@aws-cdk/core:enableStackNameDuplicates": "true",
+  "aws-cdk:enableDiffNoFail": "true"
+}

the-scalable-webhook/cdk.context.json

@@ -0,0 +1,3 @@
+{
+  "app": "npx ts-node bin/the-scalable-webhook.ts"
+}

the-scalable-webhook/cdk.json

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

the-scalable-webhook/img/arch.png

@@ -0,0 +1,46 @@
+var AWS = require('aws-sdk');
+
+exports.handler = async function(event:any) {
+  console.log("request:", JSON.stringify(event, undefined, 2));
+
+  // Create an SQS service object
+  var sqs = new AWS.SQS({apiVersion: '2012-11-05'});
+
+  var params = {
+    DelaySeconds: 10,
+    MessageAttributes: {
+      MessageDeduplicationId: {
+        DataType: "String",
+        StringValue: event.path + new Date().getTime()
+      }
+    },
+    MessageBody: "hello from "+event.path,
+    QueueUrl: process.env.queueURL,
+  };
+
+  let response;
+
+  await sqs.sendMessage(params, function(err:any, data:any) {
+    if (err) {
+      console.log("Error", err);
+      response = sendRes(500, err)
+    } else {
+      console.log("Success", data.MessageId);
+      response = sendRes(200, 'You have added a message to the queue! Message ID is '+data.MessageId)
+    }
+  }).promise();
+
+  // return response back to upstream caller
+  return response;
+};
+
+let sendRes = (status:number, body:string) => {
+  var response = {
+    statusCode: status,
+    headers: {
+      "Content-Type": "text/html"
+    },
+    body: body
+  };
+  return response;
+};

the-scalable-webhook/jest.config.js

@@ -0,0 +1,33 @@
+const { DynamoDB } = require('aws-sdk');
+
+exports.handler = async function(event:any) {
+  console.log("request:", JSON.stringify(event, undefined, 2));
+
+  let records: any[] = event.Records;
+  // create AWS SDK clients
+  const dynamo = new DynamoDB();
+  
+  for(let index in records) {
+    let payload = records[index].body;
+    let id = records[index].messageAttributes.MessageDeduplicationId.stringValue
+    console.log('received message ' + payload);
+    
+    
+    var params = {
+      TableName: process.env.tableName,
+      Item: {
+        'id' : {S: id},
+        'message' : {S: payload}
+      }
+    };
+    
+    // Call DynamoDB to add the item to the table
+    await dynamo.putItem(params, function(err:any, data:any) {
+      if (err) {
+        console.log("Error", err);
+      } else {
+        console.log("Success", data);
+      }
+    }).promise();
+  }
+};

the-scalable-webhook/lambdas/publish/lambda.ts

@@ -0,0 +1,72 @@
+import * as cdk from '@aws-cdk/core';
+import lambda = require('@aws-cdk/aws-lambda');
+import apigw = require('@aws-cdk/aws-apigateway');
+import sqs = require('@aws-cdk/aws-sqs');
+import { SqsEventSource } from '@aws-cdk/aws-lambda-event-sources';
+import dynamodb = require('@aws-cdk/aws-dynamodb');
+
+export class TheScalableWebhookStack extends cdk.Stack {
+  constructor(scope: cdk.Construct, id: string, props?: cdk.StackProps) {
+    super(scope, id, props);
+
+    /**
+     * Dynamo Setup
+     * This is standing in for what is RDS on the diagram due to simpler/cheaper setup
+     */
+    const table = new dynamodb.Table(this, 'Messages', {
+      partitionKey: { name: 'id', type: dynamodb.AttributeType.STRING } //the key being id means we squash duplicate sqs messages
+    });
+
+    /**
+     * Queue Setup
+     * SQS creation
+     */
+    const queue = new sqs.Queue(this, 'RDSPublishQueue', {
+      visibilityTimeout: cdk.Duration.seconds(300)
+    });
+
+    /**
+     * Lambdas
+     * Both publisher and subscriber from pattern
+     */
+
+    // defines an AWS Lambda resource to publish to our queue
+    const sqsPublishLambda = new lambda.Function(this, 'SQSPublishLambdaHandler', {
+      runtime: lambda.Runtime.NODEJS_12_X,      // execution environment
+      code: lambda.Code.asset('lambdas/publish'),  // code loaded from the "lambdas/publish" directory
+      handler: 'lambda.handler',                // file is "lambda", function is "handler"
+      environment: {
+        queueURL: queue.queueUrl
+      }
+    });
+    
+    queue.grantSendMessages(sqsPublishLambda);
+
+    // defines an AWS Lambda resource to pull from our queue
+    const sqsSubscribeLambda = new lambda.Function(this, 'SQSSubscribeLambdaHandler', {
+      runtime: lambda.Runtime.NODEJS_12_X,      // execution environment
+      code: lambda.Code.asset('lambdas/subscribe'),  // code loaded from the "lambdas/subscribe" directory
+      handler: 'lambda.handler',                // file is "lambda", function is "handler"
+      reservedConcurrentExecutions: 2, // throttle lambda to 2 concurrent invocations
+      environment: {
+        queueURL: queue.queueUrl,
+        tableName: table.tableName
+      },
+    });
+    queue.grantConsumeMessages(sqsSubscribeLambda);
+    sqsSubscribeLambda.addEventSource(new SqsEventSource(queue, {}));
+    table.grantReadWriteData(sqsSubscribeLambda);
+
+  
+    /**
+     * API Gateway Proxy
+     * Used to expose the webhook through a URL
+     */
+
+    // defines an API Gateway REST API resource backed by our "sqsPublishLambda" function.
+    new apigw.LambdaRestApi(this, 'Endpoint', {
+      handler: sqsPublishLambda
+    });
+    
+  }
+}