bugfix: AWS Lambda event source key incorrect for SNS in instrumenta… (open-telemetry#2612)

Author: zhihali

CHANGELOG.md

@@ -7,6 +7,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## Unreleased
 
+- `opentelemetry-instrumentation-aws-lambda` Bugfix: AWS Lambda event source key incorrect for SNS in instrumentation library. 
+  ([#2612](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/2612))
+
 ### Added
 
 - `opentelemetry-instrumentation-pyramid` Record exceptions raised when serving a request

instrumentation/opentelemetry-instrumentation-aws-lambda/src/opentelemetry/instrumentation/aws_lambda/__init__.py

@@ -306,9 +306,11 @@ def _instrumented_lambda_handler_call(  # noqa pylint: disable=too-many-branches
             disable_aws_context_propagation,
         )
 
-        span_kind = None
         try:
-            if lambda_event["Records"][0]["eventSource"] in {
+            event_source = lambda_event["Records"][0].get(
+                "eventSource"
+            ) or lambda_event["Records"][0].get("EventSource")
+            if event_source in {
                 "aws:sqs",
                 "aws:s3",
                 "aws:sns",

instrumentation/opentelemetry-instrumentation-aws-lambda/tests/test_aws_lambda_instrumentation_manual.py

@@ -349,12 +349,43 @@ def test_lambda_handles_multiple_consumers(self):
 
         mock_execute_lambda({"Records": [{"eventSource": "aws:sqs"}]})
         mock_execute_lambda({"Records": [{"eventSource": "aws:s3"}]})
-        mock_execute_lambda({"Records": [{"eventSource": "aws:sns"}]})
+        mock_execute_lambda({"Records": [{"EventSource": "aws:sns"}]})
         mock_execute_lambda({"Records": [{"eventSource": "aws:dynamodb"}]})
 
         spans = self.memory_exporter.get_finished_spans()
 
         assert spans
+        assert len(spans) == 4
+
+        for span in spans:
+            assert span.kind == SpanKind.CONSUMER
+
+        test_env_patch.stop()
+
+    def test_lambda_handles_invalid_event_source(self):
+        test_env_patch = mock.patch.dict(
+            "os.environ",
+            {
+                **os.environ,
+                # NOT Active Tracing
+                _X_AMZN_TRACE_ID: MOCK_XRAY_TRACE_CONTEXT_NOT_SAMPLED,
+                # NOT using the X-Ray Propagator
+                OTEL_PROPAGATORS: "tracecontext",
+            },
+        )
+        test_env_patch.start()
+
+        AwsLambdaInstrumentor().instrument()
+
+        mock_execute_lambda({"Records": [{"eventSource": "invalid_source"}]})
+
+        spans = self.memory_exporter.get_finished_spans()
+
+        assert spans
+        assert len(spans) == 1
+        assert (
+            spans[0].kind == SpanKind.SERVER
+        )  # Default to SERVER for unknown sources
 
         test_env_patch.stop()