feat: Add observability config for Kafka based event sources

Author: bnusunny

.cfnlintrc.yaml

@@ -107,6 +107,9 @@ ignore_templates:
   - tests/translator/output/**/function_with_snapstart.json # Snapstart intentionally not attached to a lambda version which causes lint issues
   - tests/translator/output/**/managed_policies_everything.json # intentionally contains wrong arns
   - tests/translator/output/**/function_with_metrics_config.json
+  - tests/translator/output/**/function_with_self_managed_kafka_and_schema_registry.json # cfnlint is not updated to recognize the SchemaRegistryConfig property
+  - tests/translator/output/**/function_with_msk_with_schema_registry_config.json # cfnlint is not updated to recognize the SchemaRegistryConfig property
+  - tests/translator/output/**/function_with_logging_config.json # cfnlint is not updated to recognize the LoggingConfig property
   - tests/translator/output/aws-*/*capacity_provider*.json # Ignore Capacity Provider test format in non-aws partitions
 
 ignore_checks:

samtranslator/internal/schema_source/aws_serverless_function.py

@@ -412,18 +412,20 @@ class HttpApiEvent(BaseModel):
 
 class MSKEventProperties(BaseModel):
     ConsumerGroupId: Optional[PassThroughProp] = mskeventproperties("ConsumerGroupId")
-    Enabled: Optional[PassThroughProp]  # TODO: it doesn't show up in docs yet
+    Enabled: Optional[PassThroughProp] = mskeventproperties("Enabled")
     FilterCriteria: Optional[PassThroughProp] = mskeventproperties("FilterCriteria")
-    KmsKeyArn: Optional[PassThroughProp]  # TODO: add documentation
+    KmsKeyArn: Optional[PassThroughProp] = mskeventproperties("KmsKeyArn")
     MaximumBatchingWindowInSeconds: Optional[PassThroughProp] = mskeventproperties("MaximumBatchingWindowInSeconds")
     StartingPosition: Optional[PassThroughProp] = mskeventproperties("StartingPosition")
     StartingPositionTimestamp: Optional[PassThroughProp] = mskeventproperties("StartingPositionTimestamp")
     Stream: PassThroughProp = mskeventproperties("Stream")
     Topics: PassThroughProp = mskeventproperties("Topics")
     SourceAccessConfigurations: Optional[PassThroughProp] = mskeventproperties("SourceAccessConfigurations")
-    DestinationConfig: Optional[PassThroughProp]  # TODO: add documentation
-    ProvisionedPollerConfig: Optional[PassThroughProp]
-    SchemaRegistryConfig: Optional[PassThroughProp]
+    DestinationConfig: Optional[PassThroughProp] = mskeventproperties("DestinationConfig")
+    ProvisionedPollerConfig: Optional[PassThroughProp] = mskeventproperties("ProvisionedPollerConfig")
+    SchemaRegistryConfig: Optional[PassThroughProp] = mskeventproperties("SchemaRegistryConfig")
+    MetricsConfig: Optional[PassThroughProp] = mskeventproperties("MetricsConfig")
+    LoggingConfig: Optional[PassThroughProp] = mskeventproperties("LoggingConfig")
     BisectBatchOnFunctionError: Optional[PassThroughProp] = mskeventproperties("BisectBatchOnFunctionError")
     FunctionResponseTypes: Optional[PassThroughProp] = mskeventproperties("FunctionResponseTypes")
     MaximumRecordAgeInSeconds: Optional[PassThroughProp] = mskeventproperties("MaximumRecordAgeInSeconds")
@@ -461,13 +463,15 @@ class SelfManagedKafkaEventProperties(BaseModel):
     KafkaBootstrapServers: Optional[List[SamIntrinsicable[str]]] = selfmanagedkafkaeventproperties(
         "KafkaBootstrapServers"
     )
-    KmsKeyArn: Optional[PassThroughProp]  # TODO: add documentation
+    KmsKeyArn: Optional[PassThroughProp] = selfmanagedkafkaeventproperties("KmsKeyArn")
     SourceAccessConfigurations: PassThroughProp = selfmanagedkafkaeventproperties("SourceAccessConfigurations")
-    StartingPosition: Optional[PassThroughProp]  # TODO: add documentation
-    StartingPositionTimestamp: Optional[PassThroughProp]  # TODO: add documentation
+    StartingPosition: Optional[PassThroughProp] = selfmanagedkafkaeventproperties("StartingPosition")
+    StartingPositionTimestamp: Optional[PassThroughProp] = selfmanagedkafkaeventproperties("StartingPositionTimestamp")
     Topics: PassThroughProp = selfmanagedkafkaeventproperties("Topics")
-    ProvisionedPollerConfig: Optional[PassThroughProp]
-    SchemaRegistryConfig: Optional[PassThroughProp]
+    MetricsConfig: Optional[PassThroughProp] = selfmanagedkafkaeventproperties("MetricsConfig")
+    ProvisionedPollerConfig: Optional[PassThroughProp] = selfmanagedkafkaeventproperties("ProvisionedPollerConfig")
+    SchemaRegistryConfig: Optional[PassThroughProp] = selfmanagedkafkaeventproperties("SchemaRegistryConfig")
+    LoggingConfig: Optional[PassThroughProp] = selfmanagedkafkaeventproperties("LoggingConfig")
     BisectBatchOnFunctionError: Optional[PassThroughProp] = selfmanagedkafkaeventproperties(
         "BisectBatchOnFunctionError"
     )

samtranslator/internal/schema_source/sam-docs.json

@@ -61,6 +61,7 @@ class PullEventSource(ResourceMacro, metaclass=ABCMeta):
         "ProvisionedPollerConfig": PropertyType(False, IS_DICT),
         "SchemaRegistryConfig": PropertyType(False, IS_DICT),
         "MetricsConfig": PropertyType(False, IS_DICT),
+        "LoggingConfig": PropertyType(False, IS_DICT),
     }
 
     BatchSize: Optional[Intrinsicable[int]]
@@ -87,6 +88,7 @@ class PullEventSource(ResourceMacro, metaclass=ABCMeta):
     ProvisionedPollerConfig: Optional[Dict[str, Any]]
     SchemaRegistryConfig: Optional[Dict[str, Any]]
     MetricsConfig: Optional[Dict[str, Any]]
+    LoggingConfig: Optional[Dict[str, Any]]
 
     @abstractmethod
     def get_policy_arn(self) -> Optional[str]:
@@ -159,6 +161,7 @@ def to_cloudformation(self, **kwargs):  # type: ignore[no-untyped-def] # noqa: P
         lambda_eventsourcemapping.ScalingConfig = self.ScalingConfig
         lambda_eventsourcemapping.ProvisionedPollerConfig = self.ProvisionedPollerConfig
         lambda_eventsourcemapping.MetricsConfig = self.MetricsConfig
+        lambda_eventsourcemapping.LoggingConfig = self.LoggingConfig
         self._validate_filter_criteria()
 
         if self.KafkaBootstrapServers:

samtranslator/model/eventsources/pull.py

@@ -135,6 +135,7 @@ class LambdaEventSourceMapping(Resource):
         "ScalingConfig": GeneratedProperty(),
         "ProvisionedPollerConfig": GeneratedProperty(),
         "MetricsConfig": GeneratedProperty(),
+        "LoggingConfig": GeneratedProperty(),
     }
 
     runtime_attrs = {"name": lambda self: ref(self.logical_id)}