diff --git a/docs/providers/aws/events/streams.md b/docs/providers/aws/events/streams.md index e4dada34fa6..b0c615e0d87 100644 --- a/docs/providers/aws/events/streams.md +++ b/docs/providers/aws/events/streams.md @@ -329,3 +329,22 @@ functions: arn: arn:aws:dynamodb:region:XXXXXX:table/foo/stream/1970-01-01T00:00:00.000 tumblingWindowInSeconds: 30 ``` + +## Setting filter patterns + +This configuration allows customers to filter event before lambda invocation. It accepts up to 5 filter criterion by default and up to 10 with quota extension. If one event matches at least 1 pattern, lambda will process it. + +For more details and examples of filter patterns, please see the [AWS event filtering documentation](https://docs.aws.amazon.com/lambda/latest/dg/invocation-eventfiltering.html) + +Note: Serverless only sets this property if you explicitly add it to the stream configuration (see an example below). The following example will only process inserted items in the DynamoDB table (it will skip removed and modified items). + +```yml +functions: + handleInsertedDynamoDBItem: + handler: handler.preprocess + events: + - stream: + arn: arn:aws:dynamodb:region:XXXXXX:table/foo/stream/1970-01-01T00:00:00.000 + filterPatterns: + - eventName: [INSERT] +``` diff --git a/docs/providers/aws/guide/serverless.yml.md b/docs/providers/aws/guide/serverless.yml.md index a4ba3714b1b..6bd1506282c 100644 --- a/docs/providers/aws/guide/serverless.yml.md +++ b/docs/providers/aws/guide/serverless.yml.md @@ -458,6 +458,8 @@ functions: startingPosition: LATEST enabled: true functionResponseType: ReportBatchItemFailures + filterPatterns: + - partitionKey: [ 1 ] - msk: arn: arn:aws:kafka:us-east-1:111111111111:cluster/ClusterName/a1a1a1a1a1a1a1a1a # ARN of MSK Cluster topic: kafkaTopic # name of Kafka topic to consume from diff --git a/lib/plugins/aws/package/compile/events/stream.js b/lib/plugins/aws/package/compile/events/stream.js index 03264140bcb..21377addd94 100644 --- a/lib/plugins/aws/package/compile/events/stream.js +++ b/lib/plugins/aws/package/compile/events/stream.js @@ -69,6 +69,12 @@ class AwsCompileStreamEvents { required: ['onFailure'], }, tumblingWindowInSeconds: { type: 'integer', minimum: 0, maximum: 900 }, + filterPatterns: { + type: 'array', + minItems: 1, + maxItems: 10, + items: { type: 'object' }, + }, }, additionalProperties: false, anyOf: [ @@ -273,6 +279,14 @@ class AwsCompileStreamEvents { }; } + if (event.stream.filterPatterns) { + streamResource.Properties.FilterCriteria = { + Filters: event.stream.filterPatterns.map((pattern) => ({ + Pattern: JSON.stringify(pattern), + })), + }; + } + const newStreamObject = { [streamLogicalId]: streamResource, }; 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 cc894e2df08..6b640a4122d 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 @@ -1724,4 +1724,43 @@ describe('AwsCompileStreamEvents #2', () => { ); }); }); + describe('with filterPatterns', () => { + let eventSourceMappingResource; + + before(async () => { + const { awsNaming, cfTemplate } = await runServerless({ + fixture: 'function', + configExt: { + functions: { + basic: { + events: [ + { + stream: { + arn: 'arn:aws:dynamodb:region:account:table/foo/stream/1', + filterPatterns: [{ eventName: ['INSERT'] }, { eventName: ['MODIFY'] }], + }, + }, + ], + }, + }, + }, + command: 'package', + }); + const streamLogicalId = awsNaming.getStreamLogicalId('basic', 'dynamodb', 'foo'); + eventSourceMappingResource = cfTemplate.Resources[streamLogicalId]; + }); + + it('should wrap patterns within FilterCriteria property', () => { + expect(eventSourceMappingResource.Properties.FilterCriteria).to.deep.equal({ + Filters: [ + { + Pattern: JSON.stringify({ eventName: ['INSERT'] }), + }, + { + Pattern: JSON.stringify({ eventName: ['MODIFY'] }), + }, + ], + }); + }); + }); });