feat: lambda worker - support maxConcurrency (SqsEventSource) (#86)

techfromsage · Apr 3, 2024 · dd53d96 · dd53d96
1 parent bc3a840
commit dd53d96
Show file tree

Hide file tree

Showing 5 changed files with 19 additions and 0 deletions.
diff --git a/examples/container-lambda-worker/lib/container-lambda-worker-stack.ts b/examples/container-lambda-worker/lib/container-lambda-worker-stack.ts
@@ -52,6 +52,7 @@ export class ContainerLambdaWorkerStack extends cdk.Stack {
       {
         name: `${prefix}container-lambda-worker`,
         lambdaProps: {
+          queueMaxConcurrency: 100,
           environment: {
             EXAMPLE_ENV_VAR: "example value",
           },

diff --git a/examples/simple-lambda-worker/lib/simple-lambda-worker-stack.ts b/examples/simple-lambda-worker/lib/simple-lambda-worker-stack.ts
@@ -80,6 +80,7 @@ export class SimpleLambdaWorkerStack extends cdk.Stack {
       {
         name: `${prefix}simple-lambda-worker`,
         lambdaProps: {
+          queueMaxConcurrency: 5,
           environment: {
             EXAMPLE_ENV_VAR: "example value",
             SUCCESS_QUEUE_URL: successQueue.queueUrl,

diff --git a/lib/lambda-worker/lambda-worker-props.ts b/lib/lambda-worker/lambda-worker-props.ts
@@ -45,6 +45,7 @@ export interface LambdaWorkerProps {
     timeout: cdk.Duration;
     vpc?: ec2.IVpc;
     vpcSubnets?: ec2.SubnetSelection;
+    queueMaxConcurrency: number;
   } & Partial<FunctionLambdaProps> &
     Partial<ContainerFromEcrLambdaProps> &
     Partial<ContainerFromImageAssetLambdaProps>;

diff --git a/lib/lambda-worker/lambda-worker.ts b/lib/lambda-worker/lambda-worker.ts
@@ -141,6 +141,7 @@ export class LambdaWorker extends Construct {
       new SqsEventSource(lambdaQueue, {
         enabled: props.lambdaProps.enableQueue ?? true,
         batchSize: 1,
+        maxConcurrency: props.lambdaProps.queueMaxConcurrency,
       }),
     );
     lambdaWorker.addEventSource(

diff --git a/test/infra/lambda-worker/lambda-worker.test.ts b/test/infra/lambda-worker/lambda-worker.test.ts
@@ -26,6 +26,7 @@ describe("LambdaWorker", () => {
         worker = new LambdaWorker(stack, "MyTestLambdaWorker", {
           name: "MyTestLambdaWorker",
           lambdaProps: {
+            queueMaxConcurrency: 5,
             entry: "examples/simple-lambda-worker/src/lambda/simple-worker.js",
             handler: "testWorker",
             memorySize: 2048,
@@ -216,6 +217,7 @@ describe("LambdaWorker", () => {
         new LambdaWorker(stack, "MyTestLambdaWorker", {
           name: "MyTestLambdaWorker",
           lambdaProps: {
+            queueMaxConcurrency: 5,
             entry: "examples/simple-lambda-worker/src/lambda/simple-worker.js",
             handler: "testWorker",
             memorySize: 2048,
@@ -284,6 +286,7 @@ describe("LambdaWorker", () => {
         new LambdaWorker(stack, "MyTestLambdaWorker", {
           name: "MyTestLambdaWorker",
           lambdaProps: {
+            queueMaxConcurrency: 5,
             entry: "examples/simple-lambda-worker/src/lambda/simple-worker.js",
             handler: "testWorker",
             memorySize: 2048,
@@ -340,6 +343,7 @@ describe("LambdaWorker", () => {
         new LambdaWorker(stack, "MyTestLambdaWorker", {
           name: "MyTestLambdaWorker",
           lambdaProps: {
+            queueMaxConcurrency: 5,
             entry: "examples/simple-lambda-worker/src/lambda/simple-worker.js",
             handler: "testWorker",
             memorySize: 2048,
@@ -401,6 +405,7 @@ describe("LambdaWorker", () => {
         new LambdaWorker(stack, "MyTestLambdaWorker", {
           name: "MyTestLambdaWorker",
           lambdaProps: {
+            queueMaxConcurrency: 5,
             entry: "examples/simple-lambda-worker/src/lambda/simple-worker.js",
             handler: "testWorker",
             memorySize: 2048,
@@ -436,6 +441,9 @@ describe("LambdaWorker", () => {
                 "Arn",
               ],
             },
+            ScalingConfig: {
+              MaximumConcurrency: 5,
+            },
           },
         );
       });
@@ -459,6 +467,7 @@ describe("LambdaWorker", () => {
       new LambdaWorker(stack, "MyTestLambdaWorker", {
         name: "MyTestLambdaWorker",
         lambdaProps: {
+          queueMaxConcurrency: 5,
           entry: "examples/simple-lambda-worker/src/lambda/simple-worker.js",
           handler: "testWorker",
           memorySize: 2048,
@@ -528,6 +537,7 @@ describe("LambdaWorker", () => {
         worker = new LambdaWorker(stack, "MyTestLambdaWorker", {
           name: "MyTestLambdaWorker",
           lambdaProps: {
+            queueMaxConcurrency: 5,
             dockerImageTag: "test-lambda-12345",
             ecrRepositoryArn: "arn:aws:ecr:eu-west-1:012345678910:repository",
             ecrRepositoryName: "repository",
@@ -728,6 +738,7 @@ describe("LambdaWorker", () => {
         new LambdaWorker(stack, "MyTestLambdaWorker", {
           name: "MyTestLambdaWorker",
           lambdaProps: {
+            queueMaxConcurrency: 5,
             dockerImageTag: "test-lambda-12345",
             ecrRepositoryArn: "arn:aws:ecr:eu-west-1:012345678910:repository",
             ecrRepositoryName: "repository",
@@ -797,6 +808,7 @@ describe("LambdaWorker", () => {
         });
 
         lambdaProps = {
+          queueMaxConcurrency: 5,
           memorySize: 2048,
           policyStatements: [
             new iam.PolicyStatement({
@@ -901,6 +913,7 @@ describe("LambdaWorker", () => {
         new LambdaWorker(stack, "MyTestLambdaWorker", {
           name: "MyTestLambdaWorker",
           lambdaProps: {
+            queueMaxConcurrency: 5,
             entry: "examples/simple-lambda-worker/src/lambda/simple-worker.js",
             handler: "testWorker",
             memorySize: 2048,
@@ -931,6 +944,7 @@ describe("LambdaWorker", () => {
         new LambdaWorker(stack, "MyTestLambdaWorker", {
           name: "MyTestLambdaWorker",
           lambdaProps: {
+            queueMaxConcurrency: 5,
             entry: "examples/simple-lambda-worker/src/lambda/simple-worker.js",
             handler: "testWorker",
             memorySize: 512,
@@ -1041,6 +1055,7 @@ describe("LambdaWorker", () => {
           new LambdaWorker(stack, "MyTestLambdaWorker", {
             name: "MyTestLambdaWorker",
             lambdaProps: {
+              queueMaxConcurrency: 5,
               entry: config.entry,
               handler: config.handler,
               dockerCommand: config.dockerCommand,