feat: event source filtering (#2241)

* feat: add support for event filtering (DDB, Kinesis, SQS) (#90)

* Add support for Event Filtering (DDB, Kinesis, SQS)

* PR feedback

Rename test file.
Move to a variable the list of event types that have event filtering.

* chore: change a test case that CFN doesn't support (#98)

Co-authored-by: Renato Valenzuela <37676028+valerena@users.noreply.github.com>

samtranslator/model/eventsources/pull.py

@@ -21,6 +21,9 @@ class PullEventSource(ResourceMacro):
     :cvar str policy_arn: The ARN of the AWS managed role policy corresponding to this pull event source
     """
 
+    # Event types that support `FilterCriteria`, stored as a list to keep the alphabetical order
+    RESOURCE_TYPES_WITH_EVENT_FILTERING = ["DynamoDB", "Kinesis", "SQS"]
+
     resource_type = None
     requires_stream_queue_broker = True
     property_types = {
@@ -43,6 +46,7 @@ class PullEventSource(ResourceMacro):
         "TumblingWindowInSeconds": PropertyType(False, is_type(int)),
         "FunctionResponseTypes": PropertyType(False, is_type(list)),
         "KafkaBootstrapServers": PropertyType(False, is_type(list)),
+        "FilterCriteria": PropertyType(False, is_type(dict)),
     }
 
     def get_policy_arn(self):
@@ -102,6 +106,8 @@ def to_cloudformation(self, **kwargs):
         lambda_eventsourcemapping.SourceAccessConfigurations = self.SourceAccessConfigurations
         lambda_eventsourcemapping.TumblingWindowInSeconds = self.TumblingWindowInSeconds
         lambda_eventsourcemapping.FunctionResponseTypes = self.FunctionResponseTypes
+        lambda_eventsourcemapping.FilterCriteria = self.FilterCriteria
+        self._validate_filter_criteria()
 
         if self.KafkaBootstrapServers:
             lambda_eventsourcemapping.SelfManagedEventSource = {
@@ -169,6 +175,20 @@ def _link_policy(self, role, destination_config_policy=None):
                 if not destination_config_policy.get("PolicyDocument") in [d["PolicyDocument"] for d in role.Policies]:
                     role.Policies.append(destination_config_policy)
 
+    def _validate_filter_criteria(self):
+        if not self.FilterCriteria or is_intrinsic(self.FilterCriteria):
+            return
+        if self.resource_type not in self.RESOURCE_TYPES_WITH_EVENT_FILTERING:
+            raise InvalidEventException(
+                self.relative_id,
+                "FilterCriteria is only available for {} events.".format(
+                    ", ".join(self.RESOURCE_TYPES_WITH_EVENT_FILTERING)
+                ),
+            )
+        # FilterCriteria is either empty or only has "Filters"
+        if list(self.FilterCriteria.keys()) not in [[], ["Filters"]]:
+            raise InvalidEventException(self.relative_id, "FilterCriteria field has a wrong format")
+
 
 class Kinesis(PullEventSource):
     """Kinesis event source."""

samtranslator/model/lambda_.py

@@ -79,6 +79,7 @@ class LambdaEventSourceMapping(Resource):
         "TumblingWindowInSeconds": PropertyType(False, is_type(int)),
         "FunctionResponseTypes": PropertyType(False, is_type(list)),
         "SelfManagedEventSource": PropertyType(False, is_type(dict)),
+        "FilterCriteria": PropertyType(False, is_type(dict)),
     }
 
     runtime_attrs = {"name": lambda self: ref(self.logical_id)}

tests/translator/input/error_event_filtering.yaml

@@ -0,0 +1,56 @@
+Resources:
+  WrongFilterName:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: s3://sam-demo-bucket/filtered_events.zip
+      Handler: index.handler
+      Runtime: nodejs16.x
+      Events:
+        DynamoDBStreamEvent:
+          Type: DynamoDB
+          Properties:
+            Stream: !GetAtt DynamoDBTable.StreamArn
+            StartingPosition: TRIM_HORIZON
+            FilterCriteria:
+              FiltersToUse:
+                - Pattern: '{"name": "value"}'
+
+  NotSupportedPullEvent:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: s3://sam-demo-bucket/filtered_events.zip
+      Handler: index.handler
+      Runtime: nodejs16.x
+      Events:
+        KafkaEvent:
+          Type: MSK
+          Properties:
+            StartingPosition: LATEST
+            Stream: arn:aws:kafka:us-east-1:012345678012:cluster/clusterName/abcdefab-1234-abcd-5678-cdef0123ab01-2
+            Topics: 
+              - MyTopic
+            FilterCriteria:
+              Filters:
+                - Pattern: '{"name": "value"}'
+
+
+  NotSupportedPushEvent:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: s3://sam-demo-bucket/filtered_events.zip
+      Handler: index.handler
+      Runtime: nodejs16.x
+      Events:
+        SNSEvent:
+          Type: SNS
+          Properties:
+            Topic: !GetAtt MySnsTopic.Arn
+            FilterCriteria:
+              Filters:
+                - Pattern: '{"name": "value"}'
+
+  DynamoDBTable:
+    Type: AWS::DynamoDB::Table
+
+  MySnsTopic:
+    Type: AWS::SNS::Topic

tests/translator/input/function_with_event_filtering.yaml

@@ -0,0 +1,49 @@
+Resources:
+  FilteredEventsFunction:
+    Type: AWS::Serverless::Function
+    Properties:
+      CodeUri: s3://sam-demo-bucket/filtered_events.zip
+      Handler: index.handler
+      Runtime: nodejs16.x
+      Events:
+        KinesisStream:
+          Type: Kinesis
+          Properties:
+            Stream: !GetAtt KinesisStream.Arn
+            StartingPosition: LATEST
+            FilterCriteria:
+              Filters:
+                - Pattern: '{"name": "value"}'
+                - Pattern: '{"name2": "value2"}'
+        DynamoDBStreamEvent:
+          Type: DynamoDB
+          Properties:
+            Stream: !GetAtt DynamoDBTable.StreamArn
+            StartingPosition: TRIM_HORIZON
+            FilterCriteria:
+              Filters:
+                - Pattern: '{
+                      "dynamodb": {
+                        "NewImage": {
+                          "value": { "S": ["test"] }
+                        }
+                      }
+                    }'
+        MySqsQueue:
+          Type: SQS
+          Properties:
+            Queue: !GetAtt MySqsQueue.Arn
+            FilterCriteria:
+              Filters:
+                - Pattern: '{"name": "value"}'
+
+  KinesisStream:
+    Type: AWS::Kinesis::Stream
+    Properties:
+      ShardCount: 1
+
+  DynamoDBTable:
+    Type: AWS::DynamoDB::Table
+
+  MySqsQueue:
+    Type: AWS::SQS::Queue

tests/translator/input/intrinsic_functions.yaml

@@ -108,6 +108,12 @@ Resources:
               Fn::GetAtt: [MyTable, "StreamArn"]
             BatchSize: 200
             StartingPosition: LATEST
+            FilterCriteria:
+              Fn::Select:
+                - "1"
+                -
+                  - {}
+                  - { "Filters": { "Pattern": "{\"value\": \"b\"}" }}
 
   MyTable:
     Type: AWS::DynamoDB::Table

tests/translator/output/aws-cn/function_with_event_filtering.json

@@ -0,0 +1,140 @@
+{
+  "Resources": {
+    "KinesisStream": {
+      "Type": "AWS::Kinesis::Stream",
+      "Properties": {
+        "ShardCount": 1
+      }
+    },
+    "DynamoDBTable": {
+      "Type": "AWS::DynamoDB::Table"
+    },
+    "MySqsQueue": {
+      "Type": "AWS::SQS::Queue"
+    },
+    "FilteredEventsFunction": {
+      "Type": "AWS::Lambda::Function",
+      "Properties": {
+        "Code": {
+          "S3Bucket": "sam-demo-bucket",
+          "S3Key": "filtered_events.zip"
+        },
+        "Handler": "index.handler",
+        "Role": {
+          "Fn::GetAtt": [
+            "FilteredEventsFunctionRole",
+            "Arn"
+          ]
+        },
+        "Runtime": "nodejs16.x",
+        "Tags": [
+          {
+            "Key": "lambda:createdBy",
+            "Value": "SAM"
+          }
+        ]
+      }
+    },
+    "FilteredEventsFunctionRole": {
+      "Type": "AWS::IAM::Role",
+      "Properties": {
+        "AssumeRolePolicyDocument": {
+          "Version": "2012-10-17",
+          "Statement": [
+            {
+              "Action": [
+                "sts:AssumeRole"
+              ],
+              "Effect": "Allow",
+              "Principal": {
+                "Service": [
+                  "lambda.amazonaws.com"
+                ]
+              }
+            }
+          ]
+        },
+        "ManagedPolicyArns": [
+          "arn:aws-cn:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole",
+          "arn:aws-cn:iam::aws:policy/service-role/AWSLambdaDynamoDBExecutionRole",
+          "arn:aws-cn:iam::aws:policy/service-role/AWSLambdaKinesisExecutionRole",
+          "arn:aws-cn:iam::aws:policy/service-role/AWSLambdaSQSQueueExecutionRole"
+        ],
+        "Tags": [
+          {
+            "Key": "lambda:createdBy",
+            "Value": "SAM"
+          }
+        ]
+      }
+    },
+    "FilteredEventsFunctionDynamoDBStreamEvent": {
+      "Type": "AWS::Lambda::EventSourceMapping",
+      "Properties": {
+        "EventSourceArn": {
+          "Fn::GetAtt": [
+            "DynamoDBTable",
+            "StreamArn"
+          ]
+        },
+        "FunctionName": {
+          "Ref": "FilteredEventsFunction"
+        },
+        "StartingPosition": "TRIM_HORIZON",
+        "FilterCriteria": {
+          "Filters": [
+            {
+              "Pattern": "{ \"dynamodb\": { \"NewImage\": { \"value\": { \"S\": [\"test\"] } } } }"
+            }
+          ]
+        }
+      }
+    },
+    "FilteredEventsFunctionKinesisStream": {
+      "Type": "AWS::Lambda::EventSourceMapping",
+      "Properties": {
+        "EventSourceArn": {
+          "Fn::GetAtt": [
+            "KinesisStream",
+            "Arn"
+          ]
+        },
+        "FunctionName": {
+          "Ref": "FilteredEventsFunction"
+        },
+        "StartingPosition": "LATEST",
+        "FilterCriteria": {
+          "Filters": [
+            {
+              "Pattern": "{\"name\": \"value\"}"
+            },
+            {
+              "Pattern": "{\"name2\": \"value2\"}"
+            }
+          ]
+        }
+      }
+    },
+    "FilteredEventsFunctionMySqsQueue": {
+      "Type": "AWS::Lambda::EventSourceMapping",
+      "Properties": {
+        "EventSourceArn": {
+          "Fn::GetAtt": [
+            "MySqsQueue",
+            "Arn"
+          ]
+        },
+        "FunctionName": {
+          "Ref": "FilteredEventsFunction"
+        },
+        "FilterCriteria": {
+          "Filters": [
+            {
+              "Pattern": "{\"name\": \"value\"}"
+            }
+          ]
+        }
+      }
+    }
+  }
+}

tests/translator/output/aws-cn/intrinsic_functions.json

@@ -605,6 +605,19 @@
             "MyTable",
             "StreamArn"
           ]
+        },
+        "FilterCriteria": {
+          "Fn::Select": [
+            "1",
+            [
+              {},
+              {
+                "Filters": {
+                  "Pattern": "{\"value\": \"b\"}"
+                }
+              }
+            ]
+          ]
         }
       }
     },