feat(lambda-destinations): support Lambda async S3 destination

packages/@aws-cdk-testing/framework-integ/test/aws-lambda-destinations/test/integ.destinations.ts

@@ -1,4 +1,5 @@
 import * as lambda from 'aws-cdk-lib/aws-lambda';
+import * as s3 from 'aws-cdk-lib/aws-s3';
 import * as sns from 'aws-cdk-lib/aws-sns';
 import * as sqs from 'aws-cdk-lib/aws-sqs';
 import { App, Duration, Stack, StackProps } from 'aws-cdk-lib';
@@ -54,6 +55,22 @@ class TestStack extends Stack {
       onSuccess: new destinations.LambdaDestination(onSuccessLambda),
     });
 
+    const successBucket = new s3.Bucket(this, 'OnSuccessBucket');
+    const failureBucket = new s3.Bucket(this, 'OnFailureBucket');
+
+    new lambda.Function(this, 'S3', {
+      runtime: STANDARD_NODEJS_RUNTIME,
+      handler: 'index.handler',
+      code: lambda.Code.fromInline(`exports.handler = async (event) => {
+        if (event.status === 'OK') return 'success';
+        throw new Error('failure');
+      };`),
+      onFailure: new destinations.S3Destination(failureBucket),
+      onSuccess: new destinations.S3Destination(successBucket),
+      maxEventAge: Duration.hours(4),
+      retryAttempts: 2,
+    });
+
     const version = this.fn.addVersion('MySpecialVersion');
 
     new lambda.Alias(this, 'MySpecialAlias', {

packages/aws-cdk-lib/aws-lambda-destinations/README.md

@@ -12,6 +12,7 @@ The following destinations are supported
 * SQS queue - Only standard SQS queues are supported for failure destinations, FIFO queues are not supported.
 * SNS topic
 * EventBridge event bus
+* S3 bucket
 
 Example with a SNS topic for successful invocations:
 
@@ -116,6 +117,8 @@ is either 'Lambda Function Invocation Result - Success' or 'Lambda Function Invo
 depending on whether the lambda function invocation succeeded or failed. The event field `resource`
 contains the function and destination ARNs. See [AWS Events](https://docs.aws.amazon.com/eventbridge/latest/userguide/aws-events.html)
 for the different event fields.
+* For S3 (`S3Destination`), the invocation record json is stored as a `File` in the destination bucket. The path of a destination 
+payload file in the configured bucket is `aws/lambda/async/<function-name>/YYYY/MM/DD/YYYY-MM-DDTHH.MM.SS-<Random UUID>`. 
 
 ### Auto-extract response payload with lambda destination
 

packages/aws-cdk-lib/aws-lambda-destinations/lib/index.ts

@@ -1,4 +1,5 @@
 export * from './event-bridge';
 export * from './lambda';
+export * from './s3';
 export * from './sns';
 export * from './sqs';

packages/aws-cdk-lib/aws-lambda-destinations/lib/s3.ts

@@ -0,0 +1,24 @@
+import { Construct } from 'constructs';
+import * as lambda from '../../aws-lambda';
+import * as s3 from '../../aws-s3';
+
+/**
+ * Use a S3 bucket as a Lambda destination
+ */
+export class S3Destination implements lambda.IDestination {
+  constructor(private readonly bucket: s3.IBucket) {
+  }
+
+  /**
+   * Returns a destination configuration
+   */
+  public bind(_scope: Construct, fn: lambda.IFunction, _options?: lambda.DestinationOptions): lambda.DestinationConfig {
+    // grant read and putObject permissions
+    this.bucket.grantRead(fn);
+    this.bucket.grantPut(fn);
+
+    return {
+      destination: this.bucket.bucketArn,
+    };
+  }
+}

packages/aws-cdk-lib/aws-lambda-destinations/test/destinations.test.ts

@@ -1,6 +1,7 @@
 import { Template } from '../../assertions';
 import * as events from '../../aws-events';
 import * as lambda from '../../aws-lambda';
+import * as s3 from '../../aws-s3';
 import * as sns from '../../aws-sns';
 import * as sqs from '../../aws-sqs';
 import { Stack } from '../../core';
@@ -357,3 +358,92 @@ test('sqs as destination', () => {
     },
   });
 });
+
+test('s3 as destination', () => {
+  // GIVEN
+  const bucket = new s3.Bucket(stack, 'Bucket');
+
+  // WHEN
+  new lambda.Function(stack, 'Function', {
+    ...lambdaProps,
+    onSuccess: new destinations.S3Destination(bucket),
+  });
+
+  // THEN
+  Template.fromStack(stack).hasResourceProperties('AWS::Lambda::EventInvokeConfig', {
+    DestinationConfig: {
+      OnSuccess: {
+        Destination: {
+          'Fn::GetAtt': [
+            'Bucket83908E77',
+            'Arn',
+          ],
+        },
+      },
+    },
+  });
+
+  Template.fromStack(stack).hasResourceProperties('AWS::IAM::Policy', {
+    PolicyDocument: {
+      Statement: [
+        {
+          Action: [
+            's3:GetObject*',
+            's3:GetBucket*',
+            's3:List*',
+          ],
+          Effect: 'Allow',
+          Resource: [
+            {
+              'Fn::GetAtt': [
+                'Bucket83908E77',
+                'Arn',
+              ],
+            },
+            {
+              'Fn::Join': [
+                '',
+                [
+                  {
+                    'Fn::GetAtt': [
+                      'Bucket83908E77',
+                      'Arn',
+                    ],
+                  },
+                  '/*',
+                ],
+              ],
+            },
+          ],
+        },
+        {
+          Action: [
+            's3:PutObject',
+            's3:PutObjectLegalHold',
+            's3:PutObjectRetention',
+            's3:PutObjectTagging',
+            's3:PutObjectVersionTagging',
+            's3:Abort*',
+          ],
+          Effect: 'Allow',
+          Resource:
+          {
+            'Fn::Join': [
+              '',
+              [
+                {
+                  'Fn::GetAtt': [
+                    'Bucket83908E77',
+                    'Arn',
+                  ],
+                },
+                '/*',
+              ],
+            ],
+          },
+        },
+      ],
+      Version: '2012-10-17',
+    },
+  });
+});