Merge branch 'main' into release-v1.82.0

Author: ssenchenko

docs/globals.rst

@@ -74,6 +74,7 @@ Currently, the following resources and properties are being supported:
       Architectures:
       EphemeralStorage:
       RuntimeManagementConfig:
+      LoggingConfig:
 
     Api:
       # Properties of AWS::Serverless::Api

integration/combination/test_function_with_msk.py

@@ -28,6 +28,15 @@ def test_function_with_msk_trigger_using_manage_policy(self):
         parameters.append(self.generate_parameter("MskClusterName2", cluster_name))
         self._common_validations_for_MSK("combination/function_with_msk_using_managed_policy", parameters)
 
+    def test_function_with_msk_trigger_and_s3_onfailure_events_destinations(self):
+        companion_stack_outputs = self.companion_stack_outputs
+        parameters = self.get_parameters(companion_stack_outputs)
+        cluster_name = "MskCluster3-" + generate_suffix()
+        parameters.append(self.generate_parameter("MskClusterName3", cluster_name))
+        self._common_validations_for_MSK(
+            "combination/function_with_msk_trigger_and_s3_onfailure_events_destinations", parameters
+        )
+
     def _common_validations_for_MSK(self, file_name, parameters):
         self.create_and_verify_stack(file_name, parameters)
 

integration/resources/expected/combination/function_with_msk_trigger_and_s3_onfailure_events_destinations.json

@@ -0,0 +1,22 @@
+[
+  {
+    "LogicalResourceId": "MyMskStreamProcessor",
+    "ResourceType": "AWS::Lambda::Function"
+  },
+  {
+    "LogicalResourceId": "MyLambdaExecutionRole",
+    "ResourceType": "AWS::IAM::Role"
+  },
+  {
+    "LogicalResourceId": "MyMskCluster",
+    "ResourceType": "AWS::MSK::Cluster"
+  },
+  {
+    "LogicalResourceId": "MyMskStreamProcessorMyMskEvent",
+    "ResourceType": "AWS::Lambda::EventSourceMapping"
+  },
+  {
+    "LogicalResourceId": "PreCreatedS3Bucket",
+    "ResourceType": "AWS::S3::Bucket"
+  }
+]

integration/resources/templates/combination/function_with_msk_trigger_and_s3_onfailure_events_destinations.yaml

@@ -0,0 +1,80 @@
+Parameters:
+  PreCreatedSubnetOne:
+    Type: String
+  PreCreatedSubnetTwo:
+    Type: String
+  MskClusterName3:
+    Type: String
+
+Resources:
+  MyLambdaExecutionRole:
+    Type: AWS::IAM::Role
+    Properties:
+      AssumeRolePolicyDocument:
+        Version: '2012-10-17'
+        Statement:
+        - Action: [sts:AssumeRole]
+          Effect: Allow
+          Principal:
+            Service: [lambda.amazonaws.com]
+      Policies:
+      - PolicyName: IntegrationTestExecution
+        PolicyDocument:
+          Statement:
+          - Action: [kafka:DescribeCluster, kafka:GetBootstrapBrokers, ec2:CreateNetworkInterface,
+              ec2:DescribeNetworkInterfaces, ec2:DescribeVpcs, ec2:DeleteNetworkInterface,
+              ec2:DescribeSubnets, ec2:DescribeSecurityGroups, logs:CreateLogGroup,
+              logs:CreateLogStream, logs:PutLogEvents]
+            Effect: Allow
+            Resource: '*'
+      ManagedPolicyArns: [arn:aws:iam::aws:policy/service-role/AWSLambdaBasicExecutionRole]
+      Tags:
+      - {Value: SAM, Key: lambda:createdBy}
+
+  MyMskCluster:
+    Type: AWS::MSK::Cluster
+    Properties:
+      BrokerNodeGroupInfo:
+        ClientSubnets:
+        - Ref: PreCreatedSubnetOne
+        - Ref: PreCreatedSubnetTwo
+        InstanceType: kafka.t3.small
+        StorageInfo:
+          EBSStorageInfo:
+            VolumeSize: 1
+      ClusterName:
+        Ref: MskClusterName3
+      KafkaVersion: 2.4.1.1
+      NumberOfBrokerNodes: 2
+
+  MyMskStreamProcessor:
+    Type: AWS::Serverless::Function
+    Properties:
+      Runtime: nodejs18.x
+      Handler: index.handler
+      CodeUri: ${codeuri}
+      Role:
+        Fn::GetAtt: [MyLambdaExecutionRole, Arn]
+      Events:
+        MyMskEvent:
+          Type: MSK
+          Properties:
+            StartingPosition: LATEST
+            Stream:
+              Ref: MyMskCluster
+            Topics:
+            - MyDummyTestTopic
+            DestinationConfig:
+              OnFailure:
+                Type: S3
+                Destination:
+                  Fn::GetAtt:
+                  - PreCreatedS3Bucket
+                  - Arn
+
+  PreCreatedS3Bucket:
+    Type: AWS::S3::Bucket
+    DeletionPolicy: Delete
+
+Metadata:
+  SamTransformTest: true

