feat: Event filtering for MQ, MSK, SMK (#134) (#265)

Support event filtering for Amazon MQ, MSK and SelfManagedKafka event sources

Co-authored-by: Renato Valenzuela <[email protected]>

Author: hawflau

samtranslator/model/eventsources/pull.py

@@ -21,7 +21,7 @@ class PullEventSource(ResourceMacro):
     """
 
     # Event types that support `FilterCriteria`, stored as a list to keep the alphabetical order
-    RESOURCE_TYPES_WITH_EVENT_FILTERING = ["DynamoDB", "Kinesis", "SQS"]
+    RESOURCE_TYPES_WITH_EVENT_FILTERING = ["DynamoDB", "Kinesis", "MQ", "MSK", "SelfManagedKafka", "SQS"]
 
     # Note(xinhol): `PullEventSource` should have been an abstract class. Disabling the type check for the next
     # line to avoid any potential behavior change.

tests/translator/input/error_event_filtering.yaml

@@ -15,25 +15,6 @@ Resources:
               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:

tests/translator/input/function_with_event_filtering.yaml

@@ -29,13 +29,53 @@ Resources:
                         }
                       }
                     }'
-        MySqsQueue:
+        MySqsEvent:
           Type: SQS
           Properties:
             Queue: !GetAtt MySqsQueue.Arn
             FilterCriteria:
               Filters:
                 - Pattern: '{"name": "value"}'
+        MSKEvent:
+          Type: MSK
+          Properties:
+            StartingPosition: LATEST
+            Stream: arn:aws:kafka:us-west-2:012345678901:cluster/mycluster/6cc0432b-8618-4f44-bccc-e1fbd8fb7c4d-2
+            Topics:
+              - "MyDummyTestTopic"
+            FilterCriteria:
+              Filters:
+                - Pattern: '{"name": "value"}'
+        MyKafkaEvent:
+          Type: SelfManagedKafka
+          Properties:
+            KafkaBootstrapServers:
+              - "abc.xyz.com:9092"
+            Topics:
+              - "Topic1"
+            SourceAccessConfigurations:
+              - Type: SASL_SCRAM_512_AUTH
+                URI: arn:aws:secretsmanager:us-west-2:123456789012:secret:my-path/my-secret-name-1a2b3c
+              - Type: VPC_SUBNET
+                URI: subnet:subnet-12345
+              - Type: VPC_SECURITY_GROUP
+                URI: security_group:sg-67890
+            FilterCriteria:
+              Filters:
+                - Pattern: '{"name": "value"}'
+        MyMQQueue:
+          Type: MQ
+          Properties:
+            Broker: arn:aws:mq:us-east-2:123456789012:broker:MyBroker:b-1234a5b6-78cd-901e-2fgh-3i45j6k178l9
+            Queues:
+              - "Queue1"
+            SourceAccessConfigurations:
+              - Type: BASIC_AUTH
+                URI: arn:aws:secretsmanager:us-west-2:123456789012:secret:my-path/my-secret-name-1a2b3c
+            SecretsManagerKmsKeyId: 1abc23d4-567f-8ab9-cde0-1fab234c5d67
+            FilterCriteria:
+              Filters:
+                - Pattern: '{"name": "value"}'
 
   KinesisStream:
     Type: AWS::Kinesis::Stream

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

@@ -58,8 +58,66 @@
           "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/AWSLambdaMSKExecutionRole",
           "arn:aws-cn:iam::aws:policy/service-role/AWSLambdaSQSQueueExecutionRole"
         ],
+        "Policies": [
+          {
+            "PolicyDocument": {
+              "Statement": [
+                {
+                  "Action": [
+                    "secretsmanager:GetSecretValue"
+                  ],
+                  "Effect": "Allow",
+                  "Resource": "arn:aws:secretsmanager:us-west-2:123456789012:secret:my-path/my-secret-name-1a2b3c"
+                },
+                {
+                  "Action": [
+                    "ec2:CreateNetworkInterface",
+                    "ec2:DescribeNetworkInterfaces",
+                    "ec2:DeleteNetworkInterface",
+                    "ec2:DescribeVpcs",
+                    "ec2:DescribeSubnets",
+                    "ec2:DescribeSecurityGroups"
+                  ],
+                  "Effect": "Allow",
+                  "Resource": "*"
+                }
+              ],
+              "Version": "2012-10-17"
+            },
+            "PolicyName": "SelfManagedKafkaExecutionRolePolicy"
+          },
+          {
+            "PolicyName": "SamAutoGeneratedAMQPolicy",
+            "PolicyDocument": {
+              "Statement": [
+                {
+                  "Action": [
+                    "secretsmanager:GetSecretValue"
+                  ],
+                  "Effect": "Allow",
+                  "Resource": "arn:aws:secretsmanager:us-west-2:123456789012:secret:my-path/my-secret-name-1a2b3c"
+                },
+                {
+                  "Action": [
+                    "mq:DescribeBroker"
+                  ],
+                  "Effect": "Allow",
+                  "Resource": "arn:aws:mq:us-east-2:123456789012:broker:MyBroker:b-1234a5b6-78cd-901e-2fgh-3i45j6k178l9"
+                },
+                {
+                  "Action": "kms:Decrypt",
+                  "Effect": "Allow",
+                  "Resource": {
+                    "Fn::Sub": "arn:${AWS::Partition}:kms:${AWS::Region}:${AWS::AccountId}:key/1abc23d4-567f-8ab9-cde0-1fab234c5d67"
+                  }
+                }
+              ]
+            }
+          }
+        ],
         "Tags": [
           {
             "Key": "lambda:createdBy",
@@ -115,7 +173,7 @@
         }
       }
     },
-    "FilteredEventsFunctionMySqsQueue": {
+    "FilteredEventsFunctionMySqsEvent": {
       "Type": "AWS::Lambda::EventSourceMapping",
       "Properties": {
         "EventSourceArn": {
@@ -135,6 +193,90 @@
           ]
         }
       }
+    },
+    "FilteredEventsFunctionMSKEvent": {
+      "Type": "AWS::Lambda::EventSourceMapping",
+      "Properties": {
+        "EventSourceArn": "arn:aws:kafka:us-west-2:012345678901:cluster/mycluster/6cc0432b-8618-4f44-bccc-e1fbd8fb7c4d-2",
+        "FunctionName": {
+          "Ref": "FilteredEventsFunction"
+        },
+        "StartingPosition": "LATEST",
+        "Topics": [
+          "MyDummyTestTopic"
+        ],
+        "FilterCriteria": {
+          "Filters": [
+            {
+              "Pattern": "{\"name\": \"value\"}"
+            }
+          ]
+        }
+      }
+    },
+    "FilteredEventsFunctionMyKafkaEvent": {
+      "Type": "AWS::Lambda::EventSourceMapping",
+      "Properties": {
+        "FunctionName": {
+          "Ref": "FilteredEventsFunction"
+        },
+        "Topics": [
+          "Topic1"
+        ],
+        "SourceAccessConfigurations": [
+          {
+            "Type": "SASL_SCRAM_512_AUTH",
+            "URI": "arn:aws:secretsmanager:us-west-2:123456789012:secret:my-path/my-secret-name-1a2b3c"
+          },
+          {
+            "Type": "VPC_SUBNET",
+            "URI": "subnet:subnet-12345"
+          },
+          {
+            "Type": "VPC_SECURITY_GROUP",
+            "URI": "security_group:sg-67890"
+          }
+        ],
+        "SelfManagedEventSource": {
+          "Endpoints": {
+            "KafkaBootstrapServers": [
+              "abc.xyz.com:9092"
+            ]
+          }
+        },
+        "FilterCriteria": {
+          "Filters": [
+            {
+              "Pattern": "{\"name\": \"value\"}"
+            }
+          ]
+        }
+      }
+    },
+    "FilteredEventsFunctionMyMQQueue": {
+      "Type": "AWS::Lambda::EventSourceMapping",
+      "Properties": {
+        "EventSourceArn": "arn:aws:mq:us-east-2:123456789012:broker:MyBroker:b-1234a5b6-78cd-901e-2fgh-3i45j6k178l9",
+        "FunctionName": {
+          "Ref": "FilteredEventsFunction"
+        },
+        "Queues": [
+          "Queue1"
+        ],
+        "SourceAccessConfigurations": [
+          {
+            "Type": "BASIC_AUTH",
+            "URI": "arn:aws:secretsmanager:us-west-2:123456789012:secret:my-path/my-secret-name-1a2b3c"
+          }
+        ],
+        "FilterCriteria": {
+          "Filters": [
+            {
+              "Pattern": "{\"name\": \"value\"}"
+            }
+          ]
+        }
+      }
     }
   }
 }