diff --git a/.github/scripts/tests/test_opentelemetry_conformance_workflow.py b/.github/scripts/tests/test_opentelemetry_conformance_workflow.py new file mode 100644 index 00000000..d6c3a6f9 --- /dev/null +++ b/.github/scripts/tests/test_opentelemetry_conformance_workflow.py @@ -0,0 +1,27 @@ +from pathlib import Path + + +WORKFLOW_PATH = ( + Path(__file__).parents[2] / "workflows" / "opentelemetry-conformance-tests.yml" +) + + +def test_opentelemetry_conformance_caller_uses_current_workflow_contract() -> None: + workflow = WORKFLOW_PATH.read_text() + + assert "otlp_endpoint:" not in workflow + + for secret_name in ( + "DATADOG_ACCESS_TOKEN", + "DATADOG_API_KEY", + "DATADOG_APPLICATION_KEY", + ): + mapping = f"{secret_name}: ${{{{ secrets.{secret_name} }}}}" + assert mapping in workflow + + for obsolete_secret_name in ( + "DD_API_KEY", + "DD_APPLICATION_KEY", + "DATADOG_OTLP_HEADERS", + ): + assert f"{obsolete_secret_name}:" not in workflow diff --git a/.github/workflows/opentelemetry-conformance-tests.yml b/.github/workflows/opentelemetry-conformance-tests.yml index 73e83247..3e17a226 100644 --- a/.github/workflows/opentelemetry-conformance-tests.yml +++ b/.github/workflows/opentelemetry-conformance-tests.yml @@ -34,10 +34,6 @@ on: required: true default: us-west-2 type: string - otlp_endpoint: - description: OTLP ingest endpoint for community-layer jobs - required: false - type: string conformance_test_ref: description: Conformance test commit SHA or branch name required: true @@ -57,7 +53,6 @@ jobs: phase: ${{ inputs.phase || 'short' }} delay_seconds: ${{ inputs.delay_seconds || '82800' }} aws_region: ${{ inputs.aws_region || 'us-west-2' }} - otlp_endpoint: ${{ inputs.otlp_endpoint || '' }} conformance_test_ref: ${{ inputs.conformance_test_ref || 'main' }} python_sdk_ref: ${{ github.event.pull_request.head.sha || github.sha }} secrets: @@ -65,6 +60,6 @@ jobs: CONFORMANCE_TEST_ACCOUNT_ID: ${{ secrets.TEST_ACCOUNT_ID }} CONFORMANCE_TEST_LAMBDA_EXECUTION_ROLE_ARN: ${{ secrets.TEST_LAMBDA_EXECUTION_ROLE_ARN }} DASH0_AUTH_TOKEN: ${{ secrets.DASH0_AUTH_TOKEN }} - DD_API_KEY: ${{ secrets.DD_API_KEY }} - DD_APPLICATION_KEY: ${{ secrets.DD_APPLICATION_KEY }} - DATADOG_OTLP_HEADERS: ${{ secrets.DATADOG_OTLP_HEADERS }} + DATADOG_ACCESS_TOKEN: ${{ secrets.DATADOG_ACCESS_TOKEN }} + DATADOG_API_KEY: ${{ secrets.DATADOG_API_KEY }} + DATADOG_APPLICATION_KEY: ${{ secrets.DATADOG_APPLICATION_KEY }} diff --git a/.github/workflows/test-parser.yml b/.github/workflows/test-parser.yml index fde37133..abb8101c 100644 --- a/.github/workflows/test-parser.yml +++ b/.github/workflows/test-parser.yml @@ -6,12 +6,14 @@ on: - '.github/scripts/build_lambda_layer.py' - '.github/scripts/parse_sdk_branch.py' - '.github/scripts/tests/**' + - '.github/workflows/opentelemetry-conformance-tests.yml' push: branches: [ main ] paths: - '.github/scripts/build_lambda_layer.py' - '.github/scripts/parse_sdk_branch.py' - '.github/scripts/tests/**' + - '.github/workflows/opentelemetry-conformance-tests.yml' permissions: contents: read @@ -29,4 +31,5 @@ jobs: run: | python -m pytest \ .github/scripts/tests/test_build_lambda_layer.py \ + .github/scripts/tests/test_opentelemetry_conformance_workflow.py \ .github/scripts/tests/test_parse_sdk_branch.py 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 4eb563aa..89a58302 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 @@ -187,10 +189,6 @@ from aws_durable_execution_sdk_python_otel import ( 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 a custom context extractor (default: xray_context_extractor). context_extractor=xray_context_extractor, # Custom instrumentation scope name @@ -275,12 +273,13 @@ InvocationOtelPlugin( 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). ) ) ``` +Pass `tracer_provider=...` when the application owns the OpenTelemetry SDK +provider. When omitted, the globally configured provider is used. + ### `DeterministicIdGenerator` A custom OpenTelemetry `IdGenerator` that produces reproducible trace and span IDs from execution metadata. Exported for advanced use cases. @@ -309,7 +308,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..9df29594 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", ] @@ -33,15 +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] -# 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. -instrumentation = [ - "opentelemetry-instrumentation-botocore", - "opentelemetry-instrumentation-urllib3", -] - [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 b9f1257e..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 @@ -16,11 +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 ( - register_standalone_instrumentations, ) from aws_durable_execution_sdk_python_otel.log_filter import ( OtelContextLogFilter, @@ -41,16 +36,13 @@ "DeterministicIdGenerator", "ExecutionOtelPlugin", "OtelPluginConfig", - "ExporterConfig", "InvocationOtelPlugin", "OtelContextLogFilter", "ProviderResult", - "ProviderSource", "create_tracer_provider", "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/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 0a511a04..564ad905 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 @@ -40,6 +41,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, @@ -55,16 +57,11 @@ ) 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.otel_plugin_config import ( - OtelPluginConfig, - ProviderSource, -) -from aws_durable_execution_sdk_python_otel.instrumentations import ( - register_standalone_instrumentations, -) +from aws_durable_execution_sdk_python_otel.otel_plugin_config import OtelPluginConfig 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 @@ -102,49 +99,43 @@ 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._uses_global_provider = result.uses_global_provider self._tracer: Tracer = self._provider.get_tracer(self._config.instrument_name) - - try: - register_standalone_instrumentations(self._config, result) - except Exception: - logger.exception("Failed to register standalone instrumentations") + self._id_generator = DeterministicIdGenerator() + self._bind_sdk_tracer() # 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._uses_global_provider: + 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 # ------------------------------------------------------------------ @@ -197,18 +188,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 @@ -222,56 +235,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 @@ -328,17 +325,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) @@ -352,6 +353,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 @@ -397,27 +400,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 @@ -426,6 +428,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" @@ -457,6 +461,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 deleted file mode 100644 index 5629ae56..00000000 --- a/packages/aws-durable-execution-sdk-python-otel/src/aws_durable_execution_sdk_python_otel/instrumentations.py +++ /dev/null @@ -1,123 +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 (``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. -""" - -from __future__ import annotations - -import logging -import os -from typing import TYPE_CHECKING, Any - -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.""" - 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_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. - - Args: - config: Shared plugin configuration. - result: The resolved provider and its :class:`ProviderSource`. - """ - if result.source is ProviderSource.EXPLICIT: - # Caller manages their own instrumentation: skip everything. - return - - if result.source is ProviderSource.GLOBAL: - # 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 ec09a170..f0fe0b89 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 @@ -21,14 +21,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, ) @@ -38,17 +37,13 @@ ) 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, -) +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__) @@ -71,25 +66,22 @@ 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 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). Set - ``provider_source=ProviderSource.AUTO_OTLP`` on the config to have - the plugin build and own an auto-configured OTLP provider instead. + 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" @@ -98,17 +90,13 @@ 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). 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 + 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 @@ -121,51 +109,22 @@ 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); 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, - ) + 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._uses_global_provider = result.uses_global_provider self._tracer: Tracer = self._provider.get_tracer(self._config.instrument_name) - - try: - register_standalone_instrumentations(self._config, result) - except Exception: - logger.exception("Failed to register standalone instrumentations") + self._id_generator = DeterministicIdGenerator() + self._bind_sdk_tracer() # 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 @@ -174,6 +133,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._uses_global_provider: + 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: @@ -244,6 +217,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, @@ -264,10 +247,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 @@ -284,28 +266,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 @@ -317,20 +284,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) @@ -362,9 +330,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) @@ -389,22 +370,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: @@ -451,20 +437,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 @@ -486,14 +479,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( @@ -530,6 +525,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( @@ -572,6 +569,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..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 @@ -7,13 +7,11 @@ from __future__ import annotations -from dataclasses import dataclass, field -from enum import Enum -from typing import TYPE_CHECKING, Sequence +from dataclasses import dataclass +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,29 +21,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): - """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()) - 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 @@ -56,52 +31,20 @@ 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()``). ``AUTO_OTLP`` makes the plugin - build and own an OTLP 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``. + 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. - 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. """ - provider_source: ProviderSource = ProviderSource.GLOBAL 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 - - 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`` / ``AUTO_OTLP`` (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 / AUTO_OTLP." - ) 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..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,193 +1,45 @@ -"""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). -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. -""" +"""Shared TracerProvider factory for the durable-execution OTel plugins.""" 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, -) +from aws_durable_execution_sdk_python_otel.otel_plugin_config import OtelPluginConfig 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 +@dataclass(frozen=True) 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) + uses_global_provider: bool -def _build_resource(): - """Build a Lambda resource from AWS_* env vars.""" - from opentelemetry.sdk.resources import Resource +def create_tracer_provider(config: OtelPluginConfig) -> ProviderResult: + """Resolve the configured provider or the global provider. - 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: - """Resolve a TracerProvider from the config's :attr:`provider_source`. - - 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 - 3. ``AUTO_OTLP`` -> a plugin-owned, auto-configured OTLP 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. - id_generator: Deterministic ID generator injected into an - auto-configured provider so cross-invocation trace stitching works. 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() - 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}") + 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_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 8f3247b8..9ef7a765 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 @@ -34,10 +34,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) @@ -66,7 +63,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, @@ -149,7 +145,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()) @@ -171,9 +167,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(): @@ -419,11 +436,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() @@ -431,7 +448,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, ) @@ -451,6 +467,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): @@ -469,6 +487,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 5db9a5c2..46dbed99 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,11 @@ 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; 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 @@ -36,16 +36,18 @@ 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, ) 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) @@ -57,6 +59,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 +161,119 @@ def _run_step_lifecycle(plugin: ExecutionOtelPlugin) -> None: ) +def _config_for_provider( + uses_global_provider: bool, + provider: TracerProvider, + monkeypatch: pytest.MonkeyPatch, +) -> OtelPluginConfig: + if uses_global_provider: + monkeypatch.setattr(trace, "get_tracer_provider", lambda: provider) + return OtelPluginConfig( + tracer_provider=None if uses_global_provider else provider, + context_extractor=lambda _: Context(), + enrich_logger=False, + ) + + +@pytest.mark.parametrize( + "uses_global_provider", + [False, True], + ids=["explicit", "global"], +) +def test_unrelated_root_spans_keep_provider_id_generation( + uses_global_provider: bool, 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_provider(uses_global_provider, 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( + 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( + 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) # --------------------------------------------------------------------------- @@ -165,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, @@ -191,9 +306,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) @@ -218,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, ) @@ -237,12 +351,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,12 +373,11 @@ 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( - provider_source=ProviderSource.GLOBAL, context_extractor=lambda _: Context(), enrich_logger=False, ) @@ -269,11 +385,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_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/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 3e409a4a..3e0efd8a 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 @@ -37,10 +37,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) @@ -69,7 +66,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(), ) @@ -199,9 +195,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() @@ -358,8 +378,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" @@ -386,9 +406,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 ) @@ -423,8 +443,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" @@ -463,9 +483,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(): @@ -1121,7 +1141,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 0662926c..0e4bb0d5 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,16 +39,19 @@ 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 -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) @@ -60,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) @@ -161,8 +165,122 @@ def _run_step_lifecycle(plugin: InvocationOtelPlugin) -> None: ) +def _config_for_provider( + uses_global_provider: bool, + provider: TracerProvider, + monkeypatch: pytest.MonkeyPatch, +) -> OtelPluginConfig: + if uses_global_provider: + monkeypatch.setattr(trace, "get_tracer_provider", lambda: provider) + return OtelPluginConfig( + tracer_provider=None if uses_global_provider else provider, + context_extractor=lambda _: Context(), + enrich_logger=False, + ) + + +@pytest.mark.parametrize( + "uses_global_provider", + [False, True], + ids=["explicit", "global"], +) +def test_unrelated_root_spans_keep_provider_id_generation( + uses_global_provider: bool, 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_provider(uses_global_provider, 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( + 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( + 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 +288,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 +305,46 @@ def _assert_hierarchy(exporter: InMemorySpanExporter) -> None: assert attempt.parent.span_id == operation.context.span_id +@pytest.mark.parametrize( + "uses_global_provider", + [False, True], + ids=["explicit", "global"], +) +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_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)) + 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) # --------------------------------------------------------------------------- @@ -193,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, @@ -217,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, ) @@ -231,12 +388,11 @@ 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( - provider_source=ProviderSource.EXPLICIT, tracer_provider=provider, context_extractor=lambda _: Context(), enrich_logger=False, @@ -244,7 +400,6 @@ def test_second_plugin_configures_cached_tracer_generator(monkeypatch): ) target_plugin = InvocationOtelPlugin( OtelPluginConfig( - provider_source=ProviderSource.EXPLICIT, tracer_provider=provider, context_extractor=lambda _: Context(), enrich_logger=False, @@ -252,11 +407,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_log_filter.py b/packages/aws-durable-execution-sdk-python-otel/tests/test_log_filter.py index 36236c68..c7daa077 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 e792c2ff..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 @@ -5,130 +5,31 @@ 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.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 -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) -# --------------------------------------------------------------------------- -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_global_source_rejects_tracer_provider(): - with pytest.raises(ValueError, match="only valid with provider_source=EXPLICIT"): - 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_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_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" + assert result.tracer_provider is provider + assert result.uses_global_provider is False diff --git a/pyproject.toml b/pyproject.toml index 6ea8e7c1..9aca6da7 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -49,8 +49,6 @@ extra-dependencies = [ "pytest", "boto3-stubs[lambda]", "opentelemetry-sdk>=1.20.0", - "opentelemetry-instrumentation-botocore", - "opentelemetry-instrumentation-urllib3", ] [tool.hatch.envs.types.scripts]