From b7bfe80f844ed91aba30361a99839ba055ac29bc Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Fri, 14 Aug 2026 19:56:17 +0000 Subject: [PATCH 1/3] fix(otel): scope deterministic IDs to plugin tracers --- .../README.md | 28 +- .../pyproject.toml | 7 +- .../__init__.py | 2 - .../deterministic_id_generator.py | 145 ++++---- .../execution_plugin.py | 174 +++++----- .../instrumentations.py | 78 +---- .../invocation_plugin.py | 221 +++++++------ .../otel_plugin_config.py | 37 +-- .../provider.py | 136 +------- .../tests/test_deterministic_id_generator.py | 313 +++++++++--------- .../tests/test_execution_plugin.py | 30 +- .../test_execution_plugin_integration.py | 150 ++++++++- .../tests/test_invocation_plugin.py | 44 ++- .../test_invocation_plugin_integration.py | 177 +++++++++- .../tests/test_provider.py | 76 +---- pyproject.toml | 1 - 16 files changed, 845 insertions(+), 774 deletions(-) diff --git a/packages/aws-durable-execution-sdk-python-otel/README.md b/packages/aws-durable-execution-sdk-python-otel/README.md index 4eb563aa..4a6c5eba 100644 --- a/packages/aws-durable-execution-sdk-python-otel/README.md +++ b/packages/aws-durable-execution-sdk-python-otel/README.md @@ -1,15 +1,16 @@ # AWS Durable Execution SDK - OpenTelemetry Plugin -OpenTelemetry instrumentation plugin for the [AWS Durable Execution SDK for Python](https://github.com/aws/aws-durable-execution-sdk-python). Emits distributed traces that correlate across multiple Lambda invocations of a single durable execution, producing deterministic span and trace IDs so that spans from different invocations are stitched into a single coherent trace. +OpenTelemetry instrumentation plugin for the [AWS Durable Execution SDK for Python](https://github.com/aws/aws-durable-execution-sdk-python). Emits durable execution spans with deterministic workflow and operation IDs while keeping invocation spans in the ambient Lambda trace. ## Features -- **Deterministic Trace IDs**: All invocations of the same durable execution share a single trace, derived from the X-Ray trace header or execution ARN +- **Deterministic Workflow Traces**: Durable operations use an execution-derived trace that is independent of the ambient Lambda/X-Ray trace +- **Ambient Invocation Traces**: Invocation spans inherit the active Lambda or extracted upstream context - **Span-per-Operation**: Each durable operation (step, wait, invoke) gets its own span with accurate timing -- **Continuation Spans**: Operations completing in a different invocation are linked back to the original span +- **Continuation Spans**: Operations completing in another invocation produce a new correlated span without fabricating an unobserved prior span context - **Log Correlation**: Enrich application logs with trace ID and span ID for end-to-end observability -- **Configurable Sampling**: Control trace volume via plugin options -- **Self-Contained Setup**: No manual TracerProvider configuration required +- **Provider Integration**: Use the global ADOT provider or supply an explicit SDK `TracerProvider` +- **Provider-Managed Sampling**: Use standard OpenTelemetry or ADOT sampling configuration ## Installation @@ -157,7 +158,8 @@ def handler(event: dict, context: DurableContext) -> dict: return result ``` -That's it. The plugin handles TracerProvider setup, deterministic ID generation, and span lifecycle internally. +The ADOT layer supplies the global `TracerProvider`; the plugin handles +deterministic ID generation and span lifecycle. ### 4. Grant Permissions @@ -182,15 +184,14 @@ See the [ADOT sampling configuration](https://aws-otel.github.io/docs/getting-st from aws_durable_execution_sdk_python_otel import ( InvocationOtelPlugin, OtelPluginConfig, + ProviderSource, xray_context_extractor, ) plugin = InvocationOtelPlugin( OtelPluginConfig( - # Provide your own TracerProvider if you already have one configured. - # When omitted, an OTLP provider is auto-configured (like ExecutionOtelPlugin); - # set use_default_tracer_provider=True to use the global (e.g. ADOT) provider. - tracer_provider=None, + # Use the global provider configured by ADOT (the default). + provider_source=ProviderSource.GLOBAL, # Use a custom context extractor (default: xray_context_extractor). context_extractor=xray_context_extractor, # Custom instrumentation scope name @@ -270,17 +271,19 @@ The main plugin class. Implements `DurableInstrumentationPlugin` from `aws_durab ```python InvocationOtelPlugin( OtelPluginConfig( + provider_source=ProviderSource.GLOBAL, tracer_provider=None, context_extractor=None, instrument_name="aws-durable-execution-sdk-python", enrich_logger=True, workflow_span_name="Workflow", - # ...and the rest of OtelPluginConfig (use_default_tracer_provider, - # enable_http_instrumentation, exporter_config, propagators). ) ) ``` +Set `provider_source=ProviderSource.EXPLICIT` and pass `tracer_provider=...` +when the application owns the OpenTelemetry SDK provider. + ### `DeterministicIdGenerator` A custom OpenTelemetry `IdGenerator` that produces reproducible trace and span IDs from execution metadata. Exported for advanced use cases. @@ -309,7 +312,6 @@ setups. - `aws-durable-execution-sdk-python` >= 1.8.0 - `opentelemetry-api` >= 1.20.0 - `opentelemetry-sdk` >= 1.20.0 -- `opentelemetry-exporter-otlp` ## License diff --git a/packages/aws-durable-execution-sdk-python-otel/pyproject.toml b/packages/aws-durable-execution-sdk-python-otel/pyproject.toml index e5700d8b..106cc248 100644 --- a/packages/aws-durable-execution-sdk-python-otel/pyproject.toml +++ b/packages/aws-durable-execution-sdk-python-otel/pyproject.toml @@ -25,7 +25,6 @@ dependencies = [ "aws-durable-execution-sdk-python>=1.8.0", "opentelemetry-api>=1.20.0", "opentelemetry-sdk>=1.20.0", - "opentelemetry-exporter-otlp", "opentelemetry-propagator-aws-xray", ] @@ -34,12 +33,10 @@ otel-invocation = "aws_durable_execution_sdk_python_otel.plugin_provider:INVOCAT otel-execution = "aws_durable_execution_sdk_python_otel.plugin_provider:EXECUTION_OTEL_PLUGIN_PROVIDER" [project.optional-dependencies] -# Instrumentation used by ExecutionOtelPlugin's auto-configured provider path. -# Kept optional so the InvocationOtelPlugin (ADOT / global provider) install -# stays lean; the instrumentations module degrades gracefully when absent. +# Optional AWS SDK instrumentation for the global provider path. The +# instrumentations module degrades gracefully when it is absent. instrumentation = [ "opentelemetry-instrumentation-botocore", - "opentelemetry-instrumentation-urllib3", ] [project.urls] diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/__init__.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/__init__.py index b9f1257e..6547aade 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/__init__.py +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/__init__.py @@ -16,7 +16,6 @@ ) from aws_durable_execution_sdk_python_otel.otel_plugin_config import ( OtelPluginConfig, - ExporterConfig, ProviderSource, ) from aws_durable_execution_sdk_python_otel.instrumentations import ( @@ -41,7 +40,6 @@ "DeterministicIdGenerator", "ExecutionOtelPlugin", "OtelPluginConfig", - "ExporterConfig", "InvocationOtelPlugin", "OtelContextLogFilter", "ProviderResult", diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/deterministic_id_generator.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/deterministic_id_generator.py index 1a465749..84f6ba36 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/deterministic_id_generator.py +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/deterministic_id_generator.py @@ -4,8 +4,9 @@ import contextvars import hashlib -import os -import re +from collections.abc import Iterator +from contextlib import contextmanager +from dataclasses import dataclass from datetime import UTC, datetime from typing import TYPE_CHECKING @@ -13,59 +14,22 @@ if TYPE_CHECKING: - from opentelemetry.sdk.trace import TracerProvider + from opentelemetry.sdk.trace import Tracer as SdkTracer -HASHED_ID_PATTERN = re.compile(r"^[0-9a-f]{16}$") - -# Scoping the pending span ID to the execution context ensures concurrent -# operations cannot consume each other's deterministic span ID. -_next_span_id: contextvars.ContextVar[int | None] = contextvars.ContextVar( - "next_span_id", default=None -) - - -def _parse_xray_root_trace_id(trace_header: str | None) -> str | None: - """Parse the Root trace ID from an X-Ray trace header string. - - The header format is: - Root=1-<8 hex>-<24 hex>;Parent=<16 hex>;Sampled=0|1 - - Returns the root value (e.g. "1-5759e988-bd862e3fe1be46a994272793") - or None if the header is missing or malformed. - """ - if not trace_header: - return None - match = re.search(r"Root=(1-[0-9a-fA-F]{8}-[0-9a-fA-F]{24})", trace_header) - return match.group(1) if match else None - - -def _xray_trace_id_to_otel(xray_trace_id: str) -> int: - """Convert an X-Ray trace ID to the W3C/OpenTelemetry 32-char hex format. - - X-Ray format: "1-<8hex>-<24hex>" (36 chars with prefix and dashes) - OTel format: "<8hex><24hex>" (32 lowercase hex chars) - """ - otel_id = xray_trace_id.replace("1-", "", 1).replace("-", "").lower() - return int(otel_id, 16) +@dataclass(frozen=True) +class _IdOverride: + trace_id: int | None + span_id: int | None def _to_otel_trace_id(execution_arn: str, start_timestamp: datetime | None) -> int: - """Build an OTel-compatible trace ID (128 bits) + """Build a deterministic OTel-compatible execution trace ID (128 bits). - First attempts to read the trace ID from the _X_AMZN_TRACE_ID environment - variable that Lambda populates on each invocation. This ties the durable - execution spans to the same trace that X-Ray is already tracking. - - Falls back to generating a deterministic trace ID from the execution ARN - and timestamp when the environment variable is not set (e.g. in tests or - non-Lambda environments). + The ID is independent of ambient Lambda or X-Ray trace context so the + parentless Workflow span remains the only root of the durable execution + trace. Invocation spans inherit ambient context separately. """ - env_trace_id = _parse_xray_root_trace_id(os.environ.get("_X_AMZN_TRACE_ID")) - if env_trace_id: - return _xray_trace_id_to_otel(env_trace_id) - - # Fallback: deterministic ID from execution ARN + timestamp time_part = format(int((start_timestamp or datetime.now(UTC)).timestamp()), "08x") hash_part = hashlib.blake2b(execution_arn.encode()).hexdigest()[:24] # noqa: S324 return int(f"{time_part}{hash_part}", 16) @@ -112,13 +76,12 @@ def derive_workflow_span_id(durable_execution_arn: str) -> int: class DeterministicIdGenerator(RandomIdGenerator): - """An ID generator that produces deterministic span IDs when a pending - operation ID is set, and falls back to the provided generator otherwise. + """An ID generator with invocation-scoped deterministic ID overrides. - Trace IDs are deterministic when an execution ARN is set, ensuring all - invocations of the same durable execution share a single trace. When no - deterministic ID is available, generation is delegated to the fallback - generator (the tracer provider's original ID generator by default). + Deterministic IDs are active only inside :meth:`use_ids`. All other + generation is delegated to the fallback generator. The override is stored + in a context variable so concurrent threads and async tasks cannot consume + or overwrite each other's IDs. Trace IDs embed a real timestamp so they satisfy the X-Ray format requirement (first 8 hex chars = Unix epoch seconds). @@ -129,54 +92,60 @@ class DeterministicIdGenerator(RandomIdGenerator): """ def __init__(self, fallback_id_generator: IdGenerator | None = None) -> None: - self._execution_trace_id: int | None = None self._fallback_id_generator = fallback_id_generator or RandomIdGenerator() + self._id_override: contextvars.ContextVar[_IdOverride | None] = ( + contextvars.ContextVar("durable_execution_id_override", default=None) + ) @classmethod - def install_on_provider(cls, provider: TracerProvider) -> DeterministicIdGenerator: - """Return the provider's deterministic generator, installing one if needed. + def install_on_tracer(cls, tracer: SdkTracer) -> DeterministicIdGenerator: + """Return the tracer's deterministic generator, installing one if needed. - OpenTelemetry tracers capture the provider's ID generator when they are - created. Reusing an installed generator ensures multiple plugin instances - with the same instrumentation scope configure the generator referenced by - the provider's cached tracer. + Installing on the plugin's tracer keeps unrelated instrumentation scopes + on the provider's original generator. Reusing an installed generator also + supports SDK versions that cache tracers by instrumentation scope. """ - current_generator = provider.id_generator + current_generator = tracer.id_generator if isinstance(current_generator, cls): return current_generator generator = cls(fallback_id_generator=current_generator) - provider.id_generator = generator + tracer.id_generator = generator return generator - def set_next_span_id(self, span_id: int | None) -> None: - """Set the operation ID to use for the next span's ID. - - After one span is created, it resets to random. - """ - _next_span_id.set(span_id) - - def set_trace_id( - self, execution_arn: str, start_timestamp: datetime | None - ) -> None: - """Compute and cache the deterministic trace ID for this execution. - - Args: - execution_arn: The durable execution ARN (used for the hash portion). - start_timestamp: start time of invocation - """ - self._execution_trace_id = _to_otel_trace_id(execution_arn, start_timestamp) + @contextmanager + def use_ids(self, *, trace_id: int | None, span_id: int | None) -> Iterator[None]: + """Temporarily override IDs generated in the current execution context.""" + token = self._id_override.set(_IdOverride(trace_id, span_id)) + try: + yield + finally: + self._id_override.reset(token) def generate_trace_id(self) -> int: """Generate a 128-bit trace ID.""" - return ( - self._execution_trace_id or self._fallback_id_generator.generate_trace_id() - ) + override = self._id_override.get() + if override is not None and override.trace_id is not None: + return override.trace_id + return self._fallback_id_generator.generate_trace_id() def generate_span_id(self) -> int: """Generate a 64-bit span ID.""" - span_id = _next_span_id.get() - # Consume once: the deterministic ID applies only to the next span - # created in this context; subsequent spans fall back to random. - _next_span_id.set(None) - return span_id or self._fallback_id_generator.generate_span_id() + override = self._id_override.get() + if override is not None and override.span_id is not None: + span_id = override.span_id + # Consume before returning so a re-entrant call in the same span + # creation falls back instead of reusing the deterministic ID. + self._id_override.set(_IdOverride(override.trace_id, None)) + return span_id + return self._fallback_id_generator.generate_span_id() + + def is_trace_id_random(self) -> bool: + """Report whether the current trace ID is randomly generated.""" + override = self._id_override.get() + if override is not None and override.trace_id is not None: + return False + fallback_method = getattr( + self._fallback_id_generator, "is_trace_id_random", None + ) + return bool(fallback_method()) if fallback_method is not None else False diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py index bc829db6..5967e179 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py @@ -2,13 +2,14 @@ The :class:`ExecutionOtelPlugin` produces the deterministic span hierarchy - Workflow -> Invocation -> Operation -> Attempt + Workflow -> Operation -> Attempt that stitches a single trace across every Lambda invocation of one durable execution. The Workflow span is the root (created in an empty context so it never has a parent) and is exported exactly once, when the execution reaches a terminal status. Operations are parented under the Workflow span (or their -parent operation) and *linked* to the current Invocation span. +parent operation) and *linked* to the current Invocation span. The Invocation +span belongs to the ambient Lambda trace instead of the Workflow trace. This is the Python adaptation of the JS ``ExecutionOtelPlugin`` from aws-durable-execution-sdk-js#729. Because the Python plugin interface differs @@ -42,6 +43,7 @@ from opentelemetry import context as otel_context from opentelemetry import trace from opentelemetry.context import Context +from opentelemetry.sdk.trace import Tracer as SdkTracer from opentelemetry.trace import ( Link, Span, @@ -57,6 +59,7 @@ ) from aws_durable_execution_sdk_python_otel.deterministic_id_generator import ( DeterministicIdGenerator, + _to_otel_trace_id, derive_workflow_span_id, operation_id_to_span_id, ) @@ -104,49 +107,51 @@ def __init__(self, config: OtelPluginConfig | None = None) -> None: ) self._workflow_span_name = self._config.workflow_span_name - self._id_generator = DeterministicIdGenerator() - result = create_tracer_provider( - self._config, - id_generator=self._id_generator, - ) + result = create_tracer_provider(self._config) self._provider = result.tracer_provider # GLOBAL (ADOT) mode parents the Invocation span to the ambient Lambda # invocation span instead of the Workflow span (see # _start_invocation_span). self._provider_source = result.source - # Deterministic stitching requires an SDK provider exposing id_generator. - from opentelemetry.sdk.trace import TracerProvider as SdkTracerProvider - - if isinstance(self._provider, SdkTracerProvider): - self._id_generator = DeterministicIdGenerator.install_on_provider( - self._provider - ) - else: - logger.warning( - "ExecutionOtelPlugin expected an SDK TracerProvider but got %s; " - "spans will not use deterministic IDs.", - type(self._provider).__name__, - ) - self._tracer: Tracer = self._provider.get_tracer(self._config.instrument_name) + self._id_generator = DeterministicIdGenerator() + self._bind_sdk_tracer() try: - register_standalone_instrumentations(self._config, result) + register_standalone_instrumentations(result) except Exception: logger.exception("Failed to register standalone instrumentations") # Per-invocation state. self._execution_arn = "" + self._execution_trace_id: int | None = None self._extracted_context: Context | None = None self._workflow_span: Span | None = None self._invocation_span: Span | None = None self._operation_spans: dict[str, Span] = {} self._lock = threading.RLock() + self._tracing_enabled = False if self._config.enrich_logger: install_log_filter(self) + def _bind_sdk_tracer(self) -> bool: + """Bind to an SDK tracer, retrying a deferred global provider.""" + tracer = self._tracer + if not isinstance(tracer, SdkTracer): + if self._provider_source is ProviderSource.GLOBAL: + self._provider = trace.get_tracer_provider() + tracer = self._provider.get_tracer(self._config.instrument_name) + self._tracer = tracer + if not isinstance(tracer, SdkTracer): + return False + + # Deterministic stitching is scoped to this instrumentation tracer so + # unrelated tracers on the same provider keep their original generator. + self._id_generator = DeterministicIdGenerator.install_on_tracer(tracer) + return True + # ------------------------------------------------------------------ # Span registry helpers # ------------------------------------------------------------------ @@ -199,18 +204,40 @@ def _resolve_parent(self, parent_id: str | None) -> Span | None: return existing return self._workflow_span + def _invocation_parent_context(self) -> Context: + """Return the active ambient context, then extracted upstream context.""" + ambient_context = otel_context.get_current() + ambient_span_context = trace.get_current_span( + ambient_context + ).get_span_context() + if ambient_span_context.is_valid: + return ambient_context + return self._extracted_context or ambient_context + # ------------------------------------------------------------------ # Invocation lifecycle # ------------------------------------------------------------------ def on_invocation_start(self, info: InvocationStartInfo) -> None: logger.debug("Durable invocation started: %s", info) + self._reset_state() + self._tracing_enabled = self._bind_sdk_tracer() + if not self._tracing_enabled: + logger.warning( + "ExecutionOtelPlugin expected an SDK Tracer at invocation start " + "but got %s; telemetry is disabled for this invocation. Ensure " + "the OpenTelemetry SDK is configured before invocation start.", + type(self._tracer).__name__, + ) + return + self._execution_arn = info.execution_arn or "" + self._execution_trace_id = _to_otel_trace_id( + self._execution_arn, info.execution_start_time + ) self._extracted_context = self._context_extractor(info) - self._id_generator.set_trace_id(self._execution_arn, info.execution_start_time) self._start_workflow_span(info) - # Create the Invocation span in both modes. In default-provider mode it - # is parented to the ambient Lambda invocation span. + # Keep the invocation in the ambient Lambda trace in both provider modes. self._start_invocation_span(info) # Make the Workflow span the active span so auto-instrumented spans @@ -224,56 +251,40 @@ def _start_workflow_span(self, info: InvocationStartInfo) -> None: if not self._execution_arn: logger.warning("No execution ARN; skipping Workflow span creation") return - self._id_generator.set_next_span_id( - derive_workflow_span_id(self._execution_arn) - ) start_time = _to_otel_timestamp( info.execution_start_time ) or _to_otel_timestamp(datetime.datetime.now(datetime.UTC)) # Empty context => root span with no parent. - self._workflow_span = self._tracer.start_span( - name=self._workflow_span_name, - kind=SpanKind.INTERNAL, - attributes={"durable.execution.arn": self._execution_arn}, - start_time=start_time, - context=Context(), - ) + with self._id_generator.use_ids( + trace_id=self._execution_trace_id, + span_id=derive_workflow_span_id(self._execution_arn), + ): + self._workflow_span = self._tracer.start_span( + name=self._workflow_span_name, + kind=SpanKind.INTERNAL, + attributes={"durable.execution.arn": self._execution_arn}, + start_time=start_time, + context=Context(), + ) def _start_invocation_span(self, info: InvocationStartInfo) -> None: - self._id_generator.set_next_span_id(None) - attributes: dict[str, Any] - if self._provider_source is ProviderSource.GLOBAL: - # Default-provider mode: parent the Invocation span to the ambient - # Lambda invocation span (from the ADOT layer or other - # auto-instrumentation), which is still the active context here (the - # Workflow span is created with an empty context and not yet - # attached). Lambda semantic attributes belong to that ambient span, - # so carry only durable correlation attributes here. - parent_ctx = otel_context.get_current() - attributes = { - "durable.execution.arn": self._execution_arn, - "durable.invocation.first": info.is_first_invocation, - } - else: - if self._workflow_span is None: - return - parent_ctx = trace.set_span_in_context( - self._workflow_span, self._extracted_context - ) - attributes = { - "durable.execution.arn": self._execution_arn, - "durable.invocation.first": info.is_first_invocation, - } self._invocation_span = self._tracer.start_span( name="Invocation", kind=SpanKind.INTERNAL, - attributes=attributes, - context=parent_ctx, + attributes={ + "durable.execution.arn": self._execution_arn, + "durable.invocation.first": info.is_first_invocation, + }, + context=self._invocation_parent_context(), ) self._set_span(_INVOCATION_KEY, self._invocation_span) def on_invocation_end(self, info: InvocationEndInfo) -> None: logger.debug("Durable invocation ended: %s", info) + if not self._tracing_enabled: + self._reset_state() + return + # Operation spans still open here belong to operations that suspended # (e.g. PENDING/RETRYING) rather than completed this invocation. They are # ended only by on_operation_end; drop the references without ending them @@ -330,17 +341,21 @@ def on_invocation_end(self, info: InvocationEndInfo) -> None: def _reset_state(self) -> None: self._execution_arn = "" + self._execution_trace_id = None self._extracted_context = None self._workflow_span = None self._invocation_span = None with self._lock: self._operation_spans = {} + self._tracing_enabled = False # ------------------------------------------------------------------ # Operation lifecycle # ------------------------------------------------------------------ def on_operation_start(self, info: OperationStartInfo) -> None: logger.debug("Durable operation started: %s", info) + if not self._tracing_enabled: + return if info.operation_type is OperationType.CONTEXT: return # tracked via on_user_function_start parent = self._resolve_parent(info.parent_id) @@ -354,6 +369,8 @@ def on_operation_start(self, info: OperationStartInfo) -> None: def on_operation_end(self, info: OperationEndInfo) -> None: logger.debug("Durable operation ended: %s", info) + if not self._tracing_enabled: + return span = self._get_span(info.operation_id) if span is None: # Cross-invocation stitching: operation started in a prior @@ -399,27 +416,26 @@ def _start_span( key = span_key if span_key is not None else operation_id with self._lock: links = self._build_invocation_links() - if deterministic: - # Operation spans always use the deterministic logical-operation - # span ID so a suspended-then-completed operation exports a - # single span (on completion) with a stable ID across invocations. - self._id_generator.set_next_span_id( - operation_id_to_span_id(self._execution_arn, operation_id) - ) - else: - self._id_generator.set_next_span_id(None) + span_id = ( + operation_id_to_span_id(self._execution_arn, operation_id) + if deterministic + else None + ) if parent is None: parent_ctx = self._extracted_context or Context() else: parent_ctx = trace.set_span_in_context(parent, self._extracted_context) - span = self._tracer.start_span( - name=name, - attributes=self._operation_attributes(info), - start_time=_to_otel_timestamp(start_time), - context=parent_ctx, - links=links, - ) + with self._id_generator.use_ids( + trace_id=self._execution_trace_id, span_id=span_id + ): + span = self._tracer.start_span( + name=name, + attributes=self._operation_attributes(info), + start_time=_to_otel_timestamp(start_time), + context=parent_ctx, + links=links, + ) self._operation_spans[key] = span return span @@ -428,6 +444,8 @@ def _start_span( # ------------------------------------------------------------------ def on_user_function_start(self, info: UserFunctionStartInfo) -> None: logger.debug("Durable user function started: %s", info) + if not self._tracing_enabled: + return if info.operation_type not in (OperationType.CONTEXT, OperationType.STEP): raise RuntimeError( "on_user_function_start only supports CONTEXT and STEP operations" @@ -459,6 +477,8 @@ def on_user_function_start(self, info: UserFunctionStartInfo) -> None: def on_user_function_end(self, info: UserFunctionEndInfo) -> None: logger.debug("Durable user function ended: %s", info) + if not self._tracing_enabled: + return if info.operation_type not in (OperationType.CONTEXT, OperationType.STEP): raise RuntimeError( "on_user_function_end only supports CONTEXT and STEP operations" diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/instrumentations.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/instrumentations.py index 5629ae56..e4b887bd 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/instrumentations.py +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/instrumentations.py @@ -5,45 +5,28 @@ * A custom (explicit) provider skips ALL instrumentation registration. * When the global provider is in use (``ProviderSource.GLOBAL``), only the AWS SDK instrumentation is registered (not HTTP). -* When the plugin owns an auto-configured provider, both AWS SDK and (optionally) - HTTP instrumentation are registered against that provider. - -The JS SDK uses ``AwsInstrumentation`` (AWS SDK v3) and ``HttpInstrumentation``. -The Python equivalents are ``BotocoreInstrumentor`` (boto3/botocore is the AWS -SDK for Python) and ``URLLib3Instrumentor`` (botocore's HTTP transport). Both -instrumentation packages are optional imports: when a package is not installed -the registration is skipped with a warning rather than raising, so the module -stays import-safe. + +The JS SDK uses ``AwsInstrumentation`` (AWS SDK v3). The Python equivalent is +``BotocoreInstrumentor`` because boto3/botocore is the AWS SDK for Python. The +instrumentation package is an optional import: when it is not installed, +registration is skipped with a warning rather than raising, so the module stays +import-safe. """ from __future__ import annotations import logging -import os -from typing import TYPE_CHECKING, Any +from typing import TYPE_CHECKING from aws_durable_execution_sdk_python_otel.otel_plugin_config import ProviderSource if TYPE_CHECKING: - from aws_durable_execution_sdk_python_otel.otel_plugin_config import ( - OtelPluginConfig, - ) from aws_durable_execution_sdk_python_otel.provider import ProviderResult logger = logging.getLogger(__name__) -_LOCAL_HOSTS = {"127.0.0.1", "localhost"} - - -def _runtime_api_host() -> str | None: - """Return the Lambda runtime API host (portion before ':') if set.""" - runtime_api = os.environ.get("AWS_LAMBDA_RUNTIME_API") - if not runtime_api: - return None - return runtime_api.split(":", 1)[0] - def _register_aws_instrumentation(tracer_provider: object | None) -> None: """Register AWS SDK (botocore) instrumentation, if the package is available.""" @@ -64,47 +47,10 @@ def _register_aws_instrumentation(tracer_provider: object | None) -> None: instrumentor.instrument(**kwargs) -def _register_http_instrumentation(tracer_provider: object | None) -> None: - """Register HTTP (urllib3) instrumentation with local/runtime suppression.""" - try: - from opentelemetry.instrumentation.urllib3 import URLLib3Instrumentor - except ImportError: - logger.warning( - "opentelemetry-instrumentation-urllib3 is not installed; outbound " - "HTTP calls will not be traced. Install it to enable HTTP " - "instrumentation." - ) - return - - suppressed_hosts = set(_LOCAL_HOSTS) - runtime_host = _runtime_api_host() - if runtime_host: - suppressed_hosts.add(runtime_host) - - def request_hook(span, pool, request_info) -> None: # noqa: ANN001 - # Suppress spans to the loopback collector and the Lambda runtime API by - # ending them immediately with no recording. urllib3 does - # not expose a pre-create filter, so we no-op the created span instead. - host = getattr(pool, "host", None) - if host in suppressed_hosts and span is not None and span.is_recording(): - span.set_attribute("durable.instrumentation.suppressed", True) - - instrumentor = URLLib3Instrumentor() - if not instrumentor.is_instrumented_by_opentelemetry: - kwargs: dict[str, Any] = {"request_hook": request_hook} - if tracer_provider is not None: - kwargs["tracer_provider"] = tracer_provider - instrumentor.instrument(**kwargs) - - -def register_standalone_instrumentations( - config: OtelPluginConfig, - result: ProviderResult, -) -> None: - """Register AWS SDK and HTTP instrumentations per the resolved source. +def register_standalone_instrumentations(result: ProviderResult) -> None: + """Register AWS SDK instrumentation per the resolved source. Args: - config: Shared plugin configuration. result: The resolved provider and its :class:`ProviderSource`. """ if result.source is ProviderSource.EXPLICIT: @@ -115,9 +61,3 @@ def register_standalone_instrumentations( # Global provider: register AWS instrumentation only. _register_aws_instrumentation(None) return - - # AUTO_OTLP: auto-configured, plugin-owned provider -> AWS SDK always; HTTP - # unless explicitly disabled. - _register_aws_instrumentation(result.tracer_provider) - if config.enable_http_instrumentation: - _register_http_instrumentation(result.tracer_provider) diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py index 4f42ca32..cd52e346 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py @@ -23,14 +23,13 @@ ) from opentelemetry import context, trace from opentelemetry.context import Context -from opentelemetry.sdk.trace import TracerProvider as SdkTracerProvider +from opentelemetry.sdk.trace import Tracer as SdkTracer from opentelemetry.trace import ( Link, Span, SpanContext, SpanKind, StatusCode, - TraceFlags, Tracer, ) @@ -40,12 +39,14 @@ ) from aws_durable_execution_sdk_python_otel.deterministic_id_generator import ( DeterministicIdGenerator, + _to_otel_trace_id, derive_workflow_span_id, operation_id_to_span_id, ) from aws_durable_execution_sdk_python_otel.log_filter import install_log_filter from aws_durable_execution_sdk_python_otel.otel_plugin_config import ( OtelPluginConfig, + ProviderSource, ) from aws_durable_execution_sdk_python_otel.provider import create_tracer_provider from aws_durable_execution_sdk_python_otel.instrumentations import ( @@ -73,15 +74,15 @@ class InvocationOtelPlugin(DurableInstrumentationPlugin): """OpenTelemetry instrumentation plugin for durable executions. The plugin creates spans for Lambda invocations, durable operations, and - user-function attempts. Trace IDs are derived from the durable execution ARN - and execution start time so each replay or resumed invocation contributes to - the same trace. + user-function attempts. The Workflow trace ID is derived from the durable + execution ARN and start time. Invocation spans inherit ambient or extracted + upstream context, and operation spans are correlated with the Workflow by a + span link. Operation IDs are converted into deterministic span IDs. The first observed span for an operation uses that deterministic ID; later continuation spans - use newly generated span IDs and link back to the deterministic span ID so - trace viewers can relate retries and cross-invocation completions to the - original logical operation. + use newly generated span IDs. Operation attributes and links to the Workflow + span provide execution-scoped correlation across invocations. Args: config: Shared plugin configuration (the same OtelPluginConfig accepted @@ -89,9 +90,7 @@ class InvocationOtelPlugin(DurableInstrumentationPlugin): extractor, "Workflow" span name, log enrichment on). Like ExecutionOtelPlugin and the JS SDK plugins, the default ``provider_source`` is ``GLOBAL``: the plugin uses the globally - configured tracer provider (e.g. the ADOT Lambda layer). Set - ``provider_source=ProviderSource.AUTO_OTLP`` on the config to have - the plugin build and own an auto-configured OTLP provider instead. + configured tracer provider (e.g. the ADOT Lambda layer). """ DEFAULT_INSTRUMENT_NAME = "aws-durable-execution-sdk-python" @@ -100,17 +99,14 @@ def __init__(self, config: OtelPluginConfig | None = None) -> None: """Initialize the plugin from a shared OtelPluginConfig. Accepts the same OtelPluginConfig as ExecutionOtelPlugin so both plugins - share one configuration surface (context extractor, instrumentation - name, provider selection, exporter/propagator settings, log - enrichment). Like ExecutionOtelPlugin and the JS SDK plugins, the - default ``provider_source`` is ``GLOBAL``: it uses the globally - configured (e.g. ADOT) provider. Pass - ``provider_source=ProviderSource.AUTO_OTLP`` to have the plugin build - and own an auto-configured OTLP provider instead. - - The tracer provider is configured with this plugin's deterministic ID - generator so spans for a durable execution share stable trace and - logical operation identifiers. + share one configuration surface (context extractor, instrumentation name, + provider selection, and log enrichment). Like ExecutionOtelPlugin and the + JS SDK plugins, the default ``provider_source`` is ``GLOBAL``: it uses the + globally configured (e.g. ADOT) provider. + + The plugin tracer is configured with a scoped deterministic ID generator + so durable spans share stable identifiers without changing unrelated + instrumentation scopes on the same provider. When ``enrich_logger`` is enabled (default), the plugin installs a logging filter that stamps the active OTel trace context onto every @@ -125,49 +121,28 @@ def __init__(self, config: OtelPluginConfig | None = None) -> None: # Like ExecutionOtelPlugin (and the JS SDK plugins), InvocationOtelPlugin # defaults to provider_source=GLOBAL (the globally configured, e.g. ADOT, - # provider); set provider_source=ProviderSource.AUTO_OTLP on the config - # to build and own an auto-configured OTLP provider instead. - self._id_generator = DeterministicIdGenerator() - result = create_tracer_provider( - self._config, - id_generator=self._id_generator, - ) + # provider). + result = create_tracer_provider(self._config) self._provider = result.tracer_provider - - # Deterministic trace stitching requires the SDK TracerProvider, which - # exposes id_generator/sampler. The API's default ProxyTracerProvider - # (returned before an SDK provider is configured) does not. Rather than - # fail the invocation over an observability concern, warn and continue: - # the proxy's tracer is effectively a no-op (and auto-delegates if an SDK - # provider is configured later). In a Lambda OTel/ADOT deployment the - # layer configures a real SDK provider before the handler imports. - if isinstance(self._provider, SdkTracerProvider): - self._id_generator = DeterministicIdGenerator.install_on_provider( - self._provider - ) - else: - logger.warning( - "InvocationOtelPlugin expected an SDK TracerProvider " - "(opentelemetry.sdk.trace.TracerProvider) but got %s. Spans will " - "not use deterministic IDs. " - "Ensure the OpenTelemetry SDK is configured (e.g. via the ADOT " - "Lambda layer) or pass an explicit tracer_provider.", - type(self._provider).__name__, - ) + self._provider_source = result.source self._tracer: Tracer = self._provider.get_tracer(self._config.instrument_name) + self._id_generator = DeterministicIdGenerator() + self._bind_sdk_tracer() try: - register_standalone_instrumentations(self._config, result) + register_standalone_instrumentations(result) except Exception: logger.exception("Failed to register standalone instrumentations") # per invocation status: self._execution_arn = "" + self._execution_trace_id: int | None = None self._extracted_context: Context | None = None self._workflow_span: Span | None = None # Maps operation ID (None for root) to the active span. self._operation_spans: dict[str | None, Span] = {} self._operation_spans_lock = threading.RLock() + self._tracing_enabled = False if self._enrich_logger: # Install the root-logger filter so every log record is stamped with @@ -176,6 +151,20 @@ def __init__(self, config: OtelPluginConfig | None = None) -> None: # plugin is constructed), so the handlers are available here. install_log_filter(self) + def _bind_sdk_tracer(self) -> bool: + """Bind to an SDK tracer, retrying a deferred global provider.""" + tracer = self._tracer + if not isinstance(tracer, SdkTracer): + if self._provider_source is ProviderSource.GLOBAL: + self._provider = trace.get_tracer_provider() + tracer = self._provider.get_tracer(self._config.instrument_name) + self._tracer = tracer + if not isinstance(tracer, SdkTracer): + return False + + self._id_generator = DeterministicIdGenerator.install_on_tracer(tracer) + return True + def _set_span(self, operation_id: str | None, span: Span) -> None: """Register the active span for an operation ID.""" with self._operation_spans_lock: @@ -246,6 +235,16 @@ def _resolve_parent_span(self, parent_id: str | None = None) -> Span: raise ValueError("No parent span found") + def _invocation_parent_context(self) -> Context: + """Return the active ambient context, then extracted upstream context.""" + ambient_context = context.get_current() + ambient_span_context = trace.get_current_span( + ambient_context + ).get_span_context() + if ambient_span_context.is_valid: + return ambient_context + return self._extracted_context or ambient_context + def _start_span( self, operation_id: str | None, @@ -266,10 +265,9 @@ def _start_span( attributes: Span attributes. start_time: Optional durable start timestamp. parent_span: Active parent span. When omitted, the extracted - upstream context is used as the parent. + ambient or upstream context is used as the parent. existed: Whether the logical operation already had a previous span. - Continuation spans link back to the deterministic span ID for - the operation while using a fresh generated span ID. + Continuation spans use a fresh generated span ID. span_key: Optional registry key. Defaults to ``operation_id``. deterministic_span_id: Whether to use the deterministic operation span ID. Attempt spans set this to ``False`` so they can be @@ -286,28 +284,13 @@ def _start_span( ) registry_key = span_key if span_key is not None else operation_id with self._operation_spans_lock: - if not deterministic_span_id: + links: list[Link] + if not deterministic_span_id or existed: links = [] - self._id_generator.set_next_span_id(None) - elif existed: - if not operation_id: - raise ValueError("operation id is required") - span_id = operation_id_to_span_id(self._execution_arn, operation_id) - links = [ - Link( - context=SpanContext( - trace_id=self._id_generator.generate_trace_id(), - span_id=span_id, - is_remote=False, - trace_flags=TraceFlags(TraceFlags.SAMPLED), - ) - ) - ] - self._id_generator.set_next_span_id(None) + span_id = None else: links = [] - - self._id_generator.set_next_span_id( + span_id = ( operation_id_to_span_id(self._execution_arn, operation_id) if operation_id else None @@ -319,20 +302,21 @@ def _start_span( if workflow_ctx and workflow_ctx.is_valid: links = [*links, Link(context=workflow_ctx)] if parent_span is None: - # root span - parent_context = self._extracted_context + parent_context = self._invocation_parent_context() else: parent_context = trace.set_span_in_context( parent_span, self._extracted_context ) - span = self._tracer.start_span( - name=name, - kind=SpanKind.INTERNAL, - attributes=attributes, - start_time=_to_otel_timestamp(start_time), - context=parent_context, - links=links, - ) + trace_id = self._execution_trace_id if operation_id is not None else None + with self._id_generator.use_ids(trace_id=trace_id, span_id=span_id): + span = self._tracer.start_span( + name=name, + kind=SpanKind.INTERNAL, + attributes=attributes, + start_time=_to_otel_timestamp(start_time), + context=parent_context, + links=links, + ) self._operation_spans[registry_key] = span logger.debug("Started OTel span: %s", span) @@ -364,9 +348,22 @@ def _end_span( def on_invocation_start(self, info: InvocationStartInfo) -> None: """Called at the start of each invocation. Creates the invocation span.""" logger.debug("Durable invocation started: %s", info) + self._reset_state() + self._tracing_enabled = self._bind_sdk_tracer() + if not self._tracing_enabled: + logger.warning( + "InvocationOtelPlugin expected an SDK Tracer at invocation start " + "but got %s; telemetry is disabled for this invocation. Ensure " + "the OpenTelemetry SDK is configured before invocation start.", + type(self._tracer).__name__, + ) + return + self._execution_arn = info.execution_arn or "" + self._execution_trace_id = _to_otel_trace_id( + self._execution_arn, info.execution_start_time + ) self._extracted_context = self._context_extractor(info) - self._id_generator.set_trace_id(self._execution_arn, info.execution_start_time) self._start_workflow_span(info) @@ -391,22 +388,27 @@ def _start_workflow_span(self, info: InvocationStartInfo) -> None: if not self._execution_arn: logger.warning("No execution ARN; skipping Workflow span creation") return - self._id_generator.set_next_span_id( - derive_workflow_span_id(self._execution_arn) - ) # Empty context => root span with no parent. _to_otel_timestamp falls # back to now() when execution_start_time is None. - self._workflow_span = self._tracer.start_span( - name=self._workflow_span_name, - kind=SpanKind.INTERNAL, - attributes={"durable.execution.arn": self._execution_arn}, - start_time=_to_otel_timestamp(info.execution_start_time), - context=Context(), - ) + with self._id_generator.use_ids( + trace_id=self._execution_trace_id, + span_id=derive_workflow_span_id(self._execution_arn), + ): + self._workflow_span = self._tracer.start_span( + name=self._workflow_span_name, + kind=SpanKind.INTERNAL, + attributes={"durable.execution.arn": self._execution_arn}, + start_time=_to_otel_timestamp(info.execution_start_time), + context=Context(), + ) def on_invocation_end(self, info: InvocationEndInfo) -> None: """Called at the end of each invocation. Ends the invocation span and flushes.""" logger.debug("Durable invocation ended: %s", info) + if not self._tracing_enabled: + self._reset_state() + return + # Spans are registered parent-first, so close pending spans in reverse # order to keep every child contained within its parent. with self._operation_spans_lock: @@ -453,20 +455,27 @@ def on_invocation_end(self, info: InvocationEndInfo) -> None: self._workflow_span.set_status(StatusCode.OK) self._workflow_span.end() - # Clear all per-invocation state to prevent leaks across warm Lambda reuses + self._reset_state() + + # Flush before Lambda freeze + if hasattr(self._provider, "force_flush"): + self._provider.force_flush() + + def _reset_state(self) -> None: + """Clear per-invocation state for warm Lambda environment reuse.""" self._execution_arn = "" + self._execution_trace_id = None self._extracted_context = None self._workflow_span = None with self._operation_spans_lock: self._operation_spans = {} - - # Flush before Lambda freeze - if hasattr(self._provider, "force_flush"): - self._provider.force_flush() + self._tracing_enabled = False def on_operation_start(self, info: OperationStartInfo) -> None: """Called when an operation begins. Creates a span for the operation.""" logger.debug("Durable operation started: %s", info) + if not self._tracing_enabled: + return if info.operation_type is OperationType.CONTEXT: # Context operations are tracked using on_user_function_start. return @@ -488,14 +497,16 @@ def on_operation_end(self, info: OperationEndInfo) -> None: Non-user-function operations are started by ``on_operation_start``. If an operation end is observed without a matching in-memory span, this invocation is completing an operation that began earlier, so a short - continuation span is created and linked to the deterministic logical - operation span before being ended. + continuation span with a fresh span ID is created and ended. """ logger.debug("Durable operation ended: %s", info) + if not self._tracing_enabled: + return span = self._get_span(info.operation_id) if span is None: - # the span was not started in the current invocation, so we need to - # create a new one that links to the previous one + # The operation started in a prior invocation. The prior SpanContext + # is not checkpointed, so create a new correlated segment without a + # fabricated link. parent_span = self._resolve_parent_span(info.parent_id) attributes = self._extract_attributes(info) span = self._start_span( @@ -532,6 +543,8 @@ def on_user_function_start(self, info: UserFunctionStartInfo) -> None: info: Information about the operation attempt. """ logger.debug("Durable user function started: %s", info) + if not self._tracing_enabled: + return # Context and Step operations are tracked using on_user_function_start if info.operation_type not in [OperationType.CONTEXT, OperationType.STEP]: raise RuntimeError( @@ -574,6 +587,8 @@ def on_user_function_end(self, info: UserFunctionEndInfo) -> None: info: Information about the operation attempt. """ logger.debug("Durable user function ended: %s", info) + if not self._tracing_enabled: + return if info.operation_type not in [OperationType.CONTEXT, OperationType.STEP]: raise RuntimeError( "on_user_function_end should only be called for CONTEXT and STEP operations" diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/otel_plugin_config.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/otel_plugin_config.py index cfbb0181..eedd85ee 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/otel_plugin_config.py +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/otel_plugin_config.py @@ -7,13 +7,12 @@ from __future__ import annotations -from dataclasses import dataclass, field +from dataclasses import dataclass from enum import Enum -from typing import TYPE_CHECKING, Sequence +from typing import TYPE_CHECKING if TYPE_CHECKING: - from opentelemetry.propagators.textmap import TextMapPropagator from opentelemetry.sdk.trace import TracerProvider as SdkTracerProvider from aws_durable_execution_sdk_python_otel.context_extractors import ( @@ -23,9 +22,6 @@ DEFAULT_INSTRUMENT_NAME = "aws-durable-execution-sdk-python" DEFAULT_WORKFLOW_SPAN_NAME = "Workflow" -# OTLPSpanExporter appends /v1/traces itself, so the base endpoint must NOT -# include it (mirrors the JS fix in PR #729 that removed the duplicate path). -DEFAULT_OTLP_ENDPOINT = "http://localhost:4318" class ProviderSource(Enum): @@ -37,15 +33,6 @@ class ProviderSource(Enum): EXPLICIT = "explicit" # use config.tracer_provider as-is GLOBAL = "global" # default: use the global provider (trace.get_tracer_provider()) - AUTO_OTLP = "auto_otlp" # plugin builds and owns an OTLP provider - - -@dataclass -class ExporterConfig: - """OTLP exporter configuration for the auto-configured TracerProvider.""" - - endpoint: str | None = None - headers: dict[str, str] | None = None @dataclass @@ -59,20 +46,14 @@ class OtelPluginConfig: provider_source: Selects how the tracer provider is obtained (:class:`ProviderSource`). Defaults to ``GLOBAL`` (uses the globally configured provider, e.g. the ADOT Lambda layer, via - ``trace.get_tracer_provider()``). ``AUTO_OTLP`` makes the plugin - build and own an OTLP provider. ``EXPLICIT`` uses ``tracer_provider`` - as-is and skips instrumentation registration. + ``trace.get_tracer_provider()``). ``EXPLICIT`` uses + ``tracer_provider`` as-is and skips instrumentation registration. tracer_provider: The provider used when ``provider_source`` is ``EXPLICIT``. Required in that case and must be left unset for - ``GLOBAL`` / ``AUTO_OTLP``. + ``GLOBAL``. context_extractor: Upstream trace-context extractor. Defaults to the X-Ray extractor when omitted. instrument_name: Instrumentation scope name. - enable_http_instrumentation: Whether to register HTTP instrumentation - when the plugin owns an auto-configured provider. Defaults to True. - exporter_config: OTLP exporter settings for the auto-configured provider. - propagators: Custom propagators for the auto-configured provider. - Defaults to ``[AWSXRayPropagator, W3CTraceContextPropagator]``. workflow_span_name: Name of the Workflow root span (ExecutionOtelPlugin). enrich_logger: Install the root-logger OTel context filter. """ @@ -81,9 +62,6 @@ class OtelPluginConfig: tracer_provider: SdkTracerProvider | None = None context_extractor: ContextExtractor | None = None instrument_name: str = DEFAULT_INSTRUMENT_NAME - enable_http_instrumentation: bool = True - exporter_config: ExporterConfig = field(default_factory=ExporterConfig) - propagators: Sequence[TextMapPropagator] | None = None workflow_span_name: str = DEFAULT_WORKFLOW_SPAN_NAME enrich_logger: bool = True @@ -92,8 +70,7 @@ def __post_init__(self) -> None: The config is fully driven by :attr:`provider_source`; ``tracer_provider`` is the one source-specific field, so it must be present for ``EXPLICIT`` - and absent for ``GLOBAL`` / ``AUTO_OTLP`` (where it would be silently - ignored). + and absent for ``GLOBAL`` (where it would be silently ignored). """ if self.provider_source is ProviderSource.EXPLICIT: if self.tracer_provider is None: @@ -103,5 +80,5 @@ def __post_init__(self) -> None: "tracer_provider is only valid with provider_source=EXPLICIT; " f"got provider_source={self.provider_source.name}. Set " "ProviderSource.EXPLICIT, or drop tracer_provider for " - "GLOBAL / AUTO_OTLP." + "GLOBAL." ) diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/provider.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/provider.py index e310c438..a8a0f018 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/provider.py +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/provider.py @@ -5,159 +5,34 @@ 1. ``EXPLICIT`` - the config's ``tracer_provider`` is used as-is. 2. ``GLOBAL`` - the globally configured provider is used (e.g. ADOT layer). -3. ``AUTO_OTLP`` - a fully auto-configured SDK provider is created with an OTLP - exporter, batch processor, sampler and Lambda resource attributes; this is - the only tier the plugin owns/flushes. """ from __future__ import annotations -import logging -import os from dataclasses import dataclass from typing import TYPE_CHECKING -from opentelemetry import propagate, trace -from opentelemetry.propagators.composite import CompositePropagator +from opentelemetry import trace from aws_durable_execution_sdk_python_otel.otel_plugin_config import ( - DEFAULT_OTLP_ENDPOINT, OtelPluginConfig, ProviderSource, ) if TYPE_CHECKING: - from opentelemetry.sdk.trace import IdGenerator - from opentelemetry.sdk.trace import TracerProvider as SdkTracerProvider from opentelemetry.trace import TracerProvider -logger = logging.getLogger(__name__) - -SAMPLING_RATIO_ENV = "OTEL_DURABLE_SAMPLING_RATIO" -OTLP_ENDPOINT_ENV = "OTEL_EXPORTER_OTLP_ENDPOINT" - - @dataclass class ProviderResult: - """Result of provider resolution: the provider and how it was chosen. - - ``source`` is the single value callers key their instrumentation/flush - decisions off; the plugin owns (and flushes) the provider only when it is - :attr:`ProviderSource.AUTO_OTLP`. - """ + """Result of provider resolution: the provider and how it was chosen.""" tracer_provider: TracerProvider source: ProviderSource -def _resolve_endpoint(config: OtelPluginConfig) -> str: - """Resolve the OTLP traces endpoint (config -> env -> default). - - The Python OTLP/HTTP exporter uses an explicitly-passed ``endpoint`` verbatim - (unlike the JS exporter, which appends ``/v1/traces``), so we append the - signal path here when the caller supplied only a base URL. - """ - base = ( - config.exporter_config.endpoint - or os.environ.get(OTLP_ENDPOINT_ENV) - or DEFAULT_OTLP_ENDPOINT - ) - base = base.rstrip("/") - if not base.endswith("/v1/traces"): - base = f"{base}/v1/traces" - return base - - -def _build_sampler(): - """Build the sampler from ``OTEL_DURABLE_SAMPLING_RATIO``. - - A valid ratio in ``[0, 1]`` yields a ``TraceIdRatioBased`` sampler; anything - else falls back to ``ALWAYS_ON``. If constructing the ratio sampler raises, - the error propagates and fails provider setup. - """ - from opentelemetry.sdk.trace.sampling import ALWAYS_ON, TraceIdRatioBased - - raw = os.environ.get(SAMPLING_RATIO_ENV) - if raw is None: - return ALWAYS_ON - try: - ratio = float(raw) - except (TypeError, ValueError): - return ALWAYS_ON - if not (0.0 <= ratio <= 1.0): - return ALWAYS_ON - return TraceIdRatioBased(ratio) - - -def _build_resource(): - """Build a Lambda resource from AWS_* env vars.""" - from opentelemetry.sdk.resources import Resource - - function_name = os.environ.get("AWS_LAMBDA_FUNCTION_NAME") - if not function_name: - return None - attributes: dict[str, str] = { - "service.name": function_name, - "faas.name": function_name, - "cloud.provider": "aws", - "cloud.platform": "aws_lambda", - } - region = os.environ.get("AWS_REGION") - if region: - attributes["cloud.region"] = region - version = os.environ.get("AWS_LAMBDA_FUNCTION_VERSION") - if version: - attributes["faas.version"] = version - return Resource.create(attributes) - - -def _default_propagators(config: OtelPluginConfig): - """Return configured propagators or the X-Ray + W3C default.""" - if config.propagators is not None: - return list(config.propagators) - from opentelemetry.propagators.aws import AwsXRayPropagator - from opentelemetry.trace.propagation.tracecontext import ( - TraceContextTextMapPropagator, - ) - - return [AwsXRayPropagator(), TraceContextTextMapPropagator()] - - -def _create_auto_provider( - config: OtelPluginConfig, - id_generator: IdGenerator | None, -) -> SdkTracerProvider: - """Create a fully self-configured SDK TracerProvider with OTLP export.""" - from opentelemetry.exporter.otlp.proto.http.trace_exporter import OTLPSpanExporter - from opentelemetry.sdk.trace import TracerProvider as SdkTracerProvider - from opentelemetry.sdk.trace.export import BatchSpanProcessor - - exporter = OTLPSpanExporter( - endpoint=_resolve_endpoint(config), - headers=config.exporter_config.headers or None, - ) - provider_kwargs: dict = {"sampler": _build_sampler()} - resource = _build_resource() - if resource is not None: - provider_kwargs["resource"] = resource - if id_generator is not None: - provider_kwargs["id_generator"] = id_generator - - provider = SdkTracerProvider(**provider_kwargs) - provider.add_span_processor(BatchSpanProcessor(exporter)) - - composite = CompositePropagator(_default_propagators(config)) - propagate.set_global_textmap(composite) - return provider - - -def create_tracer_provider( - config: OtelPluginConfig, - *, - id_generator: IdGenerator | None = None, -) -> ProviderResult: +def create_tracer_provider(config: OtelPluginConfig) -> ProviderResult: """Resolve a TracerProvider from the config's :attr:`provider_source`. A straight switch on ``config.provider_source``; the chosen tier is reported @@ -166,12 +41,9 @@ def create_tracer_provider( 1. ``EXPLICIT`` -> ``config.tracer_provider`` used as-is 2. ``GLOBAL`` -> the globally configured provider - 3. ``AUTO_OTLP`` -> a plugin-owned, auto-configured OTLP provider Args: config: Shared plugin configuration. - id_generator: Deterministic ID generator injected into an - auto-configured provider so cross-invocation trace stitching works. Returns: A :class:`ProviderResult`. @@ -185,8 +57,6 @@ def create_tracer_provider( provider: TracerProvider = config.tracer_provider elif source is ProviderSource.GLOBAL: provider = trace.get_tracer_provider() - elif source is ProviderSource.AUTO_OTLP: - provider = _create_auto_provider(config, id_generator) else: # pragma: no cover - exhaustive over ProviderSource raise ValueError(f"unknown provider_source: {source!r}") diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_deterministic_id_generator.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_deterministic_id_generator.py index 0dbf154f..38dbe08a 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_deterministic_id_generator.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_deterministic_id_generator.py @@ -6,13 +6,12 @@ import threading from datetime import UTC, datetime -from opentelemetry.sdk.trace import IdGenerator, RandomIdGenerator +import pytest +from opentelemetry.sdk.trace import IdGenerator, RandomIdGenerator, TracerProvider from aws_durable_execution_sdk_python_otel.deterministic_id_generator import ( DeterministicIdGenerator, - _parse_xray_root_trace_id, _to_otel_trace_id, - _xray_trace_id_to_otel, operation_id_to_span_id, ) @@ -20,9 +19,12 @@ class _StubIdGenerator(IdGenerator): """An IdGenerator that returns fixed, identifiable IDs.""" - def __init__(self, trace_id: int, span_id: int) -> None: + def __init__( + self, trace_id: int, span_id: int, *, trace_id_is_random: bool = True + ) -> None: self._trace_id = trace_id self._span_id = span_id + self._trace_id_is_random = trace_id_is_random def generate_trace_id(self) -> int: return self._trace_id @@ -30,42 +32,12 @@ def generate_trace_id(self) -> int: def generate_span_id(self) -> int: return self._span_id - -def test_parse_xray_root_trace_id_returns_root_from_header(): - """Verify X-Ray Root trace ID parsing ignores other header fields.""" - trace_header = ( - "Root=1-5759e988-bd862e3fe1be46a994272793;Parent=53995c3f42cd8ad8;Sampled=1" - ) - - assert ( - _parse_xray_root_trace_id(trace_header) == "1-5759e988-bd862e3fe1be46a994272793" - ) - - -def test_parse_xray_root_trace_id_returns_none_for_missing_or_malformed_header(): - """Verify absent or malformed X-Ray headers are ignored.""" - assert _parse_xray_root_trace_id(None) is None - assert _parse_xray_root_trace_id("") is None - assert _parse_xray_root_trace_id("Parent=53995c3f42cd8ad8;Sampled=1") is None - assert ( - _parse_xray_root_trace_id( - "Root=1-5759e988-not-enough-hex;Parent=53995c3f42cd8ad8" - ) - is None - ) - - -def test_xray_trace_id_to_otel_removes_xray_prefix_and_normalizes_case(): - """Verify X-Ray trace IDs are converted into OTel-compatible integers.""" - trace_id = "1-5759E988-BD862E3FE1BE46A994272793" - - assert _xray_trace_id_to_otel(trace_id) == int( - "5759e988bd862e3fe1be46a994272793", 16 - ) + def is_trace_id_random(self) -> bool: + return self._trace_id_is_random -def test_to_otel_trace_id_uses_xray_root_header_when_available(monkeypatch): - """Verify Lambda's X-Ray trace header takes precedence over fallback IDs.""" +def test_to_otel_trace_id_is_independent_of_xray_root_header(monkeypatch): + """The Workflow trace must not collide with the ambient X-Ray trace.""" monkeypatch.setenv( "_X_AMZN_TRACE_ID", "Root=1-5759e988-bd862e3fe1be46a994272793;Parent=53995c3f42cd8ad8;Sampled=1", @@ -73,13 +45,12 @@ def test_to_otel_trace_id_uses_xray_root_header_when_available(monkeypatch): start_timestamp = datetime(2024, 1, 2, 3, 4, 5, tzinfo=UTC) assert _to_otel_trace_id("different-execution-arn", start_timestamp) == int( - "5759e988bd862e3fe1be46a994272793", 16 + "65937d2517528419530c40ebaa7ddacf", 16 ) -def test_to_otel_trace_id_falls_back_to_timestamp_and_execution_arn(monkeypatch): - """Verify fallback trace IDs are deterministic for the same execution.""" - monkeypatch.delenv("_X_AMZN_TRACE_ID", raising=False) +def test_to_otel_trace_id_uses_timestamp_and_execution_arn(): + """Verify trace IDs are deterministic for the same execution.""" execution_arn = "arn:aws:lambda:us-west-2:123456789012:function:workflow:$LATEST" start_timestamp = datetime(2024, 1, 2, 3, 4, 5, tzinfo=UTC) @@ -97,62 +68,107 @@ def test_operation_id_to_span_id_returns_deterministic_64_bit_id(): ) -def test_deterministic_id_generator_returns_cached_trace_id(monkeypatch): - """Verify trace IDs are cached after being set for an execution.""" - monkeypatch.delenv("_X_AMZN_TRACE_ID", raising=False) - generator = DeterministicIdGenerator() +def test_use_ids_temporarily_overrides_trace_and_span_ids(): + """Verify deterministic IDs apply only within the explicit scope.""" + fallback = _StubIdGenerator(trace_id=int("a" * 32, 16), span_id=int("b" * 16, 16)) + generator = DeterministicIdGenerator(fallback_id_generator=fallback) + deterministic_trace_id = int("1" * 32, 16) + deterministic_span_id = int("2" * 16, 16) - generator.set_trace_id( - "arn:aws:lambda:us-west-2:123456789012:function:workflow:$LATEST", - datetime(2024, 1, 2, 3, 4, 5, tzinfo=UTC), - ) + assert generator.generate_trace_id() == int("a" * 32, 16) + assert generator.generate_span_id() == int("b" * 16, 16) + + with generator.use_ids( + trace_id=deterministic_trace_id, span_id=deterministic_span_id + ): + assert generator.generate_trace_id() == deterministic_trace_id + assert generator.generate_span_id() == deterministic_span_id - assert generator.generate_trace_id() == int("65937d253aa8c3f7ffe36c50d65b1a6d", 16) + assert generator.generate_trace_id() == int("a" * 32, 16) + assert generator.generate_span_id() == int("b" * 16, 16) -def test_deterministic_id_generator_falls_back_to_random_trace_id(monkeypatch): - """Verify trace IDs are random until an execution trace ID is set.""" - expected_trace_id = int("1" * 32, 16) - generator = DeterministicIdGenerator() - monkeypatch.setattr( - generator._fallback_id_generator, - "generate_trace_id", - lambda: expected_trace_id, +def test_use_ids_allows_deterministic_trace_with_fallback_span(): + """Verify trace and span generation can be overridden independently.""" + fallback_span_id = int("b" * 16, 16) + generator = DeterministicIdGenerator( + fallback_id_generator=_StubIdGenerator( + trace_id=int("a" * 32, 16), span_id=fallback_span_id + ) ) + deterministic_trace_id = int("1" * 32, 16) - assert generator.generate_trace_id() == expected_trace_id + with generator.use_ids(trace_id=deterministic_trace_id, span_id=None): + assert generator.generate_trace_id() == deterministic_trace_id + assert generator.generate_span_id() == fallback_span_id -def test_deterministic_id_generator_uses_next_span_id_once(monkeypatch): - """Verify a configured span ID only applies to the next generated span.""" +def test_use_ids_consumes_deterministic_span_id_once(): + """Verify re-entrant generation cannot reuse a deterministic span ID.""" + fallback_span_id = int("b" * 16, 16) + generator = DeterministicIdGenerator( + fallback_id_generator=_StubIdGenerator( + trace_id=int("a" * 32, 16), span_id=fallback_span_id + ) + ) + deterministic_trace_id = int("1" * 32, 16) deterministic_span_id = int("2" * 16, 16) - random_span_id = int("3" * 16, 16) - generator = DeterministicIdGenerator() - monkeypatch.setattr( - generator._fallback_id_generator, - "generate_span_id", - lambda: random_span_id, + + with generator.use_ids( + trace_id=deterministic_trace_id, span_id=deterministic_span_id + ): + assert generator.generate_span_id() == deterministic_span_id + assert generator.generate_span_id() == fallback_span_id + assert generator.generate_trace_id() == deterministic_trace_id + + +def test_use_ids_restores_fallback_after_exception(): + """Verify an interrupted span creation cannot leak deterministic IDs.""" + fallback_trace_id = int("a" * 32, 16) + fallback_span_id = int("b" * 16, 16) + generator = DeterministicIdGenerator( + fallback_id_generator=_StubIdGenerator( + trace_id=fallback_trace_id, span_id=fallback_span_id + ) ) - generator.set_next_span_id(deterministic_span_id) + with pytest.raises(RuntimeError, match="span creation failed"): + with generator.use_ids(trace_id=int("1" * 32, 16), span_id=int("2" * 16, 16)): + raise RuntimeError("span creation failed") - assert generator.generate_span_id() == deterministic_span_id - assert generator.generate_span_id() == random_span_id + assert generator.generate_trace_id() == fallback_trace_id + assert generator.generate_span_id() == fallback_span_id -def test_deterministic_id_generator_accepts_cleared_next_span_id(monkeypatch): - """Verify clearing the next span ID preserves random span generation.""" - expected_span_id = int("4" * 16, 16) +def test_use_ids_restores_outer_nested_scope(): + """Verify nested scopes restore the preceding deterministic IDs.""" generator = DeterministicIdGenerator() - monkeypatch.setattr( - generator._fallback_id_generator, - "generate_span_id", - lambda: expected_span_id, + outer_trace_id = int("1" * 32, 16) + outer_span_id = int("2" * 16, 16) + + with generator.use_ids(trace_id=outer_trace_id, span_id=outer_span_id): + with generator.use_ids(trace_id=int("3" * 32, 16), span_id=int("4" * 16, 16)): + assert generator.generate_trace_id() == int("3" * 32, 16) + assert generator.generate_trace_id() == outer_trace_id + assert generator.generate_span_id() == outer_span_id + + +def test_nested_scope_does_not_restore_consumed_outer_span_id(): + """Verify nested scope cleanup preserves prior one-shot consumption.""" + fallback_span_id = int("b" * 16, 16) + generator = DeterministicIdGenerator( + fallback_id_generator=_StubIdGenerator( + trace_id=int("a" * 32, 16), span_id=fallback_span_id + ) ) + outer_span_id = int("2" * 16, 16) + inner_span_id = int("4" * 16, 16) - generator.set_next_span_id(None) - - assert generator.generate_span_id() == expected_span_id + with generator.use_ids(trace_id=int("1" * 32, 16), span_id=outer_span_id): + assert generator.generate_span_id() == outer_span_id + with generator.use_ids(trace_id=int("3" * 32, 16), span_id=inner_span_id): + assert generator.generate_span_id() == inner_span_id + assert generator.generate_span_id() == fallback_span_id def test_deterministic_id_generator_defaults_to_random_fallback(): @@ -162,10 +178,9 @@ def test_deterministic_id_generator_defaults_to_random_fallback(): assert isinstance(generator._fallback_id_generator, RandomIdGenerator) -def test_deterministic_id_generator_uses_provided_fallback_for_trace_id(monkeypatch): +def test_deterministic_id_generator_uses_provided_fallback_for_trace_id(): """Verify the supplied fallback generator produces trace IDs when no execution trace ID is set.""" - monkeypatch.delenv("_X_AMZN_TRACE_ID", raising=False) fallback = _StubIdGenerator(trace_id=int("a" * 32, 16), span_id=int("b" * 16, 16)) generator = DeterministicIdGenerator(fallback_id_generator=fallback) @@ -181,105 +196,105 @@ def test_deterministic_id_generator_uses_provided_fallback_for_span_id(): assert generator.generate_span_id() == int("b" * 16, 16) -def test_deterministic_id_generator_prefers_execution_trace_id_over_fallback( - monkeypatch, -): - """Verify a configured execution trace ID takes precedence over the fallback.""" - monkeypatch.delenv("_X_AMZN_TRACE_ID", raising=False) - fallback = _StubIdGenerator(trace_id=int("a" * 32, 16), span_id=int("b" * 16, 16)) - generator = DeterministicIdGenerator(fallback_id_generator=fallback) +def test_install_on_tracer_does_not_replace_provider_generator(): + """Verify deterministic generation is isolated to one instrumentation scope.""" + provider = TracerProvider() + provider_generator = provider.id_generator + tracer = provider.get_tracer("durable-plugin") - generator.set_trace_id( - "arn:aws:lambda:us-west-2:123456789012:function:workflow:$LATEST", - datetime(2024, 1, 2, 3, 4, 5, tzinfo=UTC), - ) + generator = DeterministicIdGenerator.install_on_tracer(tracer) - assert generator.generate_trace_id() == int("65937d253aa8c3f7ffe36c50d65b1a6d", 16) + assert tracer.id_generator is generator + assert provider.id_generator is provider_generator + assert provider.get_tracer("unrelated-library").id_generator is provider_generator -def test_deterministic_id_generator_prefers_next_span_id_over_fallback(): - """Verify a pending deterministic span ID takes precedence over the fallback.""" - deterministic_span_id = int("c" * 16, 16) - fallback = _StubIdGenerator(trace_id=int("a" * 32, 16), span_id=int("b" * 16, 16)) - generator = DeterministicIdGenerator(fallback_id_generator=fallback) +def test_install_on_tracer_reuses_existing_generator(): + """Verify cached tracers share one scoped generator rather than wrappers.""" + provider = TracerProvider() + tracer = provider.get_tracer("durable-plugin") - generator.set_next_span_id(deterministic_span_id) + first = DeterministicIdGenerator.install_on_tracer(tracer) + second = DeterministicIdGenerator.install_on_tracer(tracer) - assert generator.generate_span_id() == deterministic_span_id - # Subsequent calls fall back to the provided generator. - assert generator.generate_span_id() == int("b" * 16, 16) + assert second is first + + +def test_is_trace_id_random_delegates_outside_override(): + """Verify deterministic trace IDs do not receive the random-ID trace flag.""" + generator = DeterministicIdGenerator( + fallback_id_generator=_StubIdGenerator( + trace_id=int("a" * 32, 16), + span_id=int("b" * 16, 16), + trace_id_is_random=True, + ) + ) + assert generator.is_trace_id_random() is True + with generator.use_ids(trace_id=int("1" * 32, 16), span_id=None): + assert generator.is_trace_id_random() is False -def test_pending_span_id_is_isolated_across_threads(): - """Verify a span ID set in one thread is not consumed by another thread. - The pending span ID is stored in a ContextVar, so each worker thread has - its own value. Without this isolation a concurrent operation could steal - another operation's deterministic span ID, producing the wrong span ID. - """ +def test_id_overrides_are_isolated_across_threads(): + """Verify concurrent threads cannot consume or overwrite each other's IDs.""" random_span_id = int("f" * 16, 16) fallback = _StubIdGenerator(trace_id=int("a" * 32, 16), span_id=random_span_id) generator = DeterministicIdGenerator(fallback_id_generator=fallback) - # The main thread sets a deterministic span ID but never consumes it. - main_deterministic_span_id = int("1" * 16, 16) - generator.set_next_span_id(main_deterministic_span_id) - barrier = threading.Barrier(2) - results: dict[str, int] = {} + results: dict[str, tuple[int, int]] = {} - def worker(name: str, span_id: int) -> None: - # Each worker starts with a fresh context (default None), so it must - # not see the main thread's pending span ID. - barrier.wait() - results[f"{name}-before-set"] = generator.generate_span_id() - generator.set_next_span_id(span_id) - results[f"{name}-after-set"] = generator.generate_span_id() + def worker(name: str, trace_id: int, span_id: int) -> None: + with generator.use_ids(trace_id=trace_id, span_id=span_id): + barrier.wait() + results[name] = ( + generator.generate_trace_id(), + generator.generate_span_id(), + ) + worker_a_trace_id = int("1" * 32, 16) worker_a_span_id = int("2" * 16, 16) + worker_b_trace_id = int("3" * 32, 16) worker_b_span_id = int("3" * 16, 16) - thread_a = threading.Thread(target=worker, args=("a", worker_a_span_id)) - thread_b = threading.Thread(target=worker, args=("b", worker_b_span_id)) + thread_a = threading.Thread( + target=worker, args=("a", worker_a_trace_id, worker_a_span_id) + ) + thread_b = threading.Thread( + target=worker, args=("b", worker_b_trace_id, worker_b_span_id) + ) thread_a.start() thread_b.start() thread_a.join() thread_b.join() - # Workers never observed the main thread's value; they fell back to random. - assert results["a-before-set"] == random_span_id - assert results["b-before-set"] == random_span_id - # Each worker consumed only its own deterministic span ID. - assert results["a-after-set"] == worker_a_span_id - assert results["b-after-set"] == worker_b_span_id - # The main thread's pending span ID was untouched by the workers. - assert generator.generate_span_id() == main_deterministic_span_id - + assert results["a"] == (worker_a_trace_id, worker_a_span_id) + assert results["b"] == (worker_b_trace_id, worker_b_span_id) + assert generator.generate_span_id() == random_span_id -def test_pending_span_id_is_isolated_across_async_tasks(): - """Verify a span ID set in one async task is not consumed by another. - Each asyncio task runs with its own copied context, so the pending span ID - stays scoped to the task that set it even across await boundaries on the - same thread. - """ +def test_id_overrides_are_isolated_across_async_tasks(): + """Verify interleaved tasks retain their own trace and span IDs.""" fallback_span_id = int("e" * 16, 16) fallback = _StubIdGenerator(trace_id=int("a" * 32, 16), span_id=fallback_span_id) generator = DeterministicIdGenerator(fallback_id_generator=fallback) + task_a_trace_id = int("4" * 32, 16) task_a_span_id = int("4" * 16, 16) + task_b_trace_id = int("5" * 32, 16) task_b_span_id = int("5" * 16, 16) - async def task(span_id: int) -> int: - generator.set_next_span_id(span_id) - # Yield control so the other task interleaves between set and consume. - await asyncio.sleep(0) - return generator.generate_span_id() + async def task(trace_id: int, span_id: int) -> tuple[int, int]: + with generator.use_ids(trace_id=trace_id, span_id=span_id): + await asyncio.sleep(0) + return generator.generate_trace_id(), generator.generate_span_id() - async def main() -> tuple[int, int]: - return await asyncio.gather(task(task_a_span_id), task(task_b_span_id)) + async def main() -> tuple[tuple[int, int], tuple[int, int]]: + return await asyncio.gather( + task(task_a_trace_id, task_a_span_id), + task(task_b_trace_id, task_b_span_id), + ) result_a, result_b = asyncio.run(main()) - # Despite interleaving, each task consumed only its own deterministic ID. - assert result_a == task_a_span_id - assert result_b == task_b_span_id + assert result_a == (task_a_trace_id, task_a_span_id) + assert result_b == (task_b_trace_id, task_b_span_id) diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py index 75488662..171dae87 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py @@ -124,7 +124,7 @@ def test_derive_workflow_span_id_rejects_empty_arn(): # --------------------------------------------------------------------------- # Workflow + invocation span hierarchy # --------------------------------------------------------------------------- -def test_workflow_span_is_root_and_invocation_is_its_child(): +def test_workflow_and_invocation_are_separate_roots_without_ambient_parent(): plugin, exporter = _create_plugin() plugin.on_invocation_start(_invocation_start_info()) @@ -146,9 +146,30 @@ def test_workflow_span_is_root_and_invocation_is_its_child(): == InvocationStatus.SUCCEEDED.value ) - # Invocation is parented under the Workflow span. + # Without ambient context, Invocation starts a separate provider trace. + assert invocation.parent is None + assert invocation.context.trace_id != workflow.context.trace_id + + +def test_explicit_mode_invocation_span_parented_to_ambient_span(): + plugin, exporter = _create_plugin() + + ambient = plugin._provider.get_tracer("ambient").start_span("lambda-invocation") + token = otel_context.attach(trace.set_span_in_context(ambient)) + try: + plugin.on_invocation_start(_invocation_start_info()) + plugin.on_invocation_end(_invocation_end_info()) + finally: + otel_context.detach(token) + ambient.end() + + spans = {s.name: s for s in exporter.get_finished_spans()} + workflow = spans["Workflow"] + invocation = spans["Invocation"] assert invocation.parent is not None - assert invocation.parent.span_id == workflow.context.span_id + assert invocation.parent.span_id == ambient.get_span_context().span_id + assert invocation.context.trace_id == ambient.get_span_context().trace_id + assert workflow.context.trace_id != ambient.get_span_context().trace_id def test_workflow_span_dropped_on_non_terminal_status(): @@ -426,6 +447,8 @@ def test_default_mode_creates_invocation_span(monkeypatch): invocation = spans["Invocation"] assert invocation.attributes["durable.execution.arn"] == EXECUTION_ARN assert invocation.attributes["durable.invocation.first"] is True + assert invocation.parent is None + assert invocation.context.trace_id != spans["Workflow"].context.trace_id def test_default_mode_invocation_span_parented_to_ambient_span(monkeypatch): @@ -444,6 +467,7 @@ def test_default_mode_invocation_span_parented_to_ambient_span(monkeypatch): invocation = {s.name: s for s in exporter.get_finished_spans()}["Invocation"] assert invocation.parent is not None assert invocation.parent.span_id == ambient.get_span_context().span_id + assert invocation.context.trace_id == ambient.get_span_context().trace_id def test_open_operation_span_not_exported_at_invocation_end(): diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin_integration.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin_integration.py index 98f450c5..bc48159b 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin_integration.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin_integration.py @@ -3,11 +3,12 @@ Drives the full plugin lifecycle against a real TracerProvider + InMemorySpanExporter for the two deployment shapes: -* Community collector layer: the plugin owns its provider - (``provider_source=AUTO_OTLP`` / ``EXPLICIT``); the Workflow span is the trace root. +* Community collector layer: the caller supplies a provider + (``provider_source=EXPLICIT``); the Workflow and Invocation spans root + separate traces when no ambient parent exists. * ADOT layer: the ADOT Lambda layer supplies the global provider and the ambient - Lambda invocation span (``provider_source=GLOBAL``); the plugin's - Invocation span parents to that ambient span. + Lambda invocation span (``provider_source=GLOBAL``); the plugin's Invocation + span parents to that ambient span. """ from __future__ import annotations @@ -36,8 +37,13 @@ from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter +from opentelemetry.trace import ( + ProxyTracerProvider, + TracerProvider as ApiTracerProvider, +) from aws_durable_execution_sdk_python_otel.deterministic_id_generator import ( + DeterministicIdGenerator, derive_workflow_span_id, operation_id_to_span_id, ) @@ -57,6 +63,7 @@ "Root=1-5759e988-bd862e3fe1be46a994272793;Parent=53995c3f42cd8ad8;Sampled=1" ) XRAY_TRACE_ID = int("5759e988bd862e3fe1be46a994272793", 16) +EXECUTION_TRACE_ID = int("65937d253aa8c3f7ffe36c50d65b1a6d", 16) @pytest.fixture(autouse=True) @@ -158,6 +165,122 @@ def _run_step_lifecycle(plugin: ExecutionOtelPlugin) -> None: ) +def _config_for_source( + source: ProviderSource, + provider: TracerProvider, + monkeypatch: pytest.MonkeyPatch, +) -> OtelPluginConfig: + if source is ProviderSource.GLOBAL: + monkeypatch.setattr(trace, "get_tracer_provider", lambda: provider) + return OtelPluginConfig( + provider_source=source, + tracer_provider=provider if source is ProviderSource.EXPLICIT else None, + context_extractor=lambda _: Context(), + enrich_logger=False, + ) + + +@pytest.mark.parametrize( + "source", + [ + ProviderSource.EXPLICIT, + ProviderSource.GLOBAL, + ], +) +def test_unrelated_root_spans_keep_provider_id_generation( + source: ProviderSource, monkeypatch: pytest.MonkeyPatch +) -> None: + """Unrelated scopes receive fresh roots throughout a durable invocation.""" + provider, _ = _provider() + provider_generator = provider.id_generator + unrelated_tracer = provider.get_tracer("unrelated-library") + before = unrelated_tracer.start_span("before", context=Context()) + + plugin = ExecutionOtelPlugin(_config_for_source(source, provider, monkeypatch)) + assert provider.id_generator is provider_generator + assert isinstance(plugin._id_generator, DeterministicIdGenerator) + + plugin.on_invocation_start(_invocation_start()) + workflow = plugin._workflow_span + assert workflow is not None + during = unrelated_tracer.start_span("during", context=Context()) + plugin.on_invocation_end(_invocation_end()) + after = unrelated_tracer.start_span("after", context=Context()) + + trace_ids = { + before.get_span_context().trace_id, + during.get_span_context().trace_id, + after.get_span_context().trace_id, + workflow.get_span_context().trace_id, + } + assert len(trace_ids) == 4 + + before.end() + during.end() + after.end() + + +def test_global_proxy_binds_sdk_provider_before_first_invocation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A plugin created before global SDK setup binds when invocation starts.""" + monkeypatch.setattr(trace, "_TRACER_PROVIDER", None) + current_provider: list[ApiTracerProvider] = [ProxyTracerProvider()] + monkeypatch.setattr(trace, "get_tracer_provider", lambda: current_provider[0]) + plugin = ExecutionOtelPlugin( + OtelPluginConfig( + provider_source=ProviderSource.GLOBAL, + context_extractor=lambda _: Context(), + enrich_logger=False, + ) + ) + + provider, exporter = _provider() + current_provider[0] = provider + plugin.on_invocation_start(_invocation_start()) + plugin.on_invocation_end(_invocation_end()) + + assert {span.name for span in exporter.get_finished_spans()} == { + "Invocation", + "Workflow", + } + + +def test_global_proxy_disables_entire_invocation_until_sdk_provider_is_ready( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """Provider setup midway through an invocation cannot produce a partial trace.""" + monkeypatch.setattr(trace, "_TRACER_PROVIDER", None) + current_provider: list[ApiTracerProvider] = [ProxyTracerProvider()] + monkeypatch.setattr(trace, "get_tracer_provider", lambda: current_provider[0]) + plugin = ExecutionOtelPlugin( + OtelPluginConfig( + provider_source=ProviderSource.GLOBAL, + context_extractor=lambda _: Context(), + enrich_logger=False, + ) + ) + provider, exporter = _provider() + + plugin.on_invocation_start(_invocation_start()) + assert "telemetry is disabled for this invocation" in caplog.text + + current_provider[0] = provider + _run_step_lifecycle(plugin) + plugin.on_invocation_end(_invocation_end()) + assert exporter.get_finished_spans() == () + + plugin.on_invocation_start(_invocation_start()) + _run_step_lifecycle(plugin) + plugin.on_invocation_end(_invocation_end()) + assert {span.name for span in exporter.get_finished_spans()} == { + "Invocation", + "Workflow", + OP_NAME, + f"{OP_NAME} attempt 1", + } + + # --------------------------------------------------------------------------- # Community collector layer (plugin owns the provider) # --------------------------------------------------------------------------- @@ -191,9 +314,9 @@ def test_community_layer_full_lifecycle_is_workflow_rooted(): == InvocationStatus.SUCCEEDED.value ) - # Invocation span is a child of the Workflow span. - assert invocation.parent is not None - assert invocation.parent.span_id == workflow.context.span_id + # Without ambient context, Invocation roots a separate provider trace. + assert invocation.parent is None + assert invocation.context.trace_id != workflow.context.trace_id # Operation span: deterministic id, parented under Workflow, linked to invocation. assert operation.context.span_id == operation_id_to_span_id(EXECUTION_ARN, OP_ID) @@ -237,12 +360,15 @@ def test_adot_layer_full_lifecycle_parents_to_ambient_span(monkeypatch): finished = exporter.get_finished_spans() spans = {s.name: s for s in finished} + workflow = spans["Workflow"] invocation = spans["Invocation"] operation = spans[OP_NAME] # Invocation span parents to the ambient ADOT span and carries the first flag. assert invocation.parent is not None assert invocation.parent.span_id == ambient.get_span_context().span_id + assert invocation.context.trace_id == ambient.get_span_context().trace_id + assert workflow.context.trace_id != ambient.get_span_context().trace_id assert invocation.attributes["durable.invocation.first"] is True # Operation span still uses the deterministic id and links to the durable @@ -256,8 +382,8 @@ def test_adot_layer_full_lifecycle_parents_to_ambient_span(monkeypatch): assert len([s for s in finished if s.name == OP_NAME]) == 1 -def test_second_plugin_configures_cached_tracer_generator(monkeypatch): - """A second handler's Workflow span uses its deterministic trace ID.""" +def test_second_plugin_uses_execution_trace_id_independent_of_xray(monkeypatch): + """Workflow trace IDs remain deterministic and separate from X-Ray.""" provider, exporter = _provider() monkeypatch.setattr(trace, "get_tracer_provider", lambda: provider) config = OtelPluginConfig( @@ -269,11 +395,13 @@ def test_second_plugin_configures_cached_tracer_generator(monkeypatch): target_plugin = ExecutionOtelPlugin(config) monkeypatch.setenv("_X_AMZN_TRACE_ID", XRAY_TRACE_HEADER) - assert target_plugin._tracer is first_plugin._tracer + if target_plugin._tracer is first_plugin._tracer: + assert target_plugin._id_generator is first_plugin._id_generator target_plugin.on_invocation_start(_invocation_start()) target_plugin.on_invocation_end(_invocation_end()) workflow = next( span for span in exporter.get_finished_spans() if span.name == "Workflow" ) - assert workflow.context.trace_id == XRAY_TRACE_ID + assert workflow.context.trace_id == EXECUTION_TRACE_ID + assert workflow.context.trace_id != XRAY_TRACE_ID diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py index d6cf063a..204b976d 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py @@ -169,9 +169,33 @@ def test_invocation_start_and_end_emit_invocation_span(): invocation.attributes["durable.invocation.status"] == InvocationStatus.SUCCEEDED.value ) + workflow = spans_by_name["Workflow"] + assert invocation.parent is None + assert invocation.context.trace_id != workflow.context.trace_id assert plugin._get_span(None) is None +def test_invocation_span_parents_to_ambient_span(): + plugin, exporter = _create_plugin() + + ambient = plugin._provider.get_tracer("ambient").start_span("lambda-invocation") + token = otel_context.attach(trace.set_span_in_context(ambient)) + try: + plugin.on_invocation_start(_invocation_start_info()) + plugin.on_invocation_end(_invocation_end_info()) + finally: + otel_context.detach(token) + ambient.end() + + spans = {span.name: span for span in exporter.get_finished_spans()} + invocation = spans["Invocation"] + workflow = spans["Workflow"] + assert invocation.parent is not None + assert invocation.parent.span_id == ambient.get_span_context().span_id + assert invocation.context.trace_id == ambient.get_span_context().trace_id + assert workflow.context.trace_id != ambient.get_span_context().trace_id + + def test_invocation_span_records_subsequent_invocation(): """Invocation spans preserve a false first-invocation attribute.""" plugin, exporter = _create_plugin() @@ -328,8 +352,8 @@ def test_operation_callbacks_emit_child_span_with_deterministic_span_id(): ) -def test_operation_end_without_start_emits_continuation_span_with_link(): - """Verify completed existing operations link to their logical operation span.""" +def test_operation_end_without_start_omits_unobserved_previous_span_link(): + """A continuation cannot link to a SpanContext that was not checkpointed.""" plugin, exporter = _create_plugin() plugin.on_invocation_start(_invocation_start_info()) operation_id = "wait-existing" @@ -356,9 +380,9 @@ def test_operation_end_without_start_emits_continuation_span_with_link(): span = exporter.get_finished_spans()[0] assert span.name == "existing-wait" assert span.context.span_id == random_span_id - assert span.links[0].context.span_id == operation_id_to_span_id( - EXECUTION_ARN, operation_id - ) + linked_span_ids = {link.context.span_id for link in span.links} + assert linked_span_ids == {derive_workflow_span_id(EXECUTION_ARN)} + assert operation_id_to_span_id(EXECUTION_ARN, operation_id) not in linked_span_ids assert ( span.attributes["durable.operation.status"] == OperationStatus.SUCCEEDED.value ) @@ -393,8 +417,8 @@ def test_continuation_span_uses_current_start_and_end_times(): assert before_callback <= span.start_time <= span.end_time <= after_callback -def test_retried_operation_start_emits_continuation_span_with_link(): - """Retried operation spans should not reuse the original deterministic span ID.""" +def test_retried_operation_uses_fresh_id_without_unobserved_previous_span_link(): + """Retried segments use fresh IDs without fabricating a prior context.""" plugin, exporter = _create_plugin() plugin.on_invocation_start(_invocation_start_info()) operation_id = "step-retried" @@ -433,9 +457,9 @@ def test_retried_operation_start_emits_continuation_span_with_link(): span = exporter.get_finished_spans()[0] assert span.name == "retried-step" assert span.context.span_id == random_span_id - assert span.links[0].context.span_id == operation_id_to_span_id( - EXECUTION_ARN, operation_id - ) + linked_span_ids = {link.context.span_id for link in span.links} + assert linked_span_ids == {derive_workflow_span_id(EXECUTION_ARN)} + assert operation_id_to_span_id(EXECUTION_ARN, operation_id) not in linked_span_ids def test_step_operation_span_parents_attempt_span(): diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin_integration.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin_integration.py index c1c6d4b1..a4631057 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin_integration.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin_integration.py @@ -39,9 +39,15 @@ from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import SimpleSpanProcessor from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter -from opentelemetry.trace import SpanKind +from opentelemetry.trace import ( + ProxyTracerProvider, + SpanKind, + TracerProvider as ApiTracerProvider, +) from aws_durable_execution_sdk_python_otel.deterministic_id_generator import ( + DeterministicIdGenerator, + derive_workflow_span_id, operation_id_to_span_id, ) from aws_durable_execution_sdk_python_otel.invocation_plugin import InvocationOtelPlugin @@ -60,6 +66,7 @@ "Root=1-5759e988-bd862e3fe1be46a994272793;Parent=53995c3f42cd8ad8;Sampled=1" ) XRAY_TRACE_ID = int("5759e988bd862e3fe1be46a994272793", 16) +EXECUTION_TRACE_ID = int("65937d253aa8c3f7ffe36c50d65b1a6d", 16) @pytest.fixture(autouse=True) @@ -161,8 +168,125 @@ def _run_step_lifecycle(plugin: InvocationOtelPlugin) -> None: ) +def _config_for_source( + source: ProviderSource, + provider: TracerProvider, + monkeypatch: pytest.MonkeyPatch, +) -> OtelPluginConfig: + if source is ProviderSource.GLOBAL: + monkeypatch.setattr(trace, "get_tracer_provider", lambda: provider) + return OtelPluginConfig( + provider_source=source, + tracer_provider=provider if source is ProviderSource.EXPLICIT else None, + context_extractor=lambda _: Context(), + enrich_logger=False, + ) + + +@pytest.mark.parametrize( + "source", + [ + ProviderSource.EXPLICIT, + ProviderSource.GLOBAL, + ], +) +def test_unrelated_root_spans_keep_provider_id_generation( + source: ProviderSource, monkeypatch: pytest.MonkeyPatch +) -> None: + """Unrelated scopes receive fresh roots throughout a durable invocation.""" + provider, _ = _provider() + provider_generator = provider.id_generator + unrelated_tracer = provider.get_tracer("unrelated-library") + before = unrelated_tracer.start_span("before", context=Context()) + + plugin = InvocationOtelPlugin(_config_for_source(source, provider, monkeypatch)) + assert provider.id_generator is provider_generator + assert isinstance(plugin._id_generator, DeterministicIdGenerator) + + plugin.on_invocation_start(_invocation_start()) + workflow = plugin._workflow_span + assert workflow is not None + during = unrelated_tracer.start_span("during", context=Context()) + plugin.on_invocation_end(_invocation_end()) + after = unrelated_tracer.start_span("after", context=Context()) + + trace_ids = { + before.get_span_context().trace_id, + during.get_span_context().trace_id, + after.get_span_context().trace_id, + workflow.get_span_context().trace_id, + } + assert len(trace_ids) == 4 + + before.end() + during.end() + after.end() + + +def test_global_proxy_binds_sdk_provider_before_first_invocation( + monkeypatch: pytest.MonkeyPatch, +) -> None: + """A plugin created before global SDK setup binds when invocation starts.""" + monkeypatch.setattr(trace, "_TRACER_PROVIDER", None) + current_provider: list[ApiTracerProvider] = [ProxyTracerProvider()] + monkeypatch.setattr(trace, "get_tracer_provider", lambda: current_provider[0]) + plugin = InvocationOtelPlugin( + OtelPluginConfig( + provider_source=ProviderSource.GLOBAL, + context_extractor=lambda _: Context(), + enrich_logger=False, + ) + ) + + provider, exporter = _provider() + current_provider[0] = provider + plugin.on_invocation_start(_invocation_start()) + plugin.on_invocation_end(_invocation_end()) + + assert {span.name for span in exporter.get_finished_spans()} == { + "Invocation", + "Workflow", + } + + +def test_global_proxy_disables_entire_invocation_until_sdk_provider_is_ready( + monkeypatch: pytest.MonkeyPatch, caplog: pytest.LogCaptureFixture +) -> None: + """Provider setup midway through an invocation cannot produce a partial trace.""" + monkeypatch.setattr(trace, "_TRACER_PROVIDER", None) + current_provider: list[ApiTracerProvider] = [ProxyTracerProvider()] + monkeypatch.setattr(trace, "get_tracer_provider", lambda: current_provider[0]) + plugin = InvocationOtelPlugin( + OtelPluginConfig( + provider_source=ProviderSource.GLOBAL, + context_extractor=lambda _: Context(), + enrich_logger=False, + ) + ) + provider, exporter = _provider() + + plugin.on_invocation_start(_invocation_start()) + assert "telemetry is disabled for this invocation" in caplog.text + + current_provider[0] = provider + _run_step_lifecycle(plugin) + plugin.on_invocation_end(_invocation_end()) + assert exporter.get_finished_spans() == () + + plugin.on_invocation_start(_invocation_start()) + _run_step_lifecycle(plugin) + plugin.on_invocation_end(_invocation_end()) + assert {span.name for span in exporter.get_finished_spans()} == { + "Invocation", + "Workflow", + OP_NAME, + f"{OP_NAME} attempt 1", + } + + def _assert_hierarchy(exporter: InMemorySpanExporter) -> None: spans = {s.name: s for s in exporter.get_finished_spans()} + workflow = spans["Workflow"] invocation = spans["Invocation"] operation = spans[OP_NAME] attempt = spans[f"{OP_NAME} attempt 1"] @@ -170,6 +294,7 @@ def _assert_hierarchy(exporter: InMemorySpanExporter) -> None: # Invocation span is a root (empty extracted context) and records status. assert invocation.parent is None assert invocation.kind is SpanKind.INTERNAL + assert invocation.context.trace_id != workflow.context.trace_id assert invocation.attributes is not None assert ( invocation.attributes["durable.invocation.status"] @@ -186,6 +311,46 @@ def _assert_hierarchy(exporter: InMemorySpanExporter) -> None: assert attempt.parent.span_id == operation.context.span_id +@pytest.mark.parametrize( + "source", + [ + ProviderSource.EXPLICIT, + ProviderSource.GLOBAL, + ], +) +def test_invocation_span_parents_to_ambient_for_all_provider_sources( + source: ProviderSource, monkeypatch: pytest.MonkeyPatch +) -> None: + provider, exporter = _provider() + plugin = InvocationOtelPlugin(_config_for_source(source, provider, monkeypatch)) + + ambient = provider.get_tracer("ambient").start_span("lambda-invocation") + token = otel_context.attach(trace.set_span_in_context(ambient)) + try: + plugin.on_invocation_start(_invocation_start()) + _run_step_lifecycle(plugin) + plugin.on_invocation_end(_invocation_end()) + finally: + otel_context.detach(token) + ambient.end() + + spans = {span.name: span for span in exporter.get_finished_spans()} + workflow = spans["Workflow"] + invocation = spans["Invocation"] + operation = spans[OP_NAME] + + assert workflow.parent is None + assert workflow.context.trace_id != ambient.get_span_context().trace_id + assert invocation.parent is not None + assert invocation.parent.span_id == ambient.get_span_context().span_id + assert invocation.context.trace_id == ambient.get_span_context().trace_id + assert operation.parent is not None + assert operation.parent.span_id == invocation.context.span_id + assert {link.context.span_id for link in operation.links} == { + derive_workflow_span_id(EXECUTION_ARN) + } + + # --------------------------------------------------------------------------- # Community collector layer (caller-supplied provider) # --------------------------------------------------------------------------- @@ -231,8 +396,8 @@ def test_adot_layer_full_lifecycle_uses_global_provider(monkeypatch): _assert_hierarchy(exporter) -def test_second_plugin_configures_cached_tracer_generator(monkeypatch): - """A second handler's Workflow span uses its deterministic trace ID.""" +def test_second_plugin_uses_execution_trace_id_independent_of_xray(monkeypatch): + """Workflow trace IDs remain deterministic and separate from X-Ray.""" provider, exporter = _provider() first_plugin = InvocationOtelPlugin( OtelPluginConfig( @@ -252,11 +417,13 @@ def test_second_plugin_configures_cached_tracer_generator(monkeypatch): ) monkeypatch.setenv("_X_AMZN_TRACE_ID", XRAY_TRACE_HEADER) - assert target_plugin._tracer is first_plugin._tracer + if target_plugin._tracer is first_plugin._tracer: + assert target_plugin._id_generator is first_plugin._id_generator target_plugin.on_invocation_start(_invocation_start()) target_plugin.on_invocation_end(_invocation_end()) workflow = next( span for span in exporter.get_finished_spans() if span.name == "Workflow" ) - assert workflow.context.trace_id == XRAY_TRACE_ID + assert workflow.context.trace_id == EXECUTION_TRACE_ID + assert workflow.context.trace_id != XRAY_TRACE_ID diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_provider.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_provider.py index e792c2ff..775df937 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_provider.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_provider.py @@ -5,20 +5,12 @@ import pytest from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider -from opentelemetry.sdk.trace.sampling import ALWAYS_ON, TraceIdRatioBased from aws_durable_execution_sdk_python_otel.otel_plugin_config import ( OtelPluginConfig, - ExporterConfig, ProviderSource, ) -from aws_durable_execution_sdk_python_otel.provider import ( - SAMPLING_RATIO_ENV, - _build_resource, - _build_sampler, - _resolve_endpoint, - create_tracer_provider, -) +from aws_durable_execution_sdk_python_otel.provider import create_tracer_provider def test_explicit_provider_is_used(): @@ -47,14 +39,6 @@ def test_unset_config_defaults_to_global_provider(): assert result.tracer_provider is trace.get_tracer_provider() -def test_auto_otlp_source_builds_sdk_provider(): - result = create_tracer_provider( - OtelPluginConfig(provider_source=ProviderSource.AUTO_OTLP) - ) - assert result.source is ProviderSource.AUTO_OTLP - assert isinstance(result.tracer_provider, TracerProvider) - - # --------------------------------------------------------------------------- # Config validation (each source has the fields it needs) # --------------------------------------------------------------------------- @@ -74,61 +58,3 @@ def test_global_source_rejects_tracer_provider(): OtelPluginConfig( provider_source=ProviderSource.GLOBAL, tracer_provider=TracerProvider() ) - - -# --------------------------------------------------------------------------- -# Sampler resolution -# --------------------------------------------------------------------------- -def test_sampler_defaults_to_always_on(monkeypatch): - monkeypatch.delenv(SAMPLING_RATIO_ENV, raising=False) - assert _build_sampler() is ALWAYS_ON - - -def test_sampler_uses_ratio_when_valid(monkeypatch): - monkeypatch.setenv(SAMPLING_RATIO_ENV, "0.25") - assert isinstance(_build_sampler(), TraceIdRatioBased) - - -@pytest.mark.parametrize("bad", ["not-a-number", "1.5", "-0.1"]) -def test_sampler_falls_back_to_always_on_for_invalid_ratio(monkeypatch, bad): - monkeypatch.setenv(SAMPLING_RATIO_ENV, bad) - assert _build_sampler() is ALWAYS_ON - - -# --------------------------------------------------------------------------- -# Resource + endpoint resolution -# --------------------------------------------------------------------------- -def test_resource_is_none_without_function_name(monkeypatch): - monkeypatch.delenv("AWS_LAMBDA_FUNCTION_NAME", raising=False) - assert _build_resource() is None - - -def test_resource_populates_lambda_attributes(monkeypatch): - monkeypatch.setenv("AWS_LAMBDA_FUNCTION_NAME", "my-fn") - monkeypatch.setenv("AWS_REGION", "us-west-2") - monkeypatch.setenv("AWS_LAMBDA_FUNCTION_VERSION", "7") - resource = _build_resource() - assert resource is not None - attrs = resource.attributes - assert attrs["service.name"] == "my-fn" - assert attrs["faas.name"] == "my-fn" - assert attrs["cloud.provider"] == "aws" - assert attrs["cloud.platform"] == "aws_lambda" - assert attrs["cloud.region"] == "us-west-2" - assert attrs["faas.version"] == "7" - - -def test_resolve_endpoint_appends_signal_path(monkeypatch): - monkeypatch.delenv("OTEL_EXPORTER_OTLP_ENDPOINT", raising=False) - config = OtelPluginConfig( - exporter_config=ExporterConfig(endpoint="http://collector:4318") - ) - assert _resolve_endpoint(config) == "http://collector:4318/v1/traces" - - -def test_resolve_endpoint_does_not_double_append(monkeypatch): - monkeypatch.delenv("OTEL_EXPORTER_OTLP_ENDPOINT", raising=False) - config = OtelPluginConfig( - exporter_config=ExporterConfig(endpoint="http://collector:4318/v1/traces") - ) - assert _resolve_endpoint(config) == "http://collector:4318/v1/traces" diff --git a/pyproject.toml b/pyproject.toml index 6ea8e7c1..9ac77a74 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -50,7 +50,6 @@ extra-dependencies = [ "boto3-stubs[lambda]", "opentelemetry-sdk>=1.20.0", "opentelemetry-instrumentation-botocore", - "opentelemetry-instrumentation-urllib3", ] [tool.hatch.envs.types.scripts] From 92faeef467ca94b6ea49f840c8bd1ee9fcc8a78f Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Sat, 15 Aug 2026 20:58:10 +0000 Subject: [PATCH 2/3] refactor(otel): infer provider selection --- .../src/otel/otel_logger_example.py | 12 +---- .../src/plugin/execution_with_otel.py | 12 +---- .../README.md | 8 +-- .../__init__.py | 2 - .../execution_plugin.py | 12 ++--- .../instrumentations.py | 19 +++---- .../invocation_plugin.py | 24 +++------ .../otel_plugin_config.py | 44 ++-------------- .../provider.py | 46 +++++------------ .../tests/test_execution_plugin.py | 13 ++--- .../test_execution_plugin_integration.py | 42 ++++++--------- .../tests/test_invocation_plugin.py | 7 +-- .../test_invocation_plugin_integration.py | 50 ++++++++---------- .../tests/test_log_filter.py | 6 +-- .../tests/test_provider.py | 51 +++++-------------- 15 files changed, 98 insertions(+), 250 deletions(-) diff --git a/packages/aws-durable-execution-sdk-python-examples/src/otel/otel_logger_example.py b/packages/aws-durable-execution-sdk-python-examples/src/otel/otel_logger_example.py index fe74d7e3..a727bf73 100644 --- a/packages/aws-durable-execution-sdk-python-examples/src/otel/otel_logger_example.py +++ b/packages/aws-durable-execution-sdk-python-examples/src/otel/otel_logger_example.py @@ -16,11 +16,7 @@ from typing import Any -from aws_durable_execution_sdk_python_otel import ( - InvocationOtelPlugin, - OtelPluginConfig, - ProviderSource, -) +from aws_durable_execution_sdk_python_otel import InvocationOtelPlugin from aws_durable_execution_sdk_python import StepContext from aws_durable_execution_sdk_python.context import ( @@ -48,11 +44,7 @@ def greet_in_child(child_context: DurableContext, name: str) -> str: return result -@durable_execution( - plugins=[ - InvocationOtelPlugin(OtelPluginConfig(provider_source=ProviderSource.GLOBAL)) - ] -) +@durable_execution(plugins=[InvocationOtelPlugin()]) def handler(_event: Any, context: DurableContext) -> str: # Logged at the top level: enriched with the invocation span_id. context.logger.info("Workflow started") diff --git a/packages/aws-durable-execution-sdk-python-examples/src/plugin/execution_with_otel.py b/packages/aws-durable-execution-sdk-python-examples/src/plugin/execution_with_otel.py index b1cff3c5..3d001d46 100644 --- a/packages/aws-durable-execution-sdk-python-examples/src/plugin/execution_with_otel.py +++ b/packages/aws-durable-execution-sdk-python-examples/src/plugin/execution_with_otel.py @@ -2,11 +2,7 @@ from typing import Any -from aws_durable_execution_sdk_python_otel import ( - InvocationOtelPlugin, - OtelPluginConfig, - ProviderSource, -) +from aws_durable_execution_sdk_python_otel import InvocationOtelPlugin from aws_durable_execution_sdk_python import StepContext from aws_durable_execution_sdk_python.config import Duration @@ -36,11 +32,7 @@ def add_numbers_in_child(child_context: DurableContext, a: int, b: int): return result -@durable_execution( - plugins=[ - InvocationOtelPlugin(OtelPluginConfig(provider_source=ProviderSource.GLOBAL)) - ] -) +@durable_execution(plugins=[InvocationOtelPlugin()]) def handler(_event: Any, context: DurableContext) -> int: result = 0 for i in range(3): diff --git a/packages/aws-durable-execution-sdk-python-otel/README.md b/packages/aws-durable-execution-sdk-python-otel/README.md index 4a6c5eba..89a58302 100644 --- a/packages/aws-durable-execution-sdk-python-otel/README.md +++ b/packages/aws-durable-execution-sdk-python-otel/README.md @@ -184,14 +184,11 @@ See the [ADOT sampling configuration](https://aws-otel.github.io/docs/getting-st from aws_durable_execution_sdk_python_otel import ( InvocationOtelPlugin, OtelPluginConfig, - ProviderSource, xray_context_extractor, ) plugin = InvocationOtelPlugin( OtelPluginConfig( - # Use the global provider configured by ADOT (the default). - provider_source=ProviderSource.GLOBAL, # Use a custom context extractor (default: xray_context_extractor). context_extractor=xray_context_extractor, # Custom instrumentation scope name @@ -271,7 +268,6 @@ The main plugin class. Implements `DurableInstrumentationPlugin` from `aws_durab ```python InvocationOtelPlugin( OtelPluginConfig( - provider_source=ProviderSource.GLOBAL, tracer_provider=None, context_extractor=None, instrument_name="aws-durable-execution-sdk-python", @@ -281,8 +277,8 @@ InvocationOtelPlugin( ) ``` -Set `provider_source=ProviderSource.EXPLICIT` and pass `tracer_provider=...` -when the application owns the OpenTelemetry SDK provider. +Pass `tracer_provider=...` when the application owns the OpenTelemetry SDK +provider. When omitted, the globally configured provider is used. ### `DeterministicIdGenerator` diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/__init__.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/__init__.py index 6547aade..f281ee3e 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/__init__.py +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/__init__.py @@ -16,7 +16,6 @@ ) from aws_durable_execution_sdk_python_otel.otel_plugin_config import ( OtelPluginConfig, - ProviderSource, ) from aws_durable_execution_sdk_python_otel.instrumentations import ( register_standalone_instrumentations, @@ -43,7 +42,6 @@ "InvocationOtelPlugin", "OtelContextLogFilter", "ProviderResult", - "ProviderSource", "create_tracer_provider", "derive_workflow_span_id", "install_log_filter", diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py index 5967e179..4a09b7a7 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py @@ -63,10 +63,7 @@ derive_workflow_span_id, operation_id_to_span_id, ) -from aws_durable_execution_sdk_python_otel.otel_plugin_config import ( - OtelPluginConfig, - ProviderSource, -) +from aws_durable_execution_sdk_python_otel.otel_plugin_config import OtelPluginConfig from aws_durable_execution_sdk_python_otel.instrumentations import ( register_standalone_instrumentations, ) @@ -109,10 +106,7 @@ def __init__(self, config: OtelPluginConfig | None = None) -> None: result = create_tracer_provider(self._config) self._provider = result.tracer_provider - # GLOBAL (ADOT) mode parents the Invocation span to the ambient Lambda - # invocation span instead of the Workflow span (see - # _start_invocation_span). - self._provider_source = result.source + self._uses_global_provider = result.uses_global_provider self._tracer: Tracer = self._provider.get_tracer(self._config.instrument_name) self._id_generator = DeterministicIdGenerator() @@ -140,7 +134,7 @@ def _bind_sdk_tracer(self) -> bool: """Bind to an SDK tracer, retrying a deferred global provider.""" tracer = self._tracer if not isinstance(tracer, SdkTracer): - if self._provider_source is ProviderSource.GLOBAL: + if self._uses_global_provider: self._provider = trace.get_tracer_provider() tracer = self._provider.get_tracer(self._config.instrument_name) self._tracer = tracer diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/instrumentations.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/instrumentations.py index e4b887bd..f3faf63b 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/instrumentations.py +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/instrumentations.py @@ -3,8 +3,8 @@ Mirrors the JS ``registerStandaloneInstrumentations``: * A custom (explicit) provider skips ALL instrumentation registration. -* When the global provider is in use (``ProviderSource.GLOBAL``), only the - AWS SDK instrumentation is registered (not HTTP). +* When the global provider is in use, only the AWS SDK instrumentation is + registered (not HTTP). The JS SDK uses ``AwsInstrumentation`` (AWS SDK v3). The Python equivalent is ``BotocoreInstrumentor`` because boto3/botocore is the AWS SDK for Python. The @@ -18,8 +18,6 @@ import logging from typing import TYPE_CHECKING -from aws_durable_execution_sdk_python_otel.otel_plugin_config import ProviderSource - if TYPE_CHECKING: from aws_durable_execution_sdk_python_otel.provider import ProviderResult @@ -48,16 +46,13 @@ def _register_aws_instrumentation(tracer_provider: object | None) -> None: def register_standalone_instrumentations(result: ProviderResult) -> None: - """Register AWS SDK instrumentation per the resolved source. + """Register AWS SDK instrumentation for the global provider. Args: - result: The resolved provider and its :class:`ProviderSource`. + result: The resolved provider and how it was selected. """ - if result.source is ProviderSource.EXPLICIT: - # Caller manages their own instrumentation: skip everything. + if not result.uses_global_provider: + # Applications that supply a provider manage their own instrumentation. return - if result.source is ProviderSource.GLOBAL: - # Global provider: register AWS instrumentation only. - _register_aws_instrumentation(None) - return + _register_aws_instrumentation(None) diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py index cd52e346..0c9b7f6c 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py @@ -44,10 +44,7 @@ operation_id_to_span_id, ) from aws_durable_execution_sdk_python_otel.log_filter import install_log_filter -from aws_durable_execution_sdk_python_otel.otel_plugin_config import ( - OtelPluginConfig, - ProviderSource, -) +from aws_durable_execution_sdk_python_otel.otel_plugin_config import OtelPluginConfig from aws_durable_execution_sdk_python_otel.provider import create_tracer_provider from aws_durable_execution_sdk_python_otel.instrumentations import ( register_standalone_instrumentations, @@ -87,10 +84,9 @@ class InvocationOtelPlugin(DurableInstrumentationPlugin): Args: config: Shared plugin configuration (the same OtelPluginConfig accepted by ExecutionOtelPlugin). When omitted, defaults are used (X-Ray - extractor, "Workflow" span name, log enrichment on). Like - ExecutionOtelPlugin and the JS SDK plugins, the default - ``provider_source`` is ``GLOBAL``: the plugin uses the globally - configured tracer provider (e.g. the ADOT Lambda layer). + extractor, "Workflow" span name, log enrichment on) and the plugin + uses the globally configured tracer provider (for example, the + provider installed by the ADOT Lambda layer). """ DEFAULT_INSTRUMENT_NAME = "aws-durable-execution-sdk-python" @@ -100,9 +96,8 @@ def __init__(self, config: OtelPluginConfig | None = None) -> None: Accepts the same OtelPluginConfig as ExecutionOtelPlugin so both plugins share one configuration surface (context extractor, instrumentation name, - provider selection, and log enrichment). Like ExecutionOtelPlugin and the - JS SDK plugins, the default ``provider_source`` is ``GLOBAL``: it uses the - globally configured (e.g. ADOT) provider. + provider selection, and log enrichment). When no provider is supplied, + the globally configured provider is used. The plugin tracer is configured with a scoped deterministic ID generator so durable spans share stable identifiers without changing unrelated @@ -119,12 +114,9 @@ def __init__(self, config: OtelPluginConfig | None = None) -> None: self._workflow_span_name = self._config.workflow_span_name self._enrich_logger = self._config.enrich_logger - # Like ExecutionOtelPlugin (and the JS SDK plugins), InvocationOtelPlugin - # defaults to provider_source=GLOBAL (the globally configured, e.g. ADOT, - # provider). result = create_tracer_provider(self._config) self._provider = result.tracer_provider - self._provider_source = result.source + self._uses_global_provider = result.uses_global_provider self._tracer: Tracer = self._provider.get_tracer(self._config.instrument_name) self._id_generator = DeterministicIdGenerator() self._bind_sdk_tracer() @@ -155,7 +147,7 @@ def _bind_sdk_tracer(self) -> bool: """Bind to an SDK tracer, retrying a deferred global provider.""" tracer = self._tracer if not isinstance(tracer, SdkTracer): - if self._provider_source is ProviderSource.GLOBAL: + if self._uses_global_provider: self._provider = trace.get_tracer_provider() tracer = self._provider.get_tracer(self._config.instrument_name) self._tracer = tracer diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/otel_plugin_config.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/otel_plugin_config.py index eedd85ee..c1d9896d 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/otel_plugin_config.py +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/otel_plugin_config.py @@ -8,7 +8,6 @@ from __future__ import annotations from dataclasses import dataclass -from enum import Enum from typing import TYPE_CHECKING @@ -24,17 +23,6 @@ DEFAULT_WORKFLOW_SPAN_NAME = "Workflow" -class ProviderSource(Enum): - """Which tracer-provider tier an :class:`OtelPluginConfig` selects. - - The single value that drives provider construction (``create_tracer_provider``) - and the plugins' instrumentation, span-parenting and flush decisions. - """ - - EXPLICIT = "explicit" # use config.tracer_provider as-is - GLOBAL = "global" # default: use the global provider (trace.get_tracer_provider()) - - @dataclass class OtelPluginConfig: """Canonical configuration shared by both OTel plugins. @@ -43,14 +31,11 @@ class OtelPluginConfig: are ignored without error by :class:`InvocationOtelPlugin`. Attributes: - provider_source: Selects how the tracer provider is obtained - (:class:`ProviderSource`). Defaults to ``GLOBAL`` (uses the globally - configured provider, e.g. the ADOT Lambda layer, via - ``trace.get_tracer_provider()``). ``EXPLICIT`` uses - ``tracer_provider`` as-is and skips instrumentation registration. - tracer_provider: The provider used when ``provider_source`` is - ``EXPLICIT``. Required in that case and must be left unset for - ``GLOBAL``. + tracer_provider: An application-owned provider to use as-is. When + omitted, the globally configured provider is used (for example, the + provider installed by the ADOT Lambda layer). Standalone + instrumentation registration is skipped for an application-owned + provider. context_extractor: Upstream trace-context extractor. Defaults to the X-Ray extractor when omitted. instrument_name: Instrumentation scope name. @@ -58,27 +43,8 @@ class OtelPluginConfig: enrich_logger: Install the root-logger OTel context filter. """ - provider_source: ProviderSource = ProviderSource.GLOBAL tracer_provider: SdkTracerProvider | None = None context_extractor: ContextExtractor | None = None instrument_name: str = DEFAULT_INSTRUMENT_NAME workflow_span_name: str = DEFAULT_WORKFLOW_SPAN_NAME enrich_logger: bool = True - - def __post_init__(self) -> None: - """Validate that each provider source has the fields it requires. - - The config is fully driven by :attr:`provider_source`; ``tracer_provider`` - is the one source-specific field, so it must be present for ``EXPLICIT`` - and absent for ``GLOBAL`` (where it would be silently ignored). - """ - if self.provider_source is ProviderSource.EXPLICIT: - if self.tracer_provider is None: - raise ValueError("provider_source=EXPLICIT requires a tracer_provider.") - elif self.tracer_provider is not None: - raise ValueError( - "tracer_provider is only valid with provider_source=EXPLICIT; " - f"got provider_source={self.provider_source.name}. Set " - "ProviderSource.EXPLICIT, or drop tracer_provider for " - "GLOBAL." - ) diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/provider.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/provider.py index a8a0f018..9aa7b30a 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/provider.py +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/provider.py @@ -1,11 +1,4 @@ -"""Shared TracerProvider factory for the durable-execution OTel plugins. - -Builds the tracer provider selected by the config's -:class:`~aws_durable_execution_sdk_python_otel.otel_plugin_config.ProviderSource`: - -1. ``EXPLICIT`` - the config's ``tracer_provider`` is used as-is. -2. ``GLOBAL`` - the globally configured provider is used (e.g. ADOT layer). -""" +"""Shared TracerProvider factory for the durable-execution OTel plugins.""" from __future__ import annotations @@ -14,33 +7,27 @@ from opentelemetry import trace -from aws_durable_execution_sdk_python_otel.otel_plugin_config import ( - OtelPluginConfig, - ProviderSource, -) +from aws_durable_execution_sdk_python_otel.otel_plugin_config import OtelPluginConfig if TYPE_CHECKING: from opentelemetry.trace import TracerProvider -@dataclass +@dataclass(frozen=True) class ProviderResult: """Result of provider resolution: the provider and how it was chosen.""" tracer_provider: TracerProvider - source: ProviderSource + uses_global_provider: bool def create_tracer_provider(config: OtelPluginConfig) -> ProviderResult: - """Resolve a TracerProvider from the config's :attr:`provider_source`. + """Resolve the configured provider or the global provider. - A straight switch on ``config.provider_source``; the chosen tier is reported - back as :class:`ProviderSource` so callers make the instrumentation/flush - decision off a single value: - - 1. ``EXPLICIT`` -> ``config.tracer_provider`` used as-is - 2. ``GLOBAL`` -> the globally configured provider + Whether ``tracer_provider`` was supplied is retained separately from the + resolved object. An explicit provider may also be installed globally, but it + still has application-owned instrumentation and initialization behavior. Args: config: Shared plugin configuration. @@ -48,16 +35,11 @@ def create_tracer_provider(config: OtelPluginConfig) -> ProviderResult: Returns: A :class:`ProviderResult`. """ - source = config.provider_source - - if source is ProviderSource.EXPLICIT: - # Explicit provider: use as-is, never wrap/modify. OtelPluginConfig - # validation guarantees tracer_provider is set for EXPLICIT. - assert config.tracer_provider is not None + if config.tracer_provider is not None: provider: TracerProvider = config.tracer_provider - elif source is ProviderSource.GLOBAL: - provider = trace.get_tracer_provider() - else: # pragma: no cover - exhaustive over ProviderSource - raise ValueError(f"unknown provider_source: {source!r}") + return ProviderResult(provider, uses_global_provider=False) - return ProviderResult(provider, source) + return ProviderResult( + trace.get_tracer_provider(), + uses_global_provider=True, + ) diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py index 171dae87..d9057dd2 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin.py @@ -33,10 +33,7 @@ operation_id_to_span_id, ) from aws_durable_execution_sdk_python_otel.execution_plugin import ExecutionOtelPlugin -from aws_durable_execution_sdk_python_otel.otel_plugin_config import ( - OtelPluginConfig, - ProviderSource, -) +from aws_durable_execution_sdk_python_otel.otel_plugin_config import OtelPluginConfig START_TIME = datetime(2024, 1, 2, 3, 4, 5, tzinfo=UTC) @@ -65,7 +62,6 @@ def _create_plugin() -> tuple[ExecutionOtelPlugin, InMemorySpanExporter]: provider.add_span_processor(SimpleSpanProcessor(exporter)) plugin = ExecutionOtelPlugin( OtelPluginConfig( - provider_source=ProviderSource.EXPLICIT, tracer_provider=provider, context_extractor=lambda _: Context(), enrich_logger=False, @@ -415,11 +411,11 @@ def test_step_attempt_span_omits_operation_status(): def _create_default_mode_plugin( monkeypatch, ) -> tuple[ExecutionOtelPlugin, InMemorySpanExporter]: - """ExecutionOtelPlugin in GLOBAL (ADOT) mode wired to an in-memory exporter. + """ExecutionOtelPlugin in global (ADOT) mode wired to an in-memory exporter. The capture provider is installed as the global provider so - ``provider_source=GLOBAL`` resolves to it, letting the test assert spans - while exercising the ambient-parenting path. + the default configuration resolves to it, letting the test assert spans while + exercising the ambient-parenting path. """ exporter = InMemorySpanExporter() provider = TracerProvider() @@ -427,7 +423,6 @@ def _create_default_mode_plugin( monkeypatch.setattr(trace, "get_tracer_provider", lambda: provider) plugin = ExecutionOtelPlugin( OtelPluginConfig( - provider_source=ProviderSource.GLOBAL, context_extractor=lambda _: Context(), enrich_logger=False, ) diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin_integration.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin_integration.py index bc48159b..76d59269 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin_integration.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_execution_plugin_integration.py @@ -3,12 +3,11 @@ Drives the full plugin lifecycle against a real TracerProvider + InMemorySpanExporter for the two deployment shapes: -* Community collector layer: the caller supplies a provider - (``provider_source=EXPLICIT``); the Workflow and Invocation spans root - separate traces when no ambient parent exists. +* Community collector layer: the caller supplies a provider; the Workflow and + Invocation spans root separate traces when no ambient parent exists. * ADOT layer: the ADOT Lambda layer supplies the global provider and the ambient - Lambda invocation span (``provider_source=GLOBAL``); the plugin's Invocation - span parents to that ambient span. + Lambda invocation span; the plugin's Invocation span parents to that ambient + span. """ from __future__ import annotations @@ -48,10 +47,7 @@ operation_id_to_span_id, ) from aws_durable_execution_sdk_python_otel.execution_plugin import ExecutionOtelPlugin -from aws_durable_execution_sdk_python_otel.otel_plugin_config import ( - OtelPluginConfig, - ProviderSource, -) +from aws_durable_execution_sdk_python_otel.otel_plugin_config import OtelPluginConfig START_TIME = datetime(2024, 1, 2, 3, 4, 5, tzinfo=UTC) @@ -165,30 +161,27 @@ def _run_step_lifecycle(plugin: ExecutionOtelPlugin) -> None: ) -def _config_for_source( - source: ProviderSource, +def _config_for_provider( + uses_global_provider: bool, provider: TracerProvider, monkeypatch: pytest.MonkeyPatch, ) -> OtelPluginConfig: - if source is ProviderSource.GLOBAL: + if uses_global_provider: monkeypatch.setattr(trace, "get_tracer_provider", lambda: provider) return OtelPluginConfig( - provider_source=source, - tracer_provider=provider if source is ProviderSource.EXPLICIT else None, + tracer_provider=None if uses_global_provider else provider, context_extractor=lambda _: Context(), enrich_logger=False, ) @pytest.mark.parametrize( - "source", - [ - ProviderSource.EXPLICIT, - ProviderSource.GLOBAL, - ], + "uses_global_provider", + [False, True], + ids=["explicit", "global"], ) def test_unrelated_root_spans_keep_provider_id_generation( - source: ProviderSource, monkeypatch: pytest.MonkeyPatch + uses_global_provider: bool, monkeypatch: pytest.MonkeyPatch ) -> None: """Unrelated scopes receive fresh roots throughout a durable invocation.""" provider, _ = _provider() @@ -196,7 +189,9 @@ def test_unrelated_root_spans_keep_provider_id_generation( unrelated_tracer = provider.get_tracer("unrelated-library") before = unrelated_tracer.start_span("before", context=Context()) - plugin = ExecutionOtelPlugin(_config_for_source(source, provider, monkeypatch)) + plugin = ExecutionOtelPlugin( + _config_for_provider(uses_global_provider, provider, monkeypatch) + ) assert provider.id_generator is provider_generator assert isinstance(plugin._id_generator, DeterministicIdGenerator) @@ -229,7 +224,6 @@ def test_global_proxy_binds_sdk_provider_before_first_invocation( monkeypatch.setattr(trace, "get_tracer_provider", lambda: current_provider[0]) plugin = ExecutionOtelPlugin( OtelPluginConfig( - provider_source=ProviderSource.GLOBAL, context_extractor=lambda _: Context(), enrich_logger=False, ) @@ -255,7 +249,6 @@ def test_global_proxy_disables_entire_invocation_until_sdk_provider_is_ready( monkeypatch.setattr(trace, "get_tracer_provider", lambda: current_provider[0]) plugin = ExecutionOtelPlugin( OtelPluginConfig( - provider_source=ProviderSource.GLOBAL, context_extractor=lambda _: Context(), enrich_logger=False, ) @@ -288,7 +281,6 @@ def test_community_layer_full_lifecycle_is_workflow_rooted(): provider, exporter = _provider() plugin = ExecutionOtelPlugin( OtelPluginConfig( - provider_source=ProviderSource.EXPLICIT, tracer_provider=provider, context_extractor=lambda _: Context(), enrich_logger=False, @@ -341,7 +333,6 @@ def test_adot_layer_full_lifecycle_parents_to_ambient_span(monkeypatch): monkeypatch.setattr(trace, "get_tracer_provider", lambda: provider) plugin = ExecutionOtelPlugin( OtelPluginConfig( - provider_source=ProviderSource.GLOBAL, context_extractor=lambda _: Context(), enrich_logger=False, ) @@ -387,7 +378,6 @@ def test_second_plugin_uses_execution_trace_id_independent_of_xray(monkeypatch): provider, exporter = _provider() monkeypatch.setattr(trace, "get_tracer_provider", lambda: provider) config = OtelPluginConfig( - provider_source=ProviderSource.GLOBAL, context_extractor=lambda _: Context(), enrich_logger=False, ) diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py index 204b976d..3e7b43dc 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin.py @@ -36,10 +36,7 @@ operation_id_to_span_id, ) from aws_durable_execution_sdk_python_otel.invocation_plugin import InvocationOtelPlugin -from aws_durable_execution_sdk_python_otel.otel_plugin_config import ( - OtelPluginConfig, - ProviderSource, -) +from aws_durable_execution_sdk_python_otel.otel_plugin_config import OtelPluginConfig START_TIME = datetime(2024, 1, 2, 3, 4, 5, tzinfo=UTC) @@ -68,7 +65,6 @@ def _create_plugin() -> tuple[InvocationOtelPlugin, InMemorySpanExporter]: trace_provider.add_span_processor(SimpleSpanProcessor(exporter)) plugin = InvocationOtelPlugin( OtelPluginConfig( - provider_source=ProviderSource.EXPLICIT, tracer_provider=trace_provider, context_extractor=lambda _: Context(), ) @@ -1115,7 +1111,6 @@ def test_workflow_span_name_is_configurable(): trace_provider.add_span_processor(SimpleSpanProcessor(exporter)) plugin = InvocationOtelPlugin( OtelPluginConfig( - provider_source=ProviderSource.EXPLICIT, tracer_provider=trace_provider, context_extractor=lambda _: Context(), workflow_span_name="MyExecution", diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin_integration.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin_integration.py index a4631057..df717739 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin_integration.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_invocation_plugin_integration.py @@ -51,10 +51,7 @@ operation_id_to_span_id, ) from aws_durable_execution_sdk_python_otel.invocation_plugin import InvocationOtelPlugin -from aws_durable_execution_sdk_python_otel.otel_plugin_config import ( - OtelPluginConfig, - ProviderSource, -) +from aws_durable_execution_sdk_python_otel.otel_plugin_config import OtelPluginConfig START_TIME = datetime(2024, 1, 2, 3, 4, 5, tzinfo=UTC) @@ -168,30 +165,27 @@ def _run_step_lifecycle(plugin: InvocationOtelPlugin) -> None: ) -def _config_for_source( - source: ProviderSource, +def _config_for_provider( + uses_global_provider: bool, provider: TracerProvider, monkeypatch: pytest.MonkeyPatch, ) -> OtelPluginConfig: - if source is ProviderSource.GLOBAL: + if uses_global_provider: monkeypatch.setattr(trace, "get_tracer_provider", lambda: provider) return OtelPluginConfig( - provider_source=source, - tracer_provider=provider if source is ProviderSource.EXPLICIT else None, + tracer_provider=None if uses_global_provider else provider, context_extractor=lambda _: Context(), enrich_logger=False, ) @pytest.mark.parametrize( - "source", - [ - ProviderSource.EXPLICIT, - ProviderSource.GLOBAL, - ], + "uses_global_provider", + [False, True], + ids=["explicit", "global"], ) def test_unrelated_root_spans_keep_provider_id_generation( - source: ProviderSource, monkeypatch: pytest.MonkeyPatch + uses_global_provider: bool, monkeypatch: pytest.MonkeyPatch ) -> None: """Unrelated scopes receive fresh roots throughout a durable invocation.""" provider, _ = _provider() @@ -199,7 +193,9 @@ def test_unrelated_root_spans_keep_provider_id_generation( unrelated_tracer = provider.get_tracer("unrelated-library") before = unrelated_tracer.start_span("before", context=Context()) - plugin = InvocationOtelPlugin(_config_for_source(source, provider, monkeypatch)) + plugin = InvocationOtelPlugin( + _config_for_provider(uses_global_provider, provider, monkeypatch) + ) assert provider.id_generator is provider_generator assert isinstance(plugin._id_generator, DeterministicIdGenerator) @@ -232,7 +228,6 @@ def test_global_proxy_binds_sdk_provider_before_first_invocation( monkeypatch.setattr(trace, "get_tracer_provider", lambda: current_provider[0]) plugin = InvocationOtelPlugin( OtelPluginConfig( - provider_source=ProviderSource.GLOBAL, context_extractor=lambda _: Context(), enrich_logger=False, ) @@ -258,7 +253,6 @@ def test_global_proxy_disables_entire_invocation_until_sdk_provider_is_ready( monkeypatch.setattr(trace, "get_tracer_provider", lambda: current_provider[0]) plugin = InvocationOtelPlugin( OtelPluginConfig( - provider_source=ProviderSource.GLOBAL, context_extractor=lambda _: Context(), enrich_logger=False, ) @@ -312,17 +306,17 @@ def _assert_hierarchy(exporter: InMemorySpanExporter) -> None: @pytest.mark.parametrize( - "source", - [ - ProviderSource.EXPLICIT, - ProviderSource.GLOBAL, - ], + "uses_global_provider", + [False, True], + ids=["explicit", "global"], ) -def test_invocation_span_parents_to_ambient_for_all_provider_sources( - source: ProviderSource, monkeypatch: pytest.MonkeyPatch +def test_invocation_span_parents_to_ambient_for_all_provider_modes( + uses_global_provider: bool, monkeypatch: pytest.MonkeyPatch ) -> None: provider, exporter = _provider() - plugin = InvocationOtelPlugin(_config_for_source(source, provider, monkeypatch)) + plugin = InvocationOtelPlugin( + _config_for_provider(uses_global_provider, provider, monkeypatch) + ) ambient = provider.get_tracer("ambient").start_span("lambda-invocation") token = otel_context.attach(trace.set_span_in_context(ambient)) @@ -358,7 +352,6 @@ def test_community_layer_full_lifecycle_uses_supplied_provider(): provider, exporter = _provider() plugin = InvocationOtelPlugin( OtelPluginConfig( - provider_source=ProviderSource.EXPLICIT, tracer_provider=provider, context_extractor=lambda _: Context(), enrich_logger=False, @@ -382,7 +375,6 @@ def test_adot_layer_full_lifecycle_uses_global_provider(monkeypatch): plugin = InvocationOtelPlugin( OtelPluginConfig( - provider_source=ProviderSource.GLOBAL, context_extractor=lambda _: Context(), enrich_logger=False, ) @@ -401,7 +393,6 @@ def test_second_plugin_uses_execution_trace_id_independent_of_xray(monkeypatch): provider, exporter = _provider() first_plugin = InvocationOtelPlugin( OtelPluginConfig( - provider_source=ProviderSource.EXPLICIT, tracer_provider=provider, context_extractor=lambda _: Context(), enrich_logger=False, @@ -409,7 +400,6 @@ def test_second_plugin_uses_execution_trace_id_independent_of_xray(monkeypatch): ) target_plugin = InvocationOtelPlugin( OtelPluginConfig( - provider_source=ProviderSource.EXPLICIT, tracer_provider=provider, context_extractor=lambda _: Context(), enrich_logger=False, diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_log_filter.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_log_filter.py index f419f9ad..29645b85 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_log_filter.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_log_filter.py @@ -23,10 +23,7 @@ install_log_filter, ) from aws_durable_execution_sdk_python_otel.invocation_plugin import InvocationOtelPlugin -from aws_durable_execution_sdk_python_otel.otel_plugin_config import ( - OtelPluginConfig, - ProviderSource, -) +from aws_durable_execution_sdk_python_otel.otel_plugin_config import OtelPluginConfig START_TIME = datetime(2024, 1, 2, 3, 4, 5, tzinfo=UTC) @@ -42,7 +39,6 @@ def _create_plugin( trace_provider.add_span_processor(SimpleSpanProcessor(exporter)) plugin = InvocationOtelPlugin( OtelPluginConfig( - provider_source=ProviderSource.EXPLICIT, tracer_provider=trace_provider, context_extractor=lambda _: Context(), enrich_logger=enrich_logger, diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_provider.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_provider.py index 775df937..2493cf0c 100644 --- a/packages/aws-durable-execution-sdk-python-otel/tests/test_provider.py +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_provider.py @@ -6,55 +6,30 @@ from opentelemetry import trace from opentelemetry.sdk.trace import TracerProvider -from aws_durable_execution_sdk_python_otel.otel_plugin_config import ( - OtelPluginConfig, - ProviderSource, -) +from aws_durable_execution_sdk_python_otel.otel_plugin_config import OtelPluginConfig from aws_durable_execution_sdk_python_otel.provider import create_tracer_provider def test_explicit_provider_is_used(): provider = TracerProvider() - result = create_tracer_provider( - OtelPluginConfig( - provider_source=ProviderSource.EXPLICIT, tracer_provider=provider - ) - ) + result = create_tracer_provider(OtelPluginConfig(tracer_provider=provider)) assert result.tracer_provider is provider - assert result.source is ProviderSource.EXPLICIT + assert result.uses_global_provider is False -def test_global_source_returns_global_provider(): - result = create_tracer_provider( - OtelPluginConfig(provider_source=ProviderSource.GLOBAL) - ) - assert result.tracer_provider is trace.get_tracer_provider() - assert result.source is ProviderSource.GLOBAL - - -def test_unset_config_defaults_to_global_provider(): - # The default: no provider_source given -> use the global provider. +def test_unset_provider_uses_global_provider(): result = create_tracer_provider(OtelPluginConfig()) - assert result.source is ProviderSource.GLOBAL assert result.tracer_provider is trace.get_tracer_provider() + assert result.uses_global_provider is True -# --------------------------------------------------------------------------- -# Config validation (each source has the fields it needs) -# --------------------------------------------------------------------------- -def test_explicit_source_requires_tracer_provider(): - with pytest.raises(ValueError, match="requires a tracer_provider"): - OtelPluginConfig(provider_source=ProviderSource.EXPLICIT) - - -def test_tracer_provider_without_explicit_source_raises(): - # Default source is GLOBAL; a stray tracer_provider would be ignored. - with pytest.raises(ValueError, match="only valid with provider_source=EXPLICIT"): - OtelPluginConfig(tracer_provider=TracerProvider()) +def test_supplied_global_provider_remains_explicit( + monkeypatch: pytest.MonkeyPatch, +) -> None: + provider = TracerProvider() + monkeypatch.setattr(trace, "get_tracer_provider", lambda: provider) + result = create_tracer_provider(OtelPluginConfig(tracer_provider=provider)) -def test_global_source_rejects_tracer_provider(): - with pytest.raises(ValueError, match="only valid with provider_source=EXPLICIT"): - OtelPluginConfig( - provider_source=ProviderSource.GLOBAL, tracer_provider=TracerProvider() - ) + assert result.tracer_provider is provider + assert result.uses_global_provider is False From 1e39ed07cb1218ef3fa2a819bdac4feca3658206 Mon Sep 17 00:00:00 2001 From: Frank Chen Date: Sat, 15 Aug 2026 21:31:08 +0000 Subject: [PATCH 3/3] refactor(otel): remove botocore registration --- .../pyproject.toml | 7 --- .../__init__.py | 4 -- .../execution_plugin.py | 8 --- .../instrumentations.py | 58 ------------------- .../invocation_plugin.py | 8 --- .../tests/test_instrumentation_ownership.py | 48 +++++++++++++++ pyproject.toml | 1 - 7 files changed, 48 insertions(+), 86 deletions(-) delete mode 100644 packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/instrumentations.py create mode 100644 packages/aws-durable-execution-sdk-python-otel/tests/test_instrumentation_ownership.py diff --git a/packages/aws-durable-execution-sdk-python-otel/pyproject.toml b/packages/aws-durable-execution-sdk-python-otel/pyproject.toml index 106cc248..9df29594 100644 --- a/packages/aws-durable-execution-sdk-python-otel/pyproject.toml +++ b/packages/aws-durable-execution-sdk-python-otel/pyproject.toml @@ -32,13 +32,6 @@ dependencies = [ otel-invocation = "aws_durable_execution_sdk_python_otel.plugin_provider:INVOCATION_OTEL_PLUGIN_PROVIDER" otel-execution = "aws_durable_execution_sdk_python_otel.plugin_provider:EXECUTION_OTEL_PLUGIN_PROVIDER" -[project.optional-dependencies] -# Optional AWS SDK instrumentation for the global provider path. The -# instrumentations module degrades gracefully when it is absent. -instrumentation = [ - "opentelemetry-instrumentation-botocore", -] - [project.urls] Documentation = "https://github.com/aws/aws-durable-execution-sdk-python#readme" Issues = "https://github.com/aws/aws-durable-execution-sdk-python/issues" diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/__init__.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/__init__.py index f281ee3e..b6552896 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/__init__.py +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/__init__.py @@ -17,9 +17,6 @@ from aws_durable_execution_sdk_python_otel.otel_plugin_config import ( OtelPluginConfig, ) -from aws_durable_execution_sdk_python_otel.instrumentations import ( - register_standalone_instrumentations, -) from aws_durable_execution_sdk_python_otel.log_filter import ( OtelContextLogFilter, install_log_filter, @@ -46,7 +43,6 @@ "derive_workflow_span_id", "install_log_filter", "operation_id_to_span_id", - "register_standalone_instrumentations", "w3c_client_context_extractor", "xray_context_extractor", ] diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py index 4a09b7a7..889ede52 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/execution_plugin.py @@ -64,9 +64,6 @@ operation_id_to_span_id, ) from aws_durable_execution_sdk_python_otel.otel_plugin_config import OtelPluginConfig -from aws_durable_execution_sdk_python_otel.instrumentations import ( - register_standalone_instrumentations, -) from aws_durable_execution_sdk_python_otel.log_filter import install_log_filter from aws_durable_execution_sdk_python_otel.provider import create_tracer_provider @@ -112,11 +109,6 @@ def __init__(self, config: OtelPluginConfig | None = None) -> None: self._id_generator = DeterministicIdGenerator() self._bind_sdk_tracer() - try: - register_standalone_instrumentations(result) - except Exception: - logger.exception("Failed to register standalone instrumentations") - # Per-invocation state. self._execution_arn = "" self._execution_trace_id: int | None = None diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/instrumentations.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/instrumentations.py deleted file mode 100644 index f3faf63b..00000000 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/instrumentations.py +++ /dev/null @@ -1,58 +0,0 @@ -"""Shared instrumentation registration for the durable-execution OTel plugins. - -Mirrors the JS ``registerStandaloneInstrumentations``: - -* A custom (explicit) provider skips ALL instrumentation registration. -* When the global provider is in use, only the AWS SDK instrumentation is - registered (not HTTP). - -The JS SDK uses ``AwsInstrumentation`` (AWS SDK v3). The Python equivalent is -``BotocoreInstrumentor`` because boto3/botocore is the AWS SDK for Python. The -instrumentation package is an optional import: when it is not installed, -registration is skipped with a warning rather than raising, so the module stays -import-safe. -""" - -from __future__ import annotations - -import logging -from typing import TYPE_CHECKING - - -if TYPE_CHECKING: - from aws_durable_execution_sdk_python_otel.provider import ProviderResult - - -logger = logging.getLogger(__name__) - - -def _register_aws_instrumentation(tracer_provider: object | None) -> None: - """Register AWS SDK (botocore) instrumentation, if the package is available.""" - try: - from opentelemetry.instrumentation.botocore import BotocoreInstrumentor - except ImportError: - logger.warning( - "opentelemetry-instrumentation-botocore is not installed; " - "AWS SDK calls will not be traced. Install it to enable AWS " - "instrumentation." - ) - return - instrumentor = BotocoreInstrumentor() - if not instrumentor.is_instrumented_by_opentelemetry: - kwargs = {} - if tracer_provider is not None: - kwargs["tracer_provider"] = tracer_provider - instrumentor.instrument(**kwargs) - - -def register_standalone_instrumentations(result: ProviderResult) -> None: - """Register AWS SDK instrumentation for the global provider. - - Args: - result: The resolved provider and how it was selected. - """ - if not result.uses_global_provider: - # Applications that supply a provider manage their own instrumentation. - return - - _register_aws_instrumentation(None) diff --git a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py index 0c9b7f6c..9300b348 100644 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py +++ b/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/invocation_plugin.py @@ -46,9 +46,6 @@ from aws_durable_execution_sdk_python_otel.log_filter import install_log_filter from aws_durable_execution_sdk_python_otel.otel_plugin_config import OtelPluginConfig from aws_durable_execution_sdk_python_otel.provider import create_tracer_provider -from aws_durable_execution_sdk_python_otel.instrumentations import ( - register_standalone_instrumentations, -) logger = logging.getLogger(__name__) @@ -121,11 +118,6 @@ def __init__(self, config: OtelPluginConfig | None = None) -> None: self._id_generator = DeterministicIdGenerator() self._bind_sdk_tracer() - try: - register_standalone_instrumentations(result) - except Exception: - logger.exception("Failed to register standalone instrumentations") - # per invocation status: self._execution_arn = "" self._execution_trace_id: int | None = None diff --git a/packages/aws-durable-execution-sdk-python-otel/tests/test_instrumentation_ownership.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_instrumentation_ownership.py new file mode 100644 index 00000000..9a11363b --- /dev/null +++ b/packages/aws-durable-execution-sdk-python-otel/tests/test_instrumentation_ownership.py @@ -0,0 +1,48 @@ +"""Tests for application ownership of OpenTelemetry instrumentation.""" + +from __future__ import annotations + +import sys +from types import ModuleType + +import pytest +from opentelemetry import trace +from opentelemetry.sdk.trace import TracerProvider + +from aws_durable_execution_sdk_python_otel.execution_plugin import ExecutionOtelPlugin +from aws_durable_execution_sdk_python_otel.invocation_plugin import InvocationOtelPlugin +from aws_durable_execution_sdk_python_otel.otel_plugin_config import OtelPluginConfig + + +PluginType = type[ExecutionOtelPlugin] | type[InvocationOtelPlugin] + + +@pytest.mark.parametrize( + "plugin_type", + [ExecutionOtelPlugin, InvocationOtelPlugin], +) +def test_plugin_does_not_register_botocore_instrumentation( + plugin_type: PluginType, + monkeypatch: pytest.MonkeyPatch, +) -> None: + provider = TracerProvider() + monkeypatch.setattr(trace, "get_tracer_provider", lambda: provider) + instrument_calls: list[dict[str, object]] = [] + + class BotocoreInstrumentor: + is_instrumented_by_opentelemetry = False + + def instrument(self, **kwargs: object) -> None: + instrument_calls.append(kwargs) + + botocore_module = ModuleType("opentelemetry.instrumentation.botocore") + setattr(botocore_module, "BotocoreInstrumentor", BotocoreInstrumentor) + monkeypatch.setitem( + sys.modules, + "opentelemetry.instrumentation.botocore", + botocore_module, + ) + + plugin_type(OtelPluginConfig(enrich_logger=False)) + + assert instrument_calls == [] diff --git a/pyproject.toml b/pyproject.toml index 9ac77a74..9aca6da7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -49,7 +49,6 @@ extra-dependencies = [ "pytest", "boto3-stubs[lambda]", "opentelemetry-sdk>=1.20.0", - "opentelemetry-instrumentation-botocore", ] [tool.hatch.envs.types.scripts]