Support S3 as a stream destinations.onFailure type

docs/events/streams.md

@@ -198,9 +198,11 @@ This configuration sets up the onFailure location for events to be sent to once
 
 [Related AWS documentation](https://docs.aws.amazon.com/AWSCloudFormation/latest/UserGuide/aws-resource-lambda-eventsourcemapping.html#cfn-lambda-eventsourcemapping-destinationconfig)
 
-The ARN for the SNS or SQS can be specified as a string, the reference to the ARN of a resource by logical ID, or the import of an ARN that was exported by a different service or CloudFormation stack.
+The destination can be an SNS topic, an SQS queue or an S3 bucket. Its ARN can be specified as a string, the reference to the ARN of a resource by logical ID, or the import of an ARN that was exported by a different service or CloudFormation stack.
 
-**Note:** The `destinationConfig` will hook up your existing SNS or SQS resources. osls won't create a new SNS or SQS for you.
+**Note:** The `destinationConfig` will hook up your existing SNS, SQS or S3 resources. osls won't create a new SNS, SQS or S3 bucket for you.
+
+For an S3 bucket, osls grants the function `s3:ListBucket` on the bucket and `s3:PutObject` on its objects.
 
 ```yaml
 functions:
@@ -254,6 +256,18 @@ functions:
                     - Ref: AWS::AccountId
                     - mySnsTopic
               type: sns
+
+  preprocess4:
+    handler: handler.preprocess
+    events:
+      - stream:
+          arn: arn:aws:kinesis:region:XXXXXX:stream/foo
+          batchSize: 100
+          maximumRetryAttempts: 10
+          startingPosition: LATEST
+          enabled: false
+          destinations:
+            onFailure: arn:aws:s3:::my-failed-batches-bucket
 ```
 
 ## Setting the ParallelizationFactor

lib/plugins/aws/package/compile/events/stream.js

@@ -64,7 +64,7 @@ class AwsCompileStreamEvents {
                       properties: {
                         // arn constraints are listed in oneOf property of this schema
                         arn: {},
-                        type: { enum: ['sns', 'sqs'] },
+                        type: { enum: ['sns', 'sqs', 's3'] },
                       },
                       additionalProperties: false,
                       anyOf: [
@@ -161,6 +161,11 @@ class AwsCompileStreamEvents {
           Action: ['sqs:ListQueues', 'sqs:SendMessage'],
           Resource: [],
         };
+        const onFailureS3Statement = {
+          Effect: 'Allow',
+          Action: ['s3:PutObject', 's3:ListBucket'],
+          Resource: [],
+        };
 
         functionObj.events.forEach((event) => {
           if (event.stream) {
@@ -284,6 +289,14 @@ class AwsCompileStreamEvents {
               // add on failure destination ARNs to PolicyDocument statements
               if (destinationType === 'sns') {
                 onFailureSnsStatement.Resource.push(OnFailureDestinationArn);
+              } else if (destinationType === 's3') {
+                // s3:ListBucket applies to the bucket, s3:PutObject to the objects in it
+                onFailureS3Statement.Resource.push(
+                  OnFailureDestinationArn,
+                  typeof OnFailureDestinationArn === 'string'
+                    ? `${OnFailureDestinationArn}/*`
+                    : { 'Fn::Join': ['', [OnFailureDestinationArn, '/*']] }
+                );
               } else {
                 onFailureSqsStatement.Resource.push(OnFailureDestinationArn);
               }
@@ -388,6 +401,9 @@ class AwsCompileStreamEvents {
           if (onFailureSqsStatement.Resource.length) {
             statement.push(onFailureSqsStatement);
           }
+          if (onFailureS3Statement.Resource.length) {
+            statement.push(onFailureS3Statement);
+          }
         }
       }
     });

test/unit/lib/plugins/aws/package/compile/events/stream.test.js

@@ -23,6 +23,8 @@ const foreignSqsImportArn = { 'Fn::ImportValue': 'ForeignSQS' };
 const someDdbTableStreamRef = { Ref: 'SomeDdbTableStreamArn' };
 const foreignKinesisStreamRef = { Ref: 'ForeignKinesisStreamArn' };
 const someSnsRef = { Ref: 'SomeSNSArn' };
+const s3DestinationArn = 'arn:aws:s3:::failed-batches';
+const s3GetAttArn = { 'Fn::GetAtt': ['FailedBatchesBucket', 'Arn'] };
 const foreignSqsRef = { Ref: 'ForeignSQSArn' };
 
 function createJoinedKinesisArn() {
@@ -470,6 +472,25 @@ describe('test/unit/lib/plugins/aws/package/compile/events/stream.test.js', () =
                     },
                   },
                 },
+                {
+                  stream: {
+                    arn: 'arn:aws:dynamodb:region:account:table/s3string/stream/1',
+                    destinations: {
+                      onFailure: s3DestinationArn,
+                    },
+                  },
+                },
+                {
+                  stream: {
+                    arn: 'arn:aws:dynamodb:region:account:table/s3getatt/stream/1',
+                    destinations: {
+                      onFailure: {
+                        arn: s3GetAttArn,
+                        type: 's3',
+                      },
+                    },
+                  },
+                },
               ],
             },
           },
@@ -644,6 +665,17 @@ describe('test/unit/lib/plugins/aws/package/compile/events/stream.test.js', () =
       ).to.equal(kinesisOptionsDestinationArn);
     });
 
+    it('should support S3 for `destinations.onFailure`', () => {
+      expect(
+        getStreamResource('destinationVariants', 'dynamodb', 's3string').Properties
+          .DestinationConfig.OnFailure.Destination
+      ).to.equal(s3DestinationArn);
+      expect(
+        getStreamResource('destinationVariants', 'dynamodb', 's3getatt').Properties
+          .DestinationConfig.OnFailure.Destination
+      ).to.deep.equal(s3GetAttArn);
+    });
+
     it('should support Fn::GetAtt for `destinations.onFailure`', () => {
       expect(
         getStreamResource('destinationVariants', 'dynamodb', 'foo').Properties.DestinationConfig
@@ -710,7 +742,7 @@ describe('test/unit/lib/plugins/aws/package/compile/events/stream.test.js', () =
       );
       const streamStatements = getIamStatements().filter(({ Action }) => {
         const actions = Array.isArray(Action) ? Action : [Action];
-        return actions.some((action) => /^(dynamodb|kinesis|sns|sqs):/.test(action));
+        return actions.some((action) => /^(dynamodb|kinesis|sns|sqs|s3):/.test(action));
       });
       const expectedStreamStatements = [
         {
@@ -866,6 +898,8 @@ describe('test/unit/lib/plugins/aws/package/compile/events/stream.test.js', () =
             'arn:aws:dynamodb:region:account:table/bar/stream/1',
             'arn:aws:dynamodb:region:account:table/buzz/stream/1',
             'arn:aws:dynamodb:region:account:table/fizz/stream/1',
+            'arn:aws:dynamodb:region:account:table/s3string/stream/1',
+            'arn:aws:dynamodb:region:account:table/s3getatt/stream/1',
           ],
         },
         {
@@ -878,6 +912,16 @@ describe('test/unit/lib/plugins/aws/package/compile/events/stream.test.js', () =
           Action: ['sqs:ListQueues', 'sqs:SendMessage'],
           Resource: [createJoinedSqsArn(), foreignSqsRef],
         },
+        {
+          Effect: 'Allow',
+          Action: ['s3:PutObject', 's3:ListBucket'],
+          Resource: [
+            s3DestinationArn,
+            `${s3DestinationArn}/*`,
+            s3GetAttArn,
+            { 'Fn::Join': ['', [s3GetAttArn, '/*']] },
+          ],
+        },
       ];
 
       expect(streamStatements).to.have.length(expectedStreamStatements.length);