samtranslator/internal/schema_source/aws_serverless_function.py

@@ -414,6 +414,7 @@ class MSKEventProperties(BaseModel):
     Stream: PassThroughProp = mskeventproperties("Stream")
     Topics: PassThroughProp = mskeventproperties("Topics")
     SourceAccessConfigurations: Optional[PassThroughProp] = mskeventproperties("SourceAccessConfigurations")
+    DestinationConfig: Optional[PassThroughProp]  # TODO: add documentation
 
 
 class MSKEvent(BaseModel):
@@ -506,6 +507,7 @@ class ScheduleV2Event(BaseModel):
 EphemeralStorage = Optional[PassThroughProp]
 SnapStart = Optional[PassThroughProp]  # TODO: check the type
 RuntimeManagementConfig = Optional[PassThroughProp]  # TODO: check the type
+LoggingConfig = Optional[PassThroughProp]  # TODO: add documentation
 
 
 class Properties(BaseModel):
@@ -631,6 +633,7 @@ class Properties(BaseModel):
     Tracing: Optional[Tracing] = prop("Tracing")
     VersionDescription: Optional[PassThroughProp] = prop("VersionDescription")
     VpcConfig: Optional[VpcConfig] = prop("VpcConfig")
+    LoggingConfig: Optional[PassThroughProp]  # TODO: add documentation
 
 
 class Globals(BaseModel):
@@ -688,6 +691,7 @@ class Globals(BaseModel):
     )
     SnapStart: Optional[SnapStart] = prop("SnapStart")
     RuntimeManagementConfig: Optional[RuntimeManagementConfig] = prop("RuntimeManagementConfig")
+    LoggingConfig: Optional[PassThroughProp]  # TODO: add documentation
 
 
 class Resource(ResourceAttributes):

samtranslator/model/eventsources/pull.py

@@ -172,13 +172,15 @@ def to_cloudformation(self, **kwargs):  # type: ignore[no-untyped-def] # noqa: P
             # `Type` property is for sam to attach the right policies
             destination_type = on_failure.get("Type")
 
-            # SAM attaches the policies for SQS or SNS only if 'Type' is given
+            # SAM attaches the policies for SQS, SNS or S3 only if 'Type' is given
             if destination_type:
                 # delete this field as its used internally for SAM to determine the policy
                 del on_failure["Type"]
-                # the values 'SQS' and 'SNS' are allowed. No intrinsics are allowed
-                if destination_type not in ["SQS", "SNS"]:
-                    raise InvalidEventException(self.logical_id, "The only valid values for 'Type' are 'SQS' and 'SNS'")
+                # the values 'SQS', 'SNS', and 'S3' are allowed. No intrinsics are allowed
+                if destination_type not in ["SQS", "SNS", "S3"]:
+                    raise InvalidEventException(
+                        self.logical_id, "The only valid values for 'Type' are 'SQS', 'SNS', and 'S3'"
+                    )
                 if destination_type == "SQS":
                     queue_arn = on_failure.get("Destination")
                     destination_config_policy = IAMRolePolicies().sqs_send_message_role_policy(
@@ -189,6 +191,11 @@ def to_cloudformation(self, **kwargs):  # type: ignore[no-untyped-def] # noqa: P
                     destination_config_policy = IAMRolePolicies().sns_publish_role_policy(
                         sns_topic_arn, self.logical_id
                     )
+                elif destination_type == "S3":
+                    s3_arn = on_failure.get("Destination")
+                    destination_config_policy = IAMRolePolicies().s3_send_event_payload_role_policy(
+                        s3_arn, self.logical_id
+                    )
 
             lambda_eventsourcemapping.DestinationConfig = self.DestinationConfig
 

samtranslator/model/iam.py

@@ -126,6 +126,19 @@ def sns_publish_role_policy(cls, topic_arn: Any, logical_id: str) -> Dict[str, A
             "PolicyDocument": {"Statement": [{"Action": "sns:publish", "Effect": "Allow", "Resource": topic_arn}]},
         }
 
+    @classmethod
+    def s3_send_event_payload_role_policy(cls, s3_arn: Any, logical_id: str) -> Dict[str, Any]:
+        s3_arn_with_wild_card = {"Fn::Join": ["/", [s3_arn, "*"]]}
+        return {
+            "PolicyName": logical_id + "S3Policy",
+            "PolicyDocument": {
+                "Statement": [
+                    {"Action": "s3:PutObject", "Effect": "Allow", "Resource": s3_arn_with_wild_card},
+                    {"Action": "s3:ListBucket", "Effect": "Allow", "Resource": s3_arn},
+                ]
+            },
+        }
+
     @classmethod
     def event_bus_put_events_role_policy(cls, event_bus_arn: Any, logical_id: str) -> Dict[str, Any]:
         return {

samtranslator/model/lambda_.py

@@ -34,6 +34,7 @@ class LambdaFunction(Resource):
         "SnapStart": GeneratedProperty(),
         "EphemeralStorage": GeneratedProperty(),
         "RuntimeManagementConfig": GeneratedProperty(),
+        "LoggingConfig": GeneratedProperty(),
     }
 
     Code: Dict[str, Any]
@@ -60,6 +61,7 @@ class LambdaFunction(Resource):
     SnapStart: Optional[Dict[str, Any]]
     EphemeralStorage: Optional[Dict[str, Any]]
     RuntimeManagementConfig: Optional[Dict[str, Any]]
+    LoggingConfig: Optional[Dict[str, Any]]
 
     runtime_attrs = {"name": lambda self: ref(self.logical_id), "arn": lambda self: fnGetAtt(self.logical_id, "Arn")}
 

samtranslator/model/sam_resources.py

@@ -179,6 +179,7 @@ class SamFunction(SamResourceMacro):
         "SnapStart": PropertyType(False, IS_DICT),
         "FunctionUrlConfig": PropertyType(False, IS_DICT),
         "RuntimeManagementConfig": PassThroughProperty(False),
+        "LoggingConfig": PassThroughProperty(False),
     }
 
     FunctionName: Optional[Intrinsicable[str]]
@@ -220,6 +221,7 @@ class SamFunction(SamResourceMacro):
     Architectures: Optional[List[Any]]
     SnapStart: Optional[Dict[str, Any]]
     FunctionUrlConfig: Optional[Dict[str, Any]]
+    LoggingConfig: Optional[Dict[str, Any]]
 
     event_resolver = ResourceTypeResolver(
         samtranslator.model.eventsources,
@@ -603,6 +605,7 @@ def _construct_lambda_function(self, intrinsics_resolver: IntrinsicsResolver) ->
         lambda_function.CodeSigningConfigArn = self.CodeSigningConfigArn
 
         lambda_function.RuntimeManagementConfig = self.RuntimeManagementConfig  # type: ignore[attr-defined]
+        lambda_function.LoggingConfig = self.LoggingConfig
         self._validate_package_type(lambda_function)
         return lambda_function
 

samtranslator/plugins/globals/globals.py

@@ -53,6 +53,7 @@ class Globals:
             "EphemeralStorage",
             "FunctionUrlConfig",
             "RuntimeManagementConfig",
+            "LoggingConfig",
         ],
         # Everything except
         #   DefinitionBody: because its hard to reason about merge of Swagger dictionaries