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()