From d6a611fec54a1dc89edb5f6633257aa58cdf052d Mon Sep 17 00:00:00 2001 From: Avanish Garg Date: Wed, 28 Oct 2026 23:01:32 +0530 Subject: [PATCH] fix(lambda-worker): create OTel tracer provider once per process apply_defaults is called from the configure callback, which run_worker invokes on every Lambda invocation. Each call built a new TracerProvider with an OTLP BatchSpanProcessor, but OpenTelemetry only allows the global provider to be set once, so from the second warm invocation on the new provider was never installed or shut down, leaking an export thread and gRPC channel per invocation. Install the provider once and reuse it on later calls. If another provider already owns the global slot, shut the unused one down. --- changelog/fixed/frugal-firefly.md | 1 + temporalio/contrib/aws/lambda_worker/otel.py | 94 ++++++++++------- tests/contrib/aws/lambda_worker/test_otel.py | 100 +++++++++++++++++++ 3 files changed, 161 insertions(+), 34 deletions(-) create mode 100644 changelog/fixed/frugal-firefly.md diff --git a/changelog/fixed/frugal-firefly.md b/changelog/fixed/frugal-firefly.md new file mode 100644 index 0000000000..8fd27ee9b7 --- /dev/null +++ b/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. diff --git a/temporalio/contrib/aws/lambda_worker/otel.py b/temporalio/contrib/aws/lambda_worker/otel.py index 5b5dc91e11..d6834aa612 100644 --- a/temporalio/contrib/aws/lambda_worker/otel.py +++ b/temporalio/contrib/aws/lambda_worker/otel.py @@ -21,6 +21,7 @@ from temporalio.contrib.aws.lambda_worker._configure import LambdaWorkerConfig from temporalio.contrib.opentelemetry import ( OpenTelemetryPlugin, + ReplaySafeTracerProvider, _serverless, create_tracer_provider, ) @@ -28,6 +29,8 @@ 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) diff --git a/tests/contrib/aws/lambda_worker/test_otel.py b/tests/contrib/aws/lambda_worker/test_otel.py index 97f3d96471..3f7e6285f6 100644 --- a/tests/contrib/aws/lambda_worker/test_otel.py +++ b/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()