diff --git a/docs/events/streams.md b/docs/events/streams.md index 571645473e..5d27943d2b 100644 --- a/docs/events/streams.md +++ b/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 diff --git a/lib/plugins/aws/package/compile/events/stream.js b/lib/plugins/aws/package/compile/events/stream.js index b9f69791dc..816373782f 100644 --- a/lib/plugins/aws/package/compile/events/stream.js +++ b/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); + } } } }); diff --git a/test/unit/lib/plugins/aws/package/compile/events/stream.test.js b/test/unit/lib/plugins/aws/package/compile/events/stream.test.js index fb21840dcb..ce2431e18e 100644 --- a/test/unit/lib/plugins/aws/package/compile/events/stream.test.js +++ b/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);