lambda instrumentation - support for flush timeout

CHANGELOG.md

@@ -7,6 +7,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 ## [Unreleased](https://github.com/open-telemetry/opentelemetry-python/compare/v1.7.0-0.26b0...HEAD)
 
+### Added
+
+- `opentelemetry-instrumentation-aws-lambda` Adds support for configurable flush timeout  via `OTEL_INSTRUMENTATION_AWS_LAMBDA_FLUSH_TIMEOUT` property. ([#825](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/825)) 
+
 ### Fixed
 
 - `opentelemetry-exporter-richconsole` Fixed attribute error on parentless spans.

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

@@ -97,6 +97,9 @@ def custom_event_context_extractor(lambda_event):
 _HANDLER = "_HANDLER"
 _X_AMZN_TRACE_ID = "_X_AMZN_TRACE_ID"
 ORIG_HANDLER = "ORIG_HANDLER"
+OTEL_INSTRUMENTATION_AWS_LAMBDA_FLUSH_TIMEOUT = (
+    "OTEL_INSTRUMENTATION_AWS_LAMBDA_FLUSH_TIMEOUT"
+)
 
 
 def _default_event_context_extractor(lambda_event: Any) -> Context:
@@ -167,6 +170,7 @@ def _determine_parent_context(
 def _instrument(
     wrapped_module_name,
     wrapped_function_name,
+    flush_timeout,
     event_context_extractor: Callable[[Any], Context],
     tracer_provider: TracerProvider = None,
 ):
@@ -222,7 +226,7 @@ def _instrumented_lambda_handler_call(
             # NOTE: `force_flush` before function quit in case of Lambda freeze.
             # Assumes we are using the OpenTelemetry SDK implementation of the
             # `TracerProvider`.
-            _tracer_provider.force_flush()
+            _tracer_provider.force_flush(flush_timeout)
         except Exception:  # pylint: disable=broad-except
             logger.error(
                 "TracerProvider was missing `force_flush` method. This is necessary in case of a Lambda freeze and would exist in the OTel SDK implementation."
@@ -262,9 +266,22 @@ def _instrument(self, **kwargs):
             self._wrapped_function_name,
         ) = lambda_handler.rsplit(".", 1)
 
+        flush_timeout_env = os.environ.get(
+            OTEL_INSTRUMENTATION_AWS_LAMBDA_FLUSH_TIMEOUT, ""
+        )
+        flush_timeout = 30000
+        try:
+            flush_timeout = int(flush_timeout_env)
+        except ValueError:
+            logger.warning(
+                "Could not convert OTEL_INSTRUMENTATION_AWS_LAMBDA_FLUSH_TIMEOUT value %s to int",
+                flush_timeout_env,
+            )
+
         _instrument(
             self._wrapped_module_name,
             self._wrapped_function_name,
+            flush_timeout,
             event_context_extractor=kwargs.get(
                 "event_context_extractor", _default_event_context_extractor
             ),

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

@@ -19,6 +19,7 @@
 from opentelemetry.instrumentation.aws_lambda import (
     _HANDLER,
     _X_AMZN_TRACE_ID,
+    OTEL_INSTRUMENTATION_AWS_LAMBDA_FLUSH_TIMEOUT,
     AwsLambdaInstrumentor,
 )
 from opentelemetry.propagate import get_global_textmap
@@ -246,3 +247,30 @@ def custom_event_context_extractor(lambda_event):
         self.assertTrue(parent_context.is_remote)
 
         test_env_patch.stop()
+
+    def test_lambda_no_error_with_invalid_flush_timeout(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",
+                OTEL_INSTRUMENTATION_AWS_LAMBDA_FLUSH_TIMEOUT: "invalid-timeout-string",
+            },
+        )
+        test_env_patch.start()
+
+        AwsLambdaInstrumentor().instrument()
+
+        mock_execute_lambda()
+
+        spans = self.memory_exporter.get_finished_spans()
+
+        assert spans
+
+        self.assertEqual(len(spans), 1)
+
+        test_env_patch.stop()