diff --git a/.github/workflows/durabletask-azurefunctions.yml b/.github/workflows/durabletask-azurefunctions.yml index a3260325..d5558bb0 100644 --- a/.github/workflows/durabletask-azurefunctions.yml +++ b/.github/workflows/durabletask-azurefunctions.yml @@ -80,7 +80,7 @@ jobs: - name: Set up Node.js (needed for Azurite and Functions Core Tools) uses: actions/setup-node@v4 with: - node-version: "20.x" + node-version: "22.x" - name: Install Azurite and Azure Functions Core Tools run: | diff --git a/.github/workflows/durabletask.yml b/.github/workflows/durabletask.yml index e249a9d2..d4749a79 100644 --- a/.github/workflows/durabletask.yml +++ b/.github/workflows/durabletask.yml @@ -48,7 +48,7 @@ jobs: - name: Set up Node.js (needed for Azurite) uses: actions/setup-node@v4 with: - node-version: '20.x' + node-version: '22.x' - name: Cache npm uses: actions/cache@v3 @@ -57,7 +57,7 @@ jobs: key: ${{ runner.os }}-npm-azurite - name: Install Azurite - run: npm install -g azurite + run: npm install -g azurite@latest - name: Install Nox run: | diff --git a/CHANGELOG.md b/CHANGELOG.md index 1a6fa8d9..828e66d8 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,6 +9,10 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ADDED +- Added `emit_trace_spans=False` to `TaskHubGrpcWorker`, +`TaskHubGrpcClient`, and `AsyncTaskHubGrpcClient` for hosts that own Durable +Task lifecycle spans. In this propagate-only mode, W3C trace context remains +active and is propagated without emitting duplicate SDK lifecycle spans. - Added asynchronous `AsyncScheduledTaskClient` / `AsyncScheduleClient` and `AsyncExportHistoryClient` / `AsyncExportHistoryJobClient` APIs. Applications using `AsyncTaskHubGrpcClient` can now manage scheduled tasks and history diff --git a/azure-functions-durable/CHANGELOG.md b/azure-functions-durable/CHANGELOG.md index cb753941..84c6d94c 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -9,6 +9,10 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0 ADDED +- Distributed tracing now correlates OpenTelemetry spans created by orchestrator +user code with the Durable Functions host trace while avoiding duplicate +orchestration, activity, entity, client-start, and client-event lifecycle spans +from the Python SDK. - Added `SyncDurableFunctionsClient`. `DFApp.durable_client_input()` now injects the synchronous client into synchronous functions and the asynchronous client into coroutine functions. Both clients support scheduled-task and history-export @@ -197,8 +201,3 @@ code: `show_history_output` flags for signature compatibility but ignore them, so the returned status has no `historyEvents`. Use `get_orchestration_history(...)` to retrieve history. -- Distributed tracing is not yet wired up. The Durable Functions host delivers - the parent trace context and emits the orchestration/activity spans itself, - so orchestrator user-code spans in the Python worker are not yet correlated - to it, and durabletask's own span emission is intentionally left disabled to - avoid duplicating the host's spans. diff --git a/azure-functions-durable/azure/durable_functions/client.py b/azure-functions-durable/azure/durable_functions/client.py index 537c0ed8..aca0b2ec 100644 --- a/azure-functions-durable/azure/durable_functions/client.py +++ b/azure-functions-durable/azure/durable_functions/client.py @@ -90,7 +90,8 @@ def __init__(self, client_as_string: str): metadata=None, interceptors=interceptors, channel_options=channel_options, - data_converter=DEFAULT_FUNCTIONS_DATA_CONVERTER) + data_converter=DEFAULT_FUNCTIONS_DATA_CONVERTER, + emit_trace_spans=False) # The gRPC aio channel is bound to the event loop it is created on. A # ``durable_client_input`` decode runs on the worker's invocation loop, @@ -531,7 +532,8 @@ def __init__(self, client_as_string: str): metadata=None, interceptors=interceptors, channel_options=channel_options, - data_converter=DEFAULT_FUNCTIONS_DATA_CONVERTER) + data_converter=DEFAULT_FUNCTIONS_DATA_CONVERTER, + emit_trace_spans=False) @classmethod def get_cached(cls, client_as_string: str) -> "SyncDurableFunctionsClient": diff --git a/azure-functions-durable/azure/durable_functions/worker.py b/azure-functions-durable/azure/durable_functions/worker.py index a27a0eaf..2d46a465 100644 --- a/azure-functions-durable/azure/durable_functions/worker.py +++ b/azure-functions-durable/azure/durable_functions/worker.py @@ -41,7 +41,10 @@ def __init__(self) -> None: # The Functions converter routes payload serialization through the # azure-functions codec (df_dumps/df_loads) so user types round-trip in # the wire format the Durable Functions host extension expects. - super().__init__(data_converter=DEFAULT_FUNCTIONS_DATA_CONVERTER) + super().__init__( + data_converter=DEFAULT_FUNCTIONS_DATA_CONVERTER, + emit_trace_spans=False, + ) self._registration_lock = Lock() self._registered_orchestrator_functions: dict[ str, task.Orchestrator[Any, Any] diff --git a/azure-functions-durable/pyproject.toml b/azure-functions-durable/pyproject.toml index 1099aebb..fb675b4c 100644 --- a/azure-functions-durable/pyproject.toml +++ b/azure-functions-durable/pyproject.toml @@ -27,7 +27,7 @@ requires-python = ">=3.13" license = {file = "LICENSE"} readme = "README.md" dependencies = [ - "durabletask>=1.8.0", + "durabletask[opentelemetry]>=1.9.0", "azure-identity>=1.19.0", "azure-functions>=2.3.0b2" ] diff --git a/durabletask/client.py b/durabletask/client.py index a4851b8d..a5974749 100644 --- a/durabletask/client.py +++ b/durabletask/client.py @@ -419,7 +419,8 @@ def __init__(self, *, resiliency_options: GrpcClientResiliencyOptions | None = None, default_version: str | None = None, payload_store: PayloadStore | None = None, - data_converter: DataConverter | None = None): + data_converter: DataConverter | None = None, + emit_trace_spans: bool = True): self._owns_channel = channel is None self._data_converter = data_converter if data_converter is not None else JsonDataConverter() @@ -484,6 +485,12 @@ def __init__(self, *, self._logger = shared.get_logger("client", log_handler, log_formatter) self.default_version = default_version self._payload_store = payload_store + self._emit_trace_spans = emit_trace_spans + + @property + def emit_trace_spans(self) -> bool: + """Return whether this client emits Durable Task lifecycle spans.""" + return self._emit_trace_spans def _compose_interceptors( self, @@ -619,30 +626,30 @@ def schedule_new_orchestration(self, orchestrator: task.Orchestrator[TInput, TOu resolved_instance_id = instance_id if instance_id else uuid.uuid4().hex resolved_version = version if version else self.default_version - with tracing.start_create_orchestration_span( - name, resolved_instance_id, version=resolved_version, - ): - req = build_schedule_new_orchestration_req( - orchestrator, input=input, instance_id=instance_id, start_at=start_at, - reuse_id_policy=reuse_id_policy, tags=tags, - version=version if version else self.default_version, - data_converter=self._data_converter) - - # Inject the active PRODUCER span context into the request so the sidecar - # stores it in the executionStarted event and the worker can parent all - # orchestration/activity/timer spans under this trace. - parent_trace_ctx = tracing.get_current_trace_context() - if parent_trace_ctx is not None: - req.parentTraceContext.CopyFrom(parent_trace_ctx) - - self._logger.info(f"Starting new '{req.name}' instance with ID = '{req.instanceId}'.") - # Externalize any large payloads in the request - if self._payload_store is not None: - payload_helpers.externalize_payloads( - req, self._payload_store, instance_id=req.instanceId, - ) - res: pb.CreateInstanceResponse = self._stub.StartInstance(req) - return res.instanceId + with tracing.suppress_span_emission(not self._emit_trace_spans): + with tracing.start_create_orchestration_span( + name, resolved_instance_id, version=resolved_version, + ): + req = build_schedule_new_orchestration_req( + orchestrator, input=input, instance_id=instance_id, start_at=start_at, + reuse_id_policy=reuse_id_policy, tags=tags, + version=version if version else self.default_version, + data_converter=self._data_converter) + + # Inject the active PRODUCER span context, or the ambient context + # when lifecycle span emission is disabled. + parent_trace_ctx = tracing.get_current_trace_context() + if parent_trace_ctx is not None: + req.parentTraceContext.CopyFrom(parent_trace_ctx) + + self._logger.info(f"Starting new '{req.name}' instance with ID = '{req.instanceId}'.") + # Externalize any large payloads in the request + if self._payload_store is not None: + payload_helpers.externalize_payloads( + req, self._payload_store, instance_id=req.instanceId, + ) + res: pb.CreateInstanceResponse = self._stub.StartInstance(req) + return res.instanceId def get_orchestration_state(self, instance_id: str, *, fetch_payloads: bool = True) -> OrchestrationState | None: req = pb.GetInstanceRequest(instanceId=instance_id, getInputsAndOutputs=fetch_payloads) @@ -757,14 +764,15 @@ def wait_for_orchestration_completion(self, instance_id: str, *, def raise_orchestration_event(self, instance_id: str, event_name: str, *, data: Any | None = None) -> None: - with tracing.start_raise_event_span(event_name, instance_id): - req = build_raise_event_req(instance_id, event_name, data, self._data_converter) - self._logger.info(f"Raising event '{event_name}' for instance '{instance_id}'.") - if self._payload_store is not None: - payload_helpers.externalize_payloads( - req, self._payload_store, instance_id=instance_id, - ) - self._stub.RaiseEvent(req) + with tracing.suppress_span_emission(not self._emit_trace_spans): + with tracing.start_raise_event_span(event_name, instance_id): + req = build_raise_event_req(instance_id, event_name, data, self._data_converter) + self._logger.info(f"Raising event '{event_name}' for instance '{instance_id}'.") + if self._payload_store is not None: + payload_helpers.externalize_payloads( + req, self._payload_store, instance_id=instance_id, + ) + self._stub.RaiseEvent(req) def terminate_orchestration(self, instance_id: str, *, output: Any | None = None, @@ -940,7 +948,8 @@ def __init__(self, *, resiliency_options: GrpcClientResiliencyOptions | None = None, default_version: str | None = None, payload_store: PayloadStore | None = None, - data_converter: DataConverter | None = None): + data_converter: DataConverter | None = None, + emit_trace_spans: bool = True): self._owns_channel = channel is None self._data_converter = data_converter if data_converter is not None else JsonDataConverter() @@ -1002,6 +1011,12 @@ def __init__(self, *, self._logger = shared.get_logger("async_client", log_handler, log_formatter) self.default_version = default_version self._payload_store = payload_store + self._emit_trace_spans = emit_trace_spans + + @property + def emit_trace_spans(self) -> bool: + """Return whether this client emits Durable Task lifecycle spans.""" + return self._emit_trace_spans def _compose_interceptors( self, @@ -1148,27 +1163,28 @@ async def schedule_new_orchestration(self, orchestrator: task.Orchestrator[TInpu resolved_instance_id = instance_id if instance_id else uuid.uuid4().hex resolved_version = version if version else self.default_version - with tracing.start_create_orchestration_span( - name, resolved_instance_id, version=resolved_version, - ): - req = build_schedule_new_orchestration_req( - orchestrator, input=input, instance_id=instance_id, start_at=start_at, - reuse_id_policy=reuse_id_policy, tags=tags, - version=version if version else self.default_version, - data_converter=self._data_converter) - - parent_trace_ctx = tracing.get_current_trace_context() - if parent_trace_ctx is not None: - req.parentTraceContext.CopyFrom(parent_trace_ctx) - - self._logger.info(f"Starting new '{req.name}' instance with ID = '{req.instanceId}'.") - # Externalize any large payloads in the request - if self._payload_store is not None: - await payload_helpers.externalize_payloads_async( - req, self._payload_store, instance_id=req.instanceId, - ) - res: pb.CreateInstanceResponse = await self._get_stub().StartInstance(req) - return res.instanceId + with tracing.suppress_span_emission(not self._emit_trace_spans): + with tracing.start_create_orchestration_span( + name, resolved_instance_id, version=resolved_version, + ): + req = build_schedule_new_orchestration_req( + orchestrator, input=input, instance_id=instance_id, start_at=start_at, + reuse_id_policy=reuse_id_policy, tags=tags, + version=version if version else self.default_version, + data_converter=self._data_converter) + + parent_trace_ctx = tracing.get_current_trace_context() + if parent_trace_ctx is not None: + req.parentTraceContext.CopyFrom(parent_trace_ctx) + + self._logger.info(f"Starting new '{req.name}' instance with ID = '{req.instanceId}'.") + # Externalize any large payloads in the request + if self._payload_store is not None: + await payload_helpers.externalize_payloads_async( + req, self._payload_store, instance_id=req.instanceId, + ) + res: pb.CreateInstanceResponse = await self._get_stub().StartInstance(req) + return res.instanceId async def get_orchestration_state(self, instance_id: str, *, fetch_payloads: bool = True) -> OrchestrationState | None: @@ -1282,14 +1298,15 @@ async def wait_for_orchestration_completion(self, instance_id: str, *, async def raise_orchestration_event(self, instance_id: str, event_name: str, *, data: Any | None = None) -> None: - with tracing.start_raise_event_span(event_name, instance_id): - req = build_raise_event_req(instance_id, event_name, data, self._data_converter) - self._logger.info(f"Raising event '{event_name}' for instance '{instance_id}'.") - if self._payload_store is not None: - await payload_helpers.externalize_payloads_async( - req, self._payload_store, instance_id=instance_id, - ) - await self._get_stub().RaiseEvent(req) + with tracing.suppress_span_emission(not self._emit_trace_spans): + with tracing.start_raise_event_span(event_name, instance_id): + req = build_raise_event_req(instance_id, event_name, data, self._data_converter) + self._logger.info(f"Raising event '{event_name}' for instance '{instance_id}'.") + if self._payload_store is not None: + await payload_helpers.externalize_payloads_async( + req, self._payload_store, instance_id=instance_id, + ) + await self._get_stub().RaiseEvent(req) async def terminate_orchestration(self, instance_id: str, *, output: Any | None = None, diff --git a/durabletask/internal/tracing.py b/durabletask/internal/tracing.py index 98ae1e94..8420dc78 100644 --- a/durabletask/internal/tracing.py +++ b/durabletask/internal/tracing.py @@ -18,6 +18,7 @@ import random import time from contextlib import contextmanager +from contextvars import ContextVar from datetime import datetime from typing import TYPE_CHECKING, Any @@ -85,6 +86,10 @@ class _StatusCodeStub: # The instrumentation scope name used when creating spans. _TRACER_NAME = "durabletask" +_SPAN_EMISSION_SUPPRESSED: ContextVar[bool] = ContextVar( + "durabletask_span_emission_suppressed", + default=False, +) # --------------------------------------------------------------------------- @@ -226,6 +231,35 @@ def extract_trace_context(proto_ctx: pb.TraceContext | None) -> Any | None: return ctx +@contextmanager +def use_trace_context(trace_context: pb.TraceContext | None): + """Make a protobuf trace context current without creating a span.""" + parent_ctx = extract_trace_context(trace_context) + if parent_ctx is None: + yield + return + + token = otel_context.attach(parent_ctx) + try: + yield + finally: + otel_context.detach(token) + + +@contextmanager +def suppress_span_emission(suppress: bool = True): + """Temporarily suppress spans emitted by this module.""" + if not suppress: + yield + return + + token = _SPAN_EMISSION_SUPPRESSED.set(True) + try: + yield + finally: + _SPAN_EMISSION_SUPPRESSED.reset(token) + + @contextmanager def start_span( name: str, @@ -254,6 +288,11 @@ def start_span( yield None return + if _SPAN_EMISSION_SUPPRESSED.get(): + with use_trace_context(trace_context): + yield None + return + parent_ctx = extract_trace_context(trace_context) if kind is None: @@ -313,7 +352,7 @@ def emit_orchestration_span( Falls back to ``tracer.start_span()`` (which generates its own span ID) when *orchestration_trace_context* is ``None``. """ - if not _OTEL_AVAILABLE: + if not _OTEL_AVAILABLE or _SPAN_EMISSION_SUPPRESSED.get(): return span_name = create_span_name(TASK_TYPE_ORCHESTRATION, name, version) @@ -529,7 +568,7 @@ def generate_client_trace_context( returned the caller should fall back to the orchestration's own trace context as the parent for downstream spans. """ - if not _OTEL_AVAILABLE: + if not _OTEL_AVAILABLE or _SPAN_EMISSION_SUPPRESSED.get(): return None if parent_trace_context is None: return None @@ -588,7 +627,7 @@ def emit_client_span( SERVER span remains connected to the orchestration span via the fallback parenting established in :func:`generate_client_trace_context`. """ - if not _OTEL_AVAILABLE: + if not _OTEL_AVAILABLE or _SPAN_EMISSION_SUPPRESSED.get(): return # SDK-internal imports — see _is_deferred_span_capable() for details. @@ -698,7 +737,7 @@ def emit_timer_span( When *parent_trace_context* is provided the span is created as a child of that context; otherwise it inherits the ambient context. """ - if not _OTEL_AVAILABLE: + if not _OTEL_AVAILABLE or _SPAN_EMISSION_SUPPRESSED.get(): return span_name = create_timer_span_name(orchestration_name) @@ -741,7 +780,7 @@ def emit_event_raised_span( When *parent_trace_context* is provided the span is created as a child of that context; otherwise it inherits the ambient context. """ - if not _OTEL_AVAILABLE: + if not _OTEL_AVAILABLE or _SPAN_EMISSION_SUPPRESSED.get(): return span_name = create_span_name(SPAN_TYPE_ORCHESTRATION_EVENT, event_name) @@ -787,7 +826,7 @@ def start_create_orchestration_span( Yields the span; caller should capture the trace context after entering the span context so it can be injected into the gRPC request. """ - if not _OTEL_AVAILABLE: + if not _OTEL_AVAILABLE or _SPAN_EMISSION_SUPPRESSED.get(): yield None return @@ -815,7 +854,7 @@ def start_raise_event_span( target_instance_id: str, ): """Context manager for a Producer span when raising an event from the client.""" - if not _OTEL_AVAILABLE: + if not _OTEL_AVAILABLE or _SPAN_EMISSION_SUPPRESSED.get(): yield None return @@ -845,7 +884,7 @@ def reconstruct_trace_context( the span ID with *span_id*. This is used to reuse a pre-determined orchestration span ID across replays. """ - if not _OTEL_AVAILABLE: + if not _OTEL_AVAILABLE or _SPAN_EMISSION_SUPPRESSED.get(): return None parsed = _parse_traceparent(parent_trace_context.traceParent) @@ -869,7 +908,7 @@ def build_orchestration_trace_context( replays so that all dispatches produce a consistent orchestration SERVER span. """ - if start_time_ns is None: + if start_time_ns is None or _SPAN_EMISSION_SUPPRESSED.get(): return None ctx = pb.OrchestrationTraceContext() diff --git a/durabletask/worker.py b/durabletask/worker.py index 46275362..77d7f44f 100644 --- a/durabletask/worker.py +++ b/durabletask/worker.py @@ -463,6 +463,10 @@ class TaskHubGrpcWorker: gRPC resiliency settings retained for reconnect handling. concurrency_options (ConcurrencyOptions | None, optional): Configuration for controlling worker concurrency limits. If None, default settings are used. + emit_trace_spans (bool, optional): Whether the worker emits Durable Task + lifecycle spans. Set to ``False`` when the host owns those spans; the + worker will still make inbound trace context current during user-code + execution. Defaults to ``True``. Attributes: concurrency_options (ConcurrencyOptions): The current concurrency configuration. @@ -533,6 +537,7 @@ def __init__( maximum_timer_interval: timedelta | None = DEFAULT_MAXIMUM_TIMER_INTERVAL, payload_store: PayloadStore | None = None, data_converter: DataConverter | None = None, + emit_trace_spans: bool = True, ): self._registry = _Registry() # Shared by every entity batch this worker processes so that reflected @@ -552,6 +557,7 @@ def __init__( self._owns_channel = channel is None self._secure_channel = secure_channel self._payload_store = payload_store + self._emit_trace_spans = emit_trace_spans self._channel_options = channel_options self._resiliency_options = ( resiliency_options @@ -597,6 +603,11 @@ def maximum_timer_interval(self) -> timedelta | None: """Get the configured maximum timer interval for long timer chunking.""" return self._maximum_timer_interval + @property + def emit_trace_spans(self) -> bool: + """Return whether this worker emits Durable Task lifecycle spans.""" + return self._emit_trace_spans + def __enter__(self) -> "TaskHubGrpcWorker": return self @@ -1166,7 +1177,11 @@ def _execute_orchestrator( persisted_orch_span_id=persisted_orch_span_id, maximum_timer_interval=self.maximum_timer_interval, data_converter=self._data_converter) - result = executor.execute(instance_id, req.pastEvents, req.newEvents) + with tracing.suppress_span_emission(not self._emit_trace_spans): + with tracing.use_trace_context( + parent_trace_ctx if not self._emit_trace_spans else None): + result = executor.execute( + instance_id, req.pastEvents, req.newEvents) # Determine completion status for span is_complete = False @@ -1180,7 +1195,7 @@ def _execute_orchestrator( is_failed = True failure_details = action.completeOrchestration.failureDetails - if is_complete: + if is_complete and self._emit_trace_spans: # Orchestration finished — emit a single span covering its lifetime tracing.emit_orchestration_span( orchestration_name, @@ -1197,8 +1212,12 @@ def _execute_orchestrator( orch_span_id = None if result._orchestration_trace_context: # pyright: ignore[reportPrivateUsage] orch_span_id = result._orchestration_trace_context.spanID # pyright: ignore[reportPrivateUsage] - orch_trace_ctx = tracing.build_orchestration_trace_context( - start_time_ns, span_id=orch_span_id) + orch_trace_ctx = ( + tracing.build_orchestration_trace_context( + start_time_ns, span_id=orch_span_id) + if self._emit_trace_spans + else None + ) res = pb.OrchestratorResponse( instanceId=instance_id, @@ -1223,14 +1242,15 @@ def _execute_orchestrator( return except Exception as ex: # Unhandled error — emit a failed span - tracing.emit_orchestration_span( - orchestration_name, - instance_id, - start_time_ns, - is_failed=True, - failure_details=ex, - parent_trace_context=parent_trace_ctx, - ) + if self._emit_trace_spans: + tracing.emit_orchestration_span( + orchestration_name, + instance_id, + start_time_ns, + is_failed=True, + failure_details=ex, + parent_trace_context=parent_trace_ctx, + ) self._logger.exception( f"An error occurred while trying to execute instance '{instance_id}': {ex}" ) @@ -1291,24 +1311,25 @@ def _execute_activity( payload_helpers.deexternalize_payloads(req, self._payload_store) try: executor = _ActivityExecutor(self._registry, self._logger, self._data_converter) - with tracing.start_span( - tracing.create_span_name("activity", req.name), - trace_context=req.parentTraceContext, - kind=tracing.SpanKind.SERVER, - attributes={ - tracing.ATTR_TASK_TYPE: "activity", - tracing.ATTR_TASK_INSTANCE_ID: instance_id, - tracing.ATTR_TASK_NAME: req.name, - tracing.ATTR_TASK_TASK_ID: str(req.taskId), - }, - ) as span: - try: - result = executor.execute( - instance_id, req.name, req.taskId, req.input.value - ) - except Exception as ex: - tracing.set_span_error(span, ex) - raise + with tracing.suppress_span_emission(not self._emit_trace_spans): + with tracing.start_span( + tracing.create_span_name("activity", req.name), + trace_context=req.parentTraceContext, + kind=tracing.SpanKind.SERVER, + attributes={ + tracing.ATTR_TASK_TYPE: "activity", + tracing.ATTR_TASK_INSTANCE_ID: instance_id, + tracing.ATTR_TASK_NAME: req.name, + tracing.ATTR_TASK_TASK_ID: str(req.taskId), + }, + ) as span: + try: + result = executor.execute( + instance_id, req.name, req.taskId, req.input.value + ) + except Exception as ex: + tracing.set_span_error(span, ex) + raise res = pb.ActivityResponse( instanceId=instance_id, taskId=req.taskId, @@ -1393,42 +1414,43 @@ def _execute_entity_batch( # Get the trace context for this operation, if available op_trace_ctx = operation.traceContext if operation.HasField("traceContext") else None - with tracing.start_span( - tracing.create_span_name("entity", f"{entity_instance_id.entity}:{operation.operation}"), - trace_context=op_trace_ctx, - kind=tracing.SpanKind.SERVER, - attributes={ - tracing.ATTR_TASK_TYPE: "entity", - tracing.ATTR_TASK_INSTANCE_ID: instance_id, - tracing.ATTR_TASK_NAME: entity_instance_id.entity, - "durabletask.entity.operation": operation.operation, - }, - ) as span: - try: - entity_result = executor.execute( - instance_id, entity_instance_id, operation.operation, entity_state, operation.input.value - ) + with tracing.suppress_span_emission(not self._emit_trace_spans): + with tracing.start_span( + tracing.create_span_name("entity", f"{entity_instance_id.entity}:{operation.operation}"), + trace_context=op_trace_ctx, + kind=tracing.SpanKind.SERVER, + attributes={ + tracing.ATTR_TASK_TYPE: "entity", + tracing.ATTR_TASK_INSTANCE_ID: instance_id, + tracing.ATTR_TASK_NAME: entity_instance_id.entity, + "durabletask.entity.operation": operation.operation, + }, + ) as span: + try: + entity_result = executor.execute( + instance_id, entity_instance_id, operation.operation, entity_state, operation.input.value + ) - entity_result = ph.get_string_value_or_empty(entity_result) - operation_result = pb.OperationResult(success=pb.OperationResultSuccess( - result=entity_result, - startTimeUtc=new_timestamp(start_time), - endTimeUtc=new_timestamp(datetime.now(timezone.utc)) - )) - results.append(operation_result) + entity_result = ph.get_string_value_or_empty(entity_result) + operation_result = pb.OperationResult(success=pb.OperationResultSuccess( + result=entity_result, + startTimeUtc=new_timestamp(start_time), + endTimeUtc=new_timestamp(datetime.now(timezone.utc)) + )) + results.append(operation_result) - entity_state.commit() - except Exception as ex: - tracing.set_span_error(span, ex) - self._logger.exception(ex) - operation_result = pb.OperationResult(failure=pb.OperationResultFailure( - failureDetails=ph.new_failure_details(ex), - startTimeUtc=new_timestamp(start_time), - endTimeUtc=new_timestamp(datetime.now(timezone.utc)) - )) - results.append(operation_result) - - entity_state.rollback() + entity_state.commit() + except Exception as ex: + tracing.set_span_error(span, ex) + self._logger.exception(ex) + operation_result = pb.OperationResult(failure=pb.OperationResultFailure( + failureDetails=ph.new_failure_details(ex), + startTimeUtc=new_timestamp(start_time), + endTimeUtc=new_timestamp(datetime.now(timezone.utc)) + )) + results.append(operation_result) + + entity_state.rollback() batch_result = pb.EntityBatchResult( results=results, diff --git a/noxfile.py b/noxfile.py index dba01efe..075c71cc 100644 --- a/noxfile.py +++ b/noxfile.py @@ -42,7 +42,7 @@ AZURE_FUNCTIONS_DURABLE = REPO_ROOT / "azure-functions-durable" E2E_APPS_DIR = REPO_ROOT / "tests" / "azure-functions-durable" / "e2e" / "apps" # Sample apps that need an in-app virtual environment for the E2E suite. -E2E_APPS = ("v1_style", "dtask_style") +E2E_APPS = ("v1_style", "dtask_style", "tracing") PYTHON_VERSIONS = ("3.10", "3.11", "3.12", "3.13", "3.14") DEFAULT_CI_PYTHON = "3.10" @@ -422,7 +422,7 @@ def functions_e2e(session: nox.Session) -> None: ] = _new_test_namespace("nox") session.install("-r", "requirements.txt") _install_packages(session, editable=True) - session.install("pytest") + session.install("pytest", "opentelemetry-exporter-otlp-proto-grpc") for app in E2E_APPS: _link_app_venv(session, E2E_APPS_DIR / app) arguments = _pytest_arguments( diff --git a/pyproject.toml b/pyproject.toml index ac2d1f1a..f3c474b4 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -9,7 +9,7 @@ build-backend = "setuptools.build_meta" [project] name = "durabletask" -version = "1.8.0" +version = "1.9.0" description = "A Durable Task Client SDK for Python" keywords = [ "durable", diff --git a/tests/azure-functions-durable/e2e/_harness.py b/tests/azure-functions-durable/e2e/_harness.py index 72db2cd3..71264239 100644 --- a/tests/azure-functions-durable/e2e/_harness.py +++ b/tests/azure-functions-durable/e2e/_harness.py @@ -22,7 +22,9 @@ import subprocess import time import urllib.error +import urllib.parse import urllib.request +import uuid from dataclasses import dataclass from pathlib import Path from typing import Any, Callable, Optional @@ -49,6 +51,17 @@ def find_free_port() -> int: return s.getsockname()[1] +def _wait_for_port(port: int, timeout: float) -> bool: + deadline = time.time() + timeout + while time.time() < deadline: + try: + with socket.create_connection(("127.0.0.1", port), timeout=1): + return True + except OSError: + time.sleep(0.25) + return False + + def func_executable() -> Optional[str]: """Return the path to the Azure Functions Core Tools (``func``), if installed.""" return shutil.which("func") @@ -63,6 +76,144 @@ def azurite_is_running() -> bool: return False +class OtelCollector: + """Run an OpenTelemetry Collector that writes received spans to JSON.""" + + _IMAGE = "otel/opentelemetry-collector-contrib:0.131.1" + + def __init__(self, work_dir: Path): + self.work_dir = work_dir + self.config_path = work_dir / "collector.yaml" + self.output_path = work_dir / "traces.json" + self.port = find_free_port() + self.endpoint = f"http://127.0.0.1:{self.port}" + self.container_name = f"durabletask-tracing-{uuid.uuid4().hex[:12]}" + + @staticmethod + def is_available() -> bool: + docker = shutil.which("docker") + if docker is None: + return False + result = subprocess.run( + [docker, "info"], + capture_output=True, + check=False, + text=True, + ) + return result.returncode == 0 + + def __enter__(self) -> "OtelCollector": + self.start() + return self + + def __exit__(self, *exc: object) -> None: + self.stop() + + def start(self) -> None: + docker = shutil.which("docker") + if docker is None: + raise RuntimeError("Docker is required for the tracing E2E test.") + + self.work_dir.mkdir(parents=True, exist_ok=True) + if os.name != "nt": + # The collector image runs as UID/GID 10001, while pytest creates + # temporary directories as owner-only on Linux. + self.work_dir.chmod(0o777) + self.config_path.write_text( + "receivers:\n" + " otlp:\n" + " protocols:\n" + " grpc:\n" + " endpoint: 0.0.0.0:4317\n" + "exporters:\n" + " file:\n" + " path: /output/traces.json\n" + " format: json\n" + "service:\n" + " pipelines:\n" + " traces:\n" + " receivers: [otlp]\n" + " exporters: [file]\n", + encoding="utf-8", + ) + + command = [ + docker, + "run", + "--detach", + "--rm", + "--name", + self.container_name, + "--publish", + f"127.0.0.1:{self.port}:4317", + "--volume", + f"{self.config_path.resolve()}:/etc/otelcol-contrib/config.yaml:ro", + "--volume", + f"{self.work_dir.resolve()}:/output", + self._IMAGE, + ] + result = subprocess.run( + command, + capture_output=True, + check=False, + text=True, + ) + if result.returncode != 0: + raise RuntimeError( + f"Failed to start OpenTelemetry Collector: {result.stderr}") + if not _wait_for_port(self.port, timeout=60): + logs = subprocess.run( + [docker, "logs", self.container_name], + capture_output=True, + check=False, + text=True, + ) + self.stop() + raise RuntimeError( + f"OpenTelemetry Collector did not start:\n{logs.stdout}\n{logs.stderr}") + + def stop(self) -> None: + docker = shutil.which("docker") + if docker is None: + return + subprocess.run( + [docker, "stop", "--time", "10", self.container_name], + capture_output=True, + check=False, + text=True, + ) + + def get_spans(self) -> list[dict[str, Any]]: + """Read and flatten all OTLP spans exported by the collector.""" + if not self.output_path.exists(): + return [] + + spans: list[dict[str, Any]] = [] + for line in self.output_path.read_text(encoding="utf-8").splitlines(): + if not line.strip(): + continue + payload = json.loads(line) + for resource_spans in payload.get("resourceSpans", []): + for scope_spans in resource_spans.get("scopeSpans", []): + scope_name = scope_spans.get("scope", {}).get("name", "") + for span in scope_spans.get("spans", []): + spans.append({**span, "scopeName": scope_name}) + return spans + + def get_logs(self) -> str: + """Return collector container logs for assertion diagnostics.""" + docker = shutil.which("docker") + if docker is None: + return "Docker is not available." + result = subprocess.run( + [docker, "logs", self.container_name], + capture_output=True, + check=False, + text=True, + ) + return result.stdout + result.stderr + + @dataclass class HttpResult: status: int diff --git a/tests/azure-functions-durable/e2e/apps/tracing/function_app.py b/tests/azure-functions-durable/e2e/apps/tracing/function_app.py new file mode 100644 index 00000000..275d0206 --- /dev/null +++ b/tests/azure-functions-durable/e2e/apps/tracing/function_app.py @@ -0,0 +1,72 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""Minimal Durable Functions app for distributed tracing validation.""" + +import json +from typing import Any + +import azure.functions as func +import azure.durable_functions as df +from durabletask import task +from opentelemetry import trace +from opentelemetry.exporter.otlp.proto.grpc.trace_exporter import OTLPSpanExporter +from opentelemetry.sdk.resources import Resource +from opentelemetry.sdk.trace import TracerProvider +from opentelemetry.sdk.trace.export import SimpleSpanProcessor + +provider = TracerProvider( + resource=Resource.create({"service.name": "durable-functions-python-worker"})) +provider.add_span_processor(SimpleSpanProcessor(OTLPSpanExporter())) +trace.set_tracer_provider(provider) +tracer = trace.get_tracer("issue-179-user-code") + +app = df.DFApp(http_auth_level=func.AuthLevel.ANONYMOUS) + + +@app.route(route="ping", methods=["GET"]) +def ping(req: func.HttpRequest) -> func.HttpResponse: + return func.HttpResponse("pong") + + +@app.route(route="start/{name}", methods=["POST"]) +@app.durable_client_input(client_name="client") +async def start_orchestration( + req: func.HttpRequest, + client: df.DurableFunctionsClient) -> func.HttpResponse: + with tracer.start_as_current_span("user-starter") as span: + instance_id = await client.schedule_new_orchestration( + req.route_params["name"]) + span.set_attribute("test.instance_id", instance_id) + return func.HttpResponse( + json.dumps({"id": instance_id}), + status_code=202, + mimetype="application/json", + ) + + +@app.route(route="status/{id}", methods=["GET"]) +@app.durable_client_input(client_name="client") +async def get_status( + req: func.HttpRequest, + client: df.DurableFunctionsClient) -> func.HttpResponse: + state = await client.get_orchestration_state( + req.route_params["id"], fetch_payloads=True) + payload = { + "runtimeStatus": state.runtime_status.name if state else None, + "output": state.serialized_output if state else None, + } + return func.HttpResponse( + json.dumps(payload), + mimetype="application/json", + ) + + +@app.orchestration_trigger(context_name="context") +def correlated_orchestrator( + ctx: task.OrchestrationContext, + _: Any) -> str: + with tracer.start_as_current_span( + "user-orchestrator", + attributes={"test.instance_id": ctx.instance_id}): + return "done" diff --git a/tests/azure-functions-durable/e2e/apps/tracing/host.json b/tests/azure-functions-durable/e2e/apps/tracing/host.json new file mode 100644 index 00000000..4f13f77a --- /dev/null +++ b/tests/azure-functions-durable/e2e/apps/tracing/host.json @@ -0,0 +1,19 @@ +{ + "version": "2.0", + "telemetryMode": "OpenTelemetry", + "extensionBundle": { + "id": "Microsoft.Azure.Functions.ExtensionBundle.Preview", + "version": "[4.*, 5.0.0)" + }, + "extensions": { + "durableTask": { + "storageProvider": { + "type": "AzureStorage" + }, + "tracing": { + "DistributedTracingEnabled": true, + "Version": "V2" + } + } + } +} diff --git a/tests/azure-functions-durable/e2e/apps/tracing/local.settings.json b/tests/azure-functions-durable/e2e/apps/tracing/local.settings.json new file mode 100644 index 00000000..a2ded917 --- /dev/null +++ b/tests/azure-functions-durable/e2e/apps/tracing/local.settings.json @@ -0,0 +1,7 @@ +{ + "IsEncrypted": false, + "Values": { + "AzureWebJobsStorage": "UseDevelopmentStorage=true", + "FUNCTIONS_WORKER_RUNTIME": "python" + } +} diff --git a/tests/azure-functions-durable/e2e/apps/tracing/requirements.txt b/tests/azure-functions-durable/e2e/apps/tracing/requirements.txt new file mode 100644 index 00000000..f26a3871 --- /dev/null +++ b/tests/azure-functions-durable/e2e/apps/tracing/requirements.txt @@ -0,0 +1,5 @@ +# The E2E harness installs local SDK packages and the OTLP exporter into the +# linked app virtual environment before starting the Functions host. +azure-functions +azure-functions-durable +opentelemetry-exporter-otlp-proto-grpc diff --git a/tests/azure-functions-durable/e2e/conftest.py b/tests/azure-functions-durable/e2e/conftest.py index a4b8c4eb..f15e2d3a 100644 --- a/tests/azure-functions-durable/e2e/conftest.py +++ b/tests/azure-functions-durable/e2e/conftest.py @@ -17,9 +17,17 @@ all test modules that use the same app). """ +import os +from collections.abc import Generator + import pytest -from ._harness import FunctionApp, azurite_is_running, func_executable +from ._harness import ( + FunctionApp, + OtelCollector, + azurite_is_running, + func_executable, +) def _require_prerequisites(app_name: str) -> None: @@ -46,3 +54,31 @@ def dtask_app(): _require_prerequisites("dtask_style") with FunctionApp("dtask_style") as app: yield app + + +@pytest.fixture(scope="session") +def tracing_app( + tmp_path_factory: pytest.TempPathFactory, +) -> Generator[tuple[FunctionApp, OtelCollector], None, None]: + _require_prerequisites("tracing") + if not OtelCollector.is_available(): + pytest.skip("Docker is required for the OpenTelemetry Collector.") + + work_dir = tmp_path_factory.mktemp("otel-collector") + old_endpoint = os.environ.get("OTEL_EXPORTER_OTLP_ENDPOINT") + old_protocol = os.environ.get("OTEL_EXPORTER_OTLP_PROTOCOL") + with OtelCollector(work_dir) as collector: + os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = collector.endpoint + os.environ["OTEL_EXPORTER_OTLP_PROTOCOL"] = "grpc" + try: + with FunctionApp("tracing") as app: + yield app, collector + finally: + if old_endpoint is None: + os.environ.pop("OTEL_EXPORTER_OTLP_ENDPOINT", None) + else: + os.environ["OTEL_EXPORTER_OTLP_ENDPOINT"] = old_endpoint + if old_protocol is None: + os.environ.pop("OTEL_EXPORTER_OTLP_PROTOCOL", None) + else: + os.environ["OTEL_EXPORTER_OTLP_PROTOCOL"] = old_protocol diff --git a/tests/azure-functions-durable/e2e/test_distributed_tracing_e2e.py b/tests/azure-functions-durable/e2e/test_distributed_tracing_e2e.py new file mode 100644 index 00000000..4787a709 --- /dev/null +++ b/tests/azure-functions-durable/e2e/test_distributed_tracing_e2e.py @@ -0,0 +1,81 @@ +# Copyright (c) Microsoft Corporation. +# Licensed under the MIT License. + +"""End-to-end distributed tracing validation using an OTEL collector.""" + +import time +from typing import Any + +import pytest + +from ._harness import FunctionApp, OtelCollector + +pytestmark = pytest.mark.functions_e2e + + +def _attribute_value(span: dict[str, Any], key: str) -> str | None: + for attribute in span.get("attributes", []): + if attribute.get("key") == key: + return attribute.get("value", {}).get("stringValue") + return None + + +def test_user_span_correlates_to_host_without_worker_lifecycle_duplicates( + tracing_app: tuple[FunctionApp, OtelCollector]) -> None: + app, collector = tracing_app + instance_id = app.start_orchestration("correlated_orchestrator") + status = app.wait_for_completion(instance_id) + assert status["runtimeStatus"] == "COMPLETED" + + deadline = time.time() + 30 + spans: list[dict[str, Any]] = [] + user_spans: list[dict[str, Any]] = [] + starter_spans: list[dict[str, Any]] = [] + parent_spans: list[dict[str, Any]] = [] + while time.time() < deadline: + spans = collector.get_spans() + user_spans = [ + span for span in spans + if span.get("name") == "user-orchestrator" + and _attribute_value(span, "test.instance_id") == instance_id + ] + starter_spans = [ + span for span in spans + if span.get("name") == "user-starter" + and _attribute_value(span, "test.instance_id") == instance_id + ] + if user_spans: + user_span = user_spans[0] + parent_spans = [ + span for span in spans + if span.get("traceId") == user_span.get("traceId") + and span.get("spanId") == user_span.get("parentSpanId") + ] + if starter_spans and parent_spans: + break + time.sleep(0.5) + + assert len(user_spans) == 1, collector.get_logs() + user_span = user_spans[0] + summary = [ + ( + span.get("name"), + span.get("scopeName"), + span.get("traceId"), + span.get("spanId"), + span.get("parentSpanId"), + ) + for span in spans + ] + assert len(starter_spans) == 1, summary + assert starter_spans[0].get("traceId") == user_span.get("traceId"), summary + assert len(parent_spans) == 1, summary + assert parent_spans[0].get("scopeName") != "durabletask" + assert parent_spans[0].get("spanId") != starter_spans[0].get("spanId"), summary + + python_lifecycle_spans = [ + span for span in spans + if span.get("traceId") == user_span.get("traceId") + if span.get("scopeName") == "durabletask" + ] + assert python_lifecycle_spans == [] diff --git a/tests/azure-functions-durable/e2e/test_dtask_entities_advanced_e2e.py b/tests/azure-functions-durable/e2e/test_dtask_entities_advanced_e2e.py index 364b6c68..2077d978 100644 --- a/tests/azure-functions-durable/e2e/test_dtask_entities_advanced_e2e.py +++ b/tests/azure-functions-durable/e2e/test_dtask_entities_advanced_e2e.py @@ -82,8 +82,7 @@ def test_call_failing_entity_handled(dtask_app): def test_client_delayed_signal_is_deferred(dtask_app): key = f"client-delay-{int(time.time() * 1000)}" - # Schedule the signal ~3s in the future. - dtask_app.signal_entity("counter", key, "add", input=9, delay_seconds=3) + dtask_app.signal_entity("counter", key, "add", input=9, delay_seconds=3.0) # It must not be delivered immediately. time.sleep(1) diff --git a/tests/azure-functions-durable/test_client_compat.py b/tests/azure-functions-durable/test_client_compat.py index c7c8b503..57105dca 100644 --- a/tests/azure-functions-durable/test_client_compat.py +++ b/tests/azure-functions-durable/test_client_compat.py @@ -49,6 +49,20 @@ def test_client_handles_null_max_grpc_message_size(): assert client.maxGrpcMessageSizeInBytes == 0 +def test_durable_clients_use_propagate_only_tracing(): + sync_client = df.SyncDurableFunctionsClient(_CLIENT_CONFIG) + try: + assert sync_client.emit_trace_spans is False + finally: + sync_client.close() + + with patch.object(AsyncTaskHubGrpcClient, "__init__", return_value=None) as init: + df.DurableFunctionsClient(_CLIENT_CONFIG) + + assert init.call_args is not None + assert init.call_args.kwargs["emit_trace_spans"] is False + + def test_client_handles_all_config_fields_sent_as_null(): # Newer host extension bundles serialize the full client configuration and # can send any field explicitly as ``null``. Every field must collapse to diff --git a/tests/azure-functions-durable/test_worker_compat.py b/tests/azure-functions-durable/test_worker_compat.py index b438a20a..ff82da7d 100644 --- a/tests/azure-functions-durable/test_worker_compat.py +++ b/tests/azure-functions-durable/test_worker_compat.py @@ -27,6 +27,12 @@ TEST_INSTANCE_ID = "inst-123" +def test_worker_uses_propagate_only_tracing(): + worker = DurableFunctionsWorker() + + assert worker.emit_trace_spans is False + + def _encode_orchestrator_request(name, encoded_input=None, instance_id=TEST_INSTANCE_ID): """Build a base64-encoded ``OrchestratorRequest`` for a single new dispatch.""" request = pb.OrchestratorRequest(instanceId=instance_id) diff --git a/tests/durabletask/test_tracing.py b/tests/durabletask/test_tracing.py index 81159191..dfdeae25 100644 --- a/tests/durabletask/test_tracing.py +++ b/tests/durabletask/test_tracing.py @@ -6,7 +6,7 @@ import json import logging from datetime import datetime, timezone -from unittest.mock import patch +from unittest.mock import AsyncMock, MagicMock, patch import pytest from google.protobuf import timestamp_pb2, wrappers_pb2 @@ -21,7 +21,7 @@ import durabletask.internal.helpers as helpers import durabletask.internal.orchestrator_service_pb2 as pb import durabletask.internal.tracing as tracing -from durabletask import task, worker +from durabletask import client, task, worker logging.basicConfig( format='%(asctime)s.%(msecs)03d %(name)s %(levelname)s: %(message)s', @@ -178,6 +178,25 @@ def test_creates_child_span_from_trace_context(self, otel_setup: InMemorySpanExp assert child_span.context is not None assert child_span.context.trace_id == int("0af7651916cd43dd8448eb211c80319c", 16) + def test_suppressed_span_activates_parent_for_user_span( + self, otel_setup: InMemorySpanExporter): + """Suppression should preserve propagation without an SDK span.""" + tracer = trace.get_tracer("user-code") + + with tracing.suppress_span_emission(): + with tracing.start_span( + "activity:sdk", trace_context=_make_parent_trace_ctx()) as span: + assert span is None + with tracer.start_as_current_span("user-activity"): + pass + + spans = otel_setup.get_finished_spans() + assert [span.name for span in spans] == ["user-activity"] + assert spans[0].context is not None + assert spans[0].context.trace_id == int(_SAMPLE_TRACE_ID, 16) + assert spans[0].parent is not None + assert spans[0].parent.span_id == int(_SAMPLE_PARENT_SPAN_ID, 16) + class TestSetSpanError: """Tests for tracing.set_span_error().""" @@ -537,6 +556,67 @@ def test_creates_producer_span(self, otel_setup: InMemorySpanExporter): assert s.attributes[tracing.ATTR_EVENT_TARGET_INSTANCE_ID] == "inst-456" +class TestClientSpanEmission: + """Tests for client propagate-only tracing.""" + + def test_clients_emit_trace_spans_by_default(self): + with patch( + "durabletask.client.stubs.TaskHubSidecarServiceStub", + return_value=MagicMock()): + sync_client = client.TaskHubGrpcClient(channel=MagicMock()) + async_client = client.AsyncTaskHubGrpcClient(channel=MagicMock()) + + assert sync_client.emit_trace_spans is True + assert async_client.emit_trace_spans is True + + def test_sync_client_propagates_ambient_context_without_lifecycle_spans( + self, otel_setup: InMemorySpanExporter): + stub = MagicMock() + stub.StartInstance.return_value = pb.CreateInstanceResponse(instanceId="inst-1") + with patch( + "durabletask.client.stubs.TaskHubSidecarServiceStub", + return_value=stub): + grpc_client = client.TaskHubGrpcClient( + channel=MagicMock(), emit_trace_spans=False) + + tracer = trace.get_tracer("host") + with tracer.start_as_current_span("host-request") as host_span: + host_span_id = f"{host_span.get_span_context().span_id:016x}" + grpc_client.schedule_new_orchestration( + "orchestrator", instance_id="inst-1") + grpc_client.raise_orchestration_event("inst-1", "event") + + start_request = stub.StartInstance.call_args.args[0] + assert start_request.parentTraceContext.spanID == host_span_id + assert [span.name for span in otel_setup.get_finished_spans()] == [ + "host-request"] + + @pytest.mark.asyncio + async def test_async_client_propagates_ambient_context_without_lifecycle_spans( + self, otel_setup: InMemorySpanExporter): + stub = MagicMock() + stub.StartInstance = AsyncMock( + return_value=pb.CreateInstanceResponse(instanceId="inst-1")) + stub.RaiseEvent = AsyncMock(return_value=pb.RaiseEventResponse()) + with patch( + "durabletask.client.stubs.TaskHubSidecarServiceStub", + return_value=stub): + grpc_client = client.AsyncTaskHubGrpcClient( + channel=MagicMock(), emit_trace_spans=False) + + tracer = trace.get_tracer("host") + with tracer.start_as_current_span("host-request") as host_span: + host_span_id = f"{host_span.get_span_context().span_id:016x}" + await grpc_client.schedule_new_orchestration( + "orchestrator", instance_id="inst-1") + await grpc_client.raise_orchestration_event("inst-1", "event") + + start_request = stub.StartInstance.call_args.args[0] + assert start_request.parentTraceContext.spanID == host_span_id + assert [span.name for span in otel_setup.get_finished_spans()] == [ + "host-request"] + + class TestOrchestrationServerSpan: """Tests for emit_orchestration_span.""" @@ -1172,6 +1252,51 @@ def orchestrator(ctx: task.OrchestrationContext, _): assert len(orch_spans) == 1 assert orch_spans[0].status.status_code == StatusCode.ERROR + def test_propagate_only_exports_user_span_under_inbound_parent( + self, otel_setup): + """Propagate-only mode should export user spans, not SDK lifecycle spans.""" + from unittest.mock import MagicMock + + tracer = trace.get_tracer("user-code") + + def orchestrator(ctx: task.OrchestrationContext, _): + with tracer.start_as_current_span("user-orchestrator"): + return "done" + + registry = worker._Registry() + name = registry.add_orchestrator(orchestrator) + w = worker.TaskHubGrpcWorker( + host_address="localhost:4001", + emit_trace_spans=False, + ) + w._registry = registry + stub = MagicMock() + req = pb.OrchestratorRequest( + instanceId=TEST_INSTANCE_ID, + newEvents=[ + helpers.new_orchestrator_started_event(), + helpers.new_execution_started_event( + name, + TEST_INSTANCE_ID, + encoded_input=None, + parent_trace_context=_make_parent_trace_ctx(), + ), + ], + ) + + w._execute_orchestrator(req, stub, "token1") + + spans = otel_setup.get_finished_spans() + assert [span.name for span in spans] == ["user-orchestrator"] + user_span = spans[0] + assert user_span.context is not None + assert user_span.context.trace_id == int(_SAMPLE_TRACE_ID, 16) + assert user_span.parent is not None + assert user_span.parent.span_id == int(_SAMPLE_PARENT_SPAN_ID, 16) + response = stub.CompleteOrchestratorTask.call_args.args[0] + assert response.orchestrationTraceContext.ListFields() == [] + assert tracing.get_current_trace_context() is None + def test_separate_instances_get_separate_spans(self, otel_setup): """Two different orchestration instances should produce independent spans when each completes."""