create OTel tracer provider once per process

changelog/fixed/frugal-firefly.md

@@ -0,0 +1 @@
+`temporalio.contrib.aws.lambda_worker.otel.apply_defaults` now creates its tracer provider once per process, so warm Lambda invocations no longer leak an OTLP span exporter thread and gRPC channel each time.

temporalio/contrib/aws/lambda_worker/otel.py

@@ -21,13 +21,16 @@
 from temporalio.contrib.aws.lambda_worker._configure import LambdaWorkerConfig
 from temporalio.contrib.opentelemetry import (
     OpenTelemetryPlugin,
+    ReplaySafeTracerProvider,
     _serverless,
     create_tracer_provider,
 )
 from temporalio.runtime import Runtime, TelemetryConfig
 
 logger = logging.getLogger(__name__)
 
+_installed_tracer_provider: ReplaySafeTracerProvider | None = None
+
 
 @dataclass
 class OtelOptions:
@@ -64,6 +67,60 @@ def _resolve_endpoint(options: OtelOptions) -> str:
     return _serverless.resolve_endpoint(options.collector_endpoint)
 
 
+def _ensure_tracer_provider(endpoint: str, service_name: str) -> None:
+    """Install the global replay-safe ``TracerProvider`` once per process.
+
+    ``configure`` runs on every Lambda invocation, but OpenTelemetry only allows the global
+    provider to be set once and a warm container keeps the process alive. Later calls
+    reuse the installed provider instead of building (and leaking) a new span processor.
+    """
+    global _installed_tracer_provider
+    if (
+        _installed_tracer_provider is not None
+        and get_tracer_provider() is _installed_tracer_provider
+    ):
+        return
+
+    resource = Resource.create({SERVICE_NAME: service_name})
+
+    # Try to use X-Ray ID generator if available.
+    try:
+        from opentelemetry.sdk.extension.aws.trace import (  # type: ignore[reportMissingTypeStubs]
+            AwsXRayIdGenerator,
+        )
+
+        tracer_provider = create_tracer_provider(
+            resource=resource, id_generator=AwsXRayIdGenerator()
+        )
+    except ImportError:
+        logger.warning(
+            "opentelemetry-sdk-extension-aws is not installed; "
+            "X-Ray trace ID generation is disabled. "
+            "Install the 'lambda-worker-otel' extra for full ADOT support."
+        )
+        tracer_provider = create_tracer_provider(resource=resource)
+
+    # Use OTLP gRPC exporter if available, otherwise skip trace export.
+    try:
+        tracer_provider.add_span_processor(
+            _serverless.build_otlp_span_processor(endpoint, insecure=True)
+        )
+    except ImportError:
+        logger.warning(
+            "opentelemetry-exporter-otlp-proto-grpc is not installed; "
+            "traces will not be exported to the OTLP collector. "
+            "Install the 'lambda-worker-otel' extra for full ADOT support."
+        )
+
+    # Set as global so the OpenTelemetryPlugin picks it up.
+    set_tracer_provider(tracer_provider)
+    if get_tracer_provider() is tracer_provider:
+        _installed_tracer_provider = tracer_provider
+    else:
+        # Another provider is already global; ours is unused, so don't leak its exporter.
+        tracer_provider.shutdown()
+
+
 def apply_defaults(
     config: LambdaWorkerConfig,
     options: OtelOptions | None = None,
@@ -78,7 +135,8 @@ def apply_defaults(
     Creates a replay-safe ``TracerProvider`` (with X-Ray ID generator and OTLP gRPC exporter if
     available) and sets it as the global OpenTelemetry tracer provider. The
     :py:class:`temporalio.contrib.opentelemetry.OpenTelemetryPlugin` uses the global provider, so
-    it must be set before the worker starts.
+    it must be set before the worker starts. The provider is created once per process: when this
+    is called again on a warm Lambda invocation, the installed provider is reused.
 
     The collector endpoint defaults to ``http://localhost:4317``, which is the endpoint expected by
     the ADOT collector Lambda layer.
@@ -108,39 +166,7 @@ def apply_defaults(
     runtime = Runtime(telemetry=telemetry_config)
     config.client_connect_config["runtime"] = runtime
 
-    resource = Resource.create({SERVICE_NAME: service_name})
-
-    # Try to use X-Ray ID generator if available.
-    try:
-        from opentelemetry.sdk.extension.aws.trace import (  # type: ignore[reportMissingTypeStubs]
-            AwsXRayIdGenerator,
-        )
-
-        tracer_provider = create_tracer_provider(
-            resource=resource, id_generator=AwsXRayIdGenerator()
-        )
-    except ImportError:
-        logger.warning(
-            "opentelemetry-sdk-extension-aws is not installed; "
-            "X-Ray trace ID generation is disabled. "
-            "Install the 'lambda-worker-otel' extra for full ADOT support."
-        )
-        tracer_provider = create_tracer_provider(resource=resource)
-
-    # Use OTLP gRPC exporter if available, otherwise skip trace export.
-    try:
-        tracer_provider.add_span_processor(
-            _serverless.build_otlp_span_processor(endpoint, insecure=True)
-        )
-    except ImportError:
-        logger.warning(
-            "opentelemetry-exporter-otlp-proto-grpc is not installed; "
-            "traces will not be exported to the OTLP collector. "
-            "Install the 'lambda-worker-otel' extra for full ADOT support."
-        )
-
-    # Set as global so the OpenTelemetryPlugin picks it up.
-    set_tracer_provider(tracer_provider)
+    _ensure_tracer_provider(endpoint, service_name)
 
     apply_tracing(config)
 

tests/contrib/aws/lambda_worker/test_otel.py

@@ -2,11 +2,23 @@
 
 from __future__ import annotations
 
+import threading
+from collections.abc import Iterator, Sequence
 from datetime import timedelta
+from typing import Any
 from unittest.mock import patch
 
 import pytest
+from opentelemetry import trace
+from opentelemetry.sdk.trace import ReadableSpan, TracerProvider
+from opentelemetry.sdk.trace.export import (
+    BatchSpanProcessor,
+    SpanExporter,
+    SpanExportResult,
+)
+from opentelemetry.util._once import Once
 
+from temporalio.contrib.aws.lambda_worker import otel
 from temporalio.contrib.aws.lambda_worker._configure import (
     LambdaWorkerConfig,
     _run_shutdown_hooks,
@@ -149,6 +161,94 @@ def test_default_options_used_when_none(self) -> None:
         assert len(config.shutdown_hooks) == 1
 
 
+class _RecordingExporter(SpanExporter):
+    def __init__(self) -> None:
+        super().__init__()
+        self.exported: list[str] = []
+
+    def export(self, spans: Sequence[ReadableSpan]) -> SpanExportResult:
+        self.exported.extend(span.name for span in spans)
+        return SpanExportResult.SUCCESS
+
+
+def _span_export_threads() -> int:
+    return sum(
+        1 for t in threading.enumerate() if t.name == "OtelBatchSpanRecordProcessor"
+    )
+
+
+class TestApplyDefaultsWarmInvocations:
+    """``configure`` runs on every Lambda invocation in the same warm process."""
+
+    @pytest.fixture(autouse=True)
+    def exporters(
+        self, monkeypatch: pytest.MonkeyPatch
+    ) -> Iterator[list[_RecordingExporter]]:
+        """Isolate the global provider and record every span exporter that gets built."""
+        # OpenTelemetry only allows the global provider to be set once per process.
+        exporters: list[_RecordingExporter] = []
+        monkeypatch.setattr(trace, "_TRACER_PROVIDER", None)
+        monkeypatch.setattr(trace, "_TRACER_PROVIDER_SET_ONCE", Once())
+        monkeypatch.setattr(otel, "_installed_tracer_provider", None, raising=False)
+
+        def build_processor(*_args: Any, **_kwargs: Any) -> BatchSpanProcessor:
+            exporter = _RecordingExporter()
+            exporters.append(exporter)
+            return BatchSpanProcessor(exporter)
+
+        monkeypatch.setattr(
+            otel._serverless,  # type: ignore[reportPrivateUsage]
+            "build_otlp_span_processor",
+            build_processor,
+        )
+        yield exporters
+        provider = trace.get_tracer_provider()
+        if isinstance(provider, TracerProvider):
+            provider.shutdown()
+
+    def test_reuses_provider_across_invocations(
+        self, exporters: list[_RecordingExporter]
+    ) -> None:
+        baseline = _span_export_threads()
+        providers = []
+        for _ in range(4):
+            apply_defaults(LambdaWorkerConfig())
+            providers.append(trace.get_tracer_provider())
+        assert len({id(p) for p in providers}) == 1
+        assert len(exporters) == 1
+        assert _span_export_threads() == baseline + 1
+
+    def test_does_not_warn_about_overriding_provider(
+        self, caplog: pytest.LogCaptureFixture
+    ) -> None:
+        apply_defaults(LambdaWorkerConfig())
+        caplog.clear()
+        apply_defaults(LambdaWorkerConfig())
+        assert "Overriding of current TracerProvider" not in caplog.text
+
+    @pytest.mark.asyncio
+    async def test_flush_hook_exports_on_every_invocation(
+        self, exporters: list[_RecordingExporter]
+    ) -> None:
+        for invocation in range(3):
+            config = LambdaWorkerConfig()
+            apply_defaults(config)
+            tracer = trace.get_tracer("test")
+            with tracer.start_as_current_span(f"span-{invocation}"):
+                pass
+            await _run_shutdown_hooks(config)
+        assert exporters[0].exported == ["span-0", "span-1", "span-2"]
+
+    def test_unused_provider_is_shut_down_when_global_already_set(self) -> None:
+        foreign = TracerProvider()
+        trace.set_tracer_provider(foreign)
+        baseline = _span_export_threads()
+        for _ in range(3):
+            apply_defaults(LambdaWorkerConfig())
+        assert trace.get_tracer_provider() is foreign
+        assert _span_export_threads() == baseline
+
+
 class TestOtelOptions:
     def test_defaults(self) -> None:
         opts = OtelOptions()