feat: automatically add DependsOn for Lambda to SQS DLQ on ESM

samtranslator/model/__init__.py

@@ -1,7 +1,8 @@
 """ CloudFormation Resource serialization, deserialization, and validation """
+import copy
 import re
 import inspect
-from typing import Any, Callable, Dict
+from typing import Any, Callable, Dict, Tuple, Iterator
 
 from samtranslator.model.exceptions import InvalidResourceException
 from samtranslator.plugins import LifeCycleEvents
@@ -527,9 +528,13 @@ def __init__(self, resources: Dict[str, Any]):
 
         self.resources = resources
 
-    def get_all_resources(self) -> Dict[str, Any]:
-        """Return a dictionary of all resources from the SAM template."""
-        return self.resources
+    def get_resources_by_type(self, resource_type: str) -> Iterator[Tuple[str, Dict[str, Any]]]:
+        """
+        Return generator of (logical_id, properties) tuples of resources with specified type.
+        """
+        for logical_id, resource in self.resources.items():
+            if resource.get("Type") == resource_type:
+                yield logical_id, copy.deepcopy(resource.get("Properties") or {})
 
     def get_resource_by_logical_id(self, input: str) -> Dict[str, Any]:
         """

samtranslator/model/connector/connector.py

@@ -3,6 +3,7 @@
 
 from samtranslator.model import ResourceResolver
 from samtranslator.model.intrinsics import get_logical_id_from_intrinsic, ref, fnGetAtt
+from samtranslator.utils import dictfind
 
 
 # TODO: Switch to dataclass
@@ -54,21 +55,33 @@ def get_event_source_mappings(event_source_id: str, function_id: str, resource_r
     """
     Get logical IDs of `AWS::Lambda::EventSourceMapping`s between resource logical IDs.
     """
-    resources = resource_resolver.get_all_resources()
-    for logical_id, resource in resources.items():
-        if resource.get("Type") == "AWS::Lambda::EventSourceMapping":
-            properties = resource.get("Properties", {})
-            # Not taking intrinsics as input to function as FunctionName could be a number of
-            # formats, which would require parsing it anyway
-            resource_function_id = get_logical_id_from_intrinsic(properties.get("FunctionName"))
-            resource_event_source_id = get_logical_id_from_intrinsic(properties.get("EventSourceArn"))
-            if (
-                resource_function_id
-                and resource_event_source_id
-                and function_id == resource_function_id
-                and event_source_id == resource_event_source_id
-            ):
-                yield logical_id
+    resources = resource_resolver.get_resources_by_type("AWS::Lambda::EventSourceMapping")
+    for logical_id, properties in resources:
+        # Not taking intrinsics as input to function as FunctionName could be a number of
+        # formats, which would require parsing it anyway
+        resource_function_id = get_logical_id_from_intrinsic(properties.get("FunctionName"))
+        resource_event_source_id = get_logical_id_from_intrinsic(properties.get("EventSourceArn"))
+        if (
+            resource_function_id
+            and resource_event_source_id
+            and function_id == resource_function_id
+            and event_source_id == resource_event_source_id
+        ):
+            yield logical_id
+
+
+def get_event_source_mapping_dlqs(function_id: str, dlq_id: str, resource_resolver: ResourceResolver):
+    resources = resource_resolver.get_resources_by_type("AWS::Lambda::EventSourceMapping")
+    for logical_id, properties in resources:
+        resource_function_id = get_logical_id_from_intrinsic(properties.get("FunctionName"))
+        resource_dlq_id = get_logical_id_from_intrinsic(dictfind(properties, "DestinationConfig.OnFailure.Destination"))
+        if (
+            resource_function_id
+            and resource_dlq_id
+            and function_id == resource_function_id
+            and dlq_id == resource_dlq_id
+        ):
+            yield logical_id
 
 
 def _is_valid_resource_reference(obj: Dict[str, Any]) -> bool:

samtranslator/model/connector_profiles/profiles.json

@@ -234,6 +234,7 @@
         "Type": "AWS_IAM_ROLE_MANAGED_POLICY",
         "Properties": {
           "SourcePolicy": true,
+          "DependedBy": "SOURCE_EVENT_SOURCE_MAPPING_DLQ",
           "AccessCategories": {
             "Read": {
               "Statement": [

samtranslator/model/sam_resources.py

@@ -7,6 +7,7 @@
     ConnectorResourceError,
     add_depends_on,
     get_event_source_mappings,
+    get_event_source_mapping_dlqs,
     get_resource_reference,
 )
 from samtranslator.model.connector_profiles.profile import (
@@ -1759,6 +1760,14 @@ def _construct_iam_policy(
                 # There can only be a single ESM from a resource to function, otherwise deployment fails
                 if len(esm_ids) == 1:
                     add_depends_on(esm_ids[0], policy.logical_id, resource_resolver)
+        if depended_by == "SOURCE_EVENT_SOURCE_MAPPING_DLQ":
+            if source.logical_id and destination.logical_id:
+                # The dependency type assumes Source is a AWS::Lambda::Function
+                esm_ids = list(
+                    get_event_source_mapping_dlqs(source.logical_id, destination.logical_id, resource_resolver)
+                )
+                if len(esm_ids) == 1:
+                    add_depends_on(esm_ids[0], policy.logical_id, resource_resolver)
         if depended_by == "SOURCE":
             if source.logical_id:
                 add_depends_on(source.logical_id, policy.logical_id, resource_resolver)

samtranslator/utils/__init__.py

@@ -0,0 +1,13 @@
+from typing import Any
+
+
+def dictfind(obj: Any, path: str, separator: str = ".") -> Any:
+    if not isinstance(obj, dict):
+        return None
+    keys = path.split(separator)
+    v = obj
+    for k in keys:
+        if k not in v:
+            return None
+        v = v[k]
+    return v

tests/translator/input/connector_esm_dlq.yaml

@@ -0,0 +1,71 @@
+Transform: AWS::Serverless-2016-10-31
+Resources:
+  MyTable:
+    Type: AWS::DynamoDB::Table
+    Properties:
+      BillingMode: PAY_PER_REQUEST
+      AttributeDefinitions:
+        - AttributeName: id
+          AttributeType: S
+      KeySchema:
+        - AttributeName: id
+          KeyType: HASH
+      StreamSpecification:
+        StreamViewType: NEW_IMAGE
+
+  MyRole:
+    Type: AWS::IAM::Role
+    Properties:
+      AssumeRolePolicyDocument:
+        Statement:
+          - Effect: Allow
+            Action: sts:AssumeRole
+            Principal:
+              Service: lambda.amazonaws.com
+      ManagedPolicyArns:
+        - arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole
+
+  MyFunction:
+    Type: AWS::Lambda::Function
+    Properties:
+      Runtime: nodejs16.x
+      Handler: index.handler
+      Role: !GetAtt MyRole.Arn
+      Code:
+        ZipFile: |
+          exports.handler = async (event, context) => {
+            console.log(JSON.stringify(event));
+          };
+
+  MyQueue:
+    Type: AWS::SQS::Queue
+
+  MyEventSourceMapping:
+    Type: AWS::Lambda::EventSourceMapping
+    Properties:
+      EventSourceArn: !GetAtt MyTable.StreamArn
+      FunctionName: !Ref MyFunction
+      StartingPosition: TRIM_HORIZON
+      DestinationConfig:
+        OnFailure:
+          Destination: !GetAtt MyQueue.Arn
+
+  MyConnector:
+    Type: AWS::Serverless::Connector
+    Properties:
+      Source: 
+        Id: MyTable
+      Destination: 
+        Id: MyFunction
+      Permissions:
+        - Read
+
+  LambdaToQueue:
+    Type: AWS::Serverless::Connector
+    Properties:
+      Source: 
+        Id: MyFunction
+      Destination: 
+        Id: MyQueue
+      Permissions:
+        - Write

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

@@ -0,0 +1,187 @@
+{
+ "Resources": {
+  "MyTable": {
+   "Type": "AWS::DynamoDB::Table",
+   "Properties": {
+    "BillingMode": "PAY_PER_REQUEST",
+    "AttributeDefinitions": [
+     {
+      "AttributeName": "id",
+      "AttributeType": "S"
+     }
+    ],
+    "KeySchema": [
+     {
+      "AttributeName": "id",
+      "KeyType": "HASH"
+     }
+    ],
+    "StreamSpecification": {
+     "StreamViewType": "NEW_IMAGE"
+    }
+   }
+  },
+  "MyRole": {
+   "Type": "AWS::IAM::Role",
+   "Properties": {
+    "AssumeRolePolicyDocument": {
+     "Statement": [
+      {
+       "Effect": "Allow",
+       "Action": "sts:AssumeRole",
+       "Principal": {
+        "Service": "lambda.amazonaws.com"
+       }
+      }
+     ]
+    },
+    "ManagedPolicyArns": [
+     "arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole"
+    ]
+   }
+  },
+  "MyFunction": {
+   "Type": "AWS::Lambda::Function",
+   "Properties": {
+    "Runtime": "nodejs16.x",
+    "Handler": "index.handler",
+    "Role": {
+     "Fn::GetAtt": [
+      "MyRole",
+      "Arn"
+     ]
+    },
+    "Code": {
+     "ZipFile": "exports.handler = async (event, context) => {\n  console.log(JSON.stringify(event));\n};\n"
+    }
+   }
+  },
+  "MyQueue": {
+   "Type": "AWS::SQS::Queue"
+  },
+  "MyEventSourceMapping": {
+   "Type": "AWS::Lambda::EventSourceMapping",
+   "Properties": {
+    "EventSourceArn": {
+     "Fn::GetAtt": [
+      "MyTable",
+      "StreamArn"
+     ]
+    },
+    "FunctionName": {
+     "Ref": "MyFunction"
+    },
+    "StartingPosition": "TRIM_HORIZON",
+    "DestinationConfig": {
+     "OnFailure": {
+      "Destination": {
+       "Fn::GetAtt": [
+        "MyQueue",
+        "Arn"
+       ]
+      }
+     }
+    }
+   },
+   "DependsOn": [
+    "MyConnectorPolicy",
+    "LambdaToQueuePolicy"
+   ]
+  },
+  "MyConnectorPolicy": {
+   "Type": "AWS::IAM::ManagedPolicy",
+   "Metadata": {
+    "aws:sam:connectors": {
+     "MyConnector": {
+      "Source": {
+       "Type": "AWS::DynamoDB::Table"
+      },
+      "Destination": {
+       "Type": "AWS::Lambda::Function"
+      }
+     }
+    }
+   },
+   "Properties": {
+    "PolicyDocument": {
+     "Version": "2012-10-17",
+     "Statement": [
+      {
+       "Effect": "Allow",
+       "Action": [
+        "dynamodb:DescribeStream",
+        "dynamodb:GetRecords",
+        "dynamodb:GetShardIterator",
+        "dynamodb:ListStreams"
+       ],
+       "Resource": [
+        {
+         "Fn::Sub": [
+          "${SourceArn}/stream/*",
+          {
+           "SourceArn": {
+            "Fn::GetAtt": [
+             "MyTable",
+             "Arn"
+            ]
+           }
+          }
+         ]
+        }
+       ]
+      }
+     ]
+    },
+    "Roles": [
+     {
+      "Ref": "MyRole"
+     }
+    ]
+   }
+  },
+  "LambdaToQueuePolicy": {
+   "Type": "AWS::IAM::ManagedPolicy",
+   "Metadata": {
+    "aws:sam:connectors": {
+     "LambdaToQueue": {
+      "Source": {
+       "Type": "AWS::Lambda::Function"
+      },
+      "Destination": {
+       "Type": "AWS::SQS::Queue"
+      }
+     }
+    }
+   },
+   "Properties": {
+    "PolicyDocument": {
+     "Version": "2012-10-17",
+     "Statement": [
+      {
+       "Effect": "Allow",
+       "Action": [
+        "sqs:DeleteMessage",
+        "sqs:SendMessage",
+        "sqs:ChangeMessageVisibility",
+        "sqs:PurgeQueue"
+       ],
+       "Resource": [
+        {
+         "Fn::GetAtt": [
+          "MyQueue",
+          "Arn"
+         ]
+        }
+       ]
+      }
+     ]
+    },
+    "Roles": [
+     {
+      "Ref": "MyRole"
+     }
+    ]
+   }
+  }
+ }
+}