From 450b31271dee64f834855abc5ce460eb4520e0f1 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Tue, 28 Jul 2026 13:25:55 -0600 Subject: [PATCH 1/9] Add propagate-only distributed tracing Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: b872ac61-5bac-40e3-b9c7-daf295725fde --- CHANGELOG.md | 4 + azure-functions-durable/CHANGELOG.md | 8 +- .../azure/durable_functions/worker.py | 5 +- azure-functions-durable/pyproject.toml | 2 +- durabletask/internal/tracing.py | 57 +++++-- durabletask/worker.py | 150 ++++++++++-------- noxfile.py | 4 +- pyproject.toml | 2 +- tests/azure-functions-durable/e2e/_harness.py | 134 ++++++++++++++++ .../e2e/apps/tracing/function_app.py | 70 ++++++++ .../e2e/apps/tracing/host.json | 19 +++ .../e2e/apps/tracing/local.settings.json | 7 + .../e2e/apps/tracing/requirements.txt | 5 + tests/azure-functions-durable/e2e/conftest.py | 38 ++++- .../e2e/test_distributed_tracing_e2e.py | 73 +++++++++ .../test_worker_compat.py | 6 + tests/durabletask/test_tracing.py | 64 ++++++++ 17 files changed, 564 insertions(+), 84 deletions(-) create mode 100644 tests/azure-functions-durable/e2e/apps/tracing/function_app.py create mode 100644 tests/azure-functions-durable/e2e/apps/tracing/host.json create mode 100644 tests/azure-functions-durable/e2e/apps/tracing/local.settings.json create mode 100644 tests/azure-functions-durable/e2e/apps/tracing/requirements.txt create mode 100644 tests/azure-functions-durable/e2e/test_distributed_tracing_e2e.py diff --git a/CHANGELOG.md b/CHANGELOG.md index c2015651..f138714a 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 `TaskHubGrpcWorker(..., emit_trace_spans=False)` for hosts that own +Durable Task lifecycle spans. In this propagate-only mode, incoming W3C trace +context remains active while orchestrator, activity, and entity user code runs, +without emitting duplicate worker 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 5bea7251..1f3c5eb6 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -9,6 +9,9 @@ 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, and entity lifecycle spans from the Python worker. - 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 @@ -178,8 +181,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/worker.py b/azure-functions-durable/azure/durable_functions/worker.py index 8f2bd727..f01fc00f 100644 --- a/azure-functions-durable/azure/durable_functions/worker.py +++ b/azure-functions-durable/azure/durable_functions/worker.py @@ -40,7 +40,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, + ) def add_named_orchestrator(self, name: str, func: task.Orchestrator[Any, Any]) -> None: self._registry.add_named_orchestrator(name, func) 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/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..7bb79a4a 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,127 @@ 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) + 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 + + @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..a1f423c9 --- /dev/null +++ b/tests/azure-functions-durable/e2e/apps/tracing/function_app.py @@ -0,0 +1,70 @@ +# 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: + instance_id = await client.schedule_new_orchestration( + req.route_params["name"]) + 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..c385a45a --- /dev/null +++ b/tests/azure-functions-durable/e2e/test_distributed_tracing_e2e.py @@ -0,0 +1,73 @@ +# 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]] = [] + 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 + ] + 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 parent_spans: + break + time.sleep(0.5) + + assert len(user_spans) == 1 + 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(parent_spans) == 1, summary + assert parent_spans[0].get("scopeName") != "durabletask" + + python_lifecycle_spans = [ + span for span in spans + if span.get("scopeName") == "durabletask" + and _attribute_value( + span, "durabletask.task.instance_id") == instance_id + ] + assert python_lifecycle_spans == [] diff --git a/tests/azure-functions-durable/test_worker_compat.py b/tests/azure-functions-durable/test_worker_compat.py index 227c1ef9..73ff888d 100644 --- a/tests/azure-functions-durable/test_worker_compat.py +++ b/tests/azure-functions-durable/test_worker_compat.py @@ -24,6 +24,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..a38b9e6a 100644 --- a/tests/durabletask/test_tracing.py +++ b/tests/durabletask/test_tracing.py @@ -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().""" @@ -1172,6 +1191,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.""" From 452f253bd37b07d3554c295e4df12761cfeb64d7 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Tue, 28 Jul 2026 13:37:55 -0600 Subject: [PATCH 2/9] Fix collector output permissions in CI Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: b872ac61-5bac-40e3-b9c7-daf295725fde --- tests/azure-functions-durable/e2e/_harness.py | 17 +++++++++++++++++ .../e2e/test_distributed_tracing_e2e.py | 2 +- 2 files changed, 18 insertions(+), 1 deletion(-) diff --git a/tests/azure-functions-durable/e2e/_harness.py b/tests/azure-functions-durable/e2e/_harness.py index 7bb79a4a..71264239 100644 --- a/tests/azure-functions-durable/e2e/_harness.py +++ b/tests/azure-functions-durable/e2e/_harness.py @@ -115,6 +115,10 @@ def start(self) -> 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" @@ -196,6 +200,19 @@ def get_spans(self) -> list[dict[str, Any]]: 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: diff --git a/tests/azure-functions-durable/e2e/test_distributed_tracing_e2e.py b/tests/azure-functions-durable/e2e/test_distributed_tracing_e2e.py index c385a45a..72d2a2d0 100644 --- a/tests/azure-functions-durable/e2e/test_distributed_tracing_e2e.py +++ b/tests/azure-functions-durable/e2e/test_distributed_tracing_e2e.py @@ -49,7 +49,7 @@ def test_user_span_correlates_to_host_without_worker_lifecycle_duplicates( break time.sleep(0.5) - assert len(user_spans) == 1 + assert len(user_spans) == 1, collector.get_logs() user_span = user_spans[0] summary = [ ( From 341418ae996cc9f9cb018a661b668bbd1b8d853e Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Tue, 28 Jul 2026 13:56:33 -0600 Subject: [PATCH 3/9] Stabilize delayed entity E2E polling Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: b872ac61-5bac-40e3-b9c7-daf295725fde --- tests/azure-functions-durable/e2e/apps/dtask_style/host.json | 3 ++- 1 file changed, 2 insertions(+), 1 deletion(-) diff --git a/tests/azure-functions-durable/e2e/apps/dtask_style/host.json b/tests/azure-functions-durable/e2e/apps/dtask_style/host.json index dd14c74c..a406f335 100644 --- a/tests/azure-functions-durable/e2e/apps/dtask_style/host.json +++ b/tests/azure-functions-durable/e2e/apps/dtask_style/host.json @@ -12,7 +12,8 @@ "extensions": { "durableTask": { "storageProvider": { - "type": "AzureStorage" + "type": "AzureStorage", + "maxQueuePollingInterval": "00:00:01" } } } From c5eba5a49eca3e966d8c84649a16cee4a6edf068 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Tue, 28 Jul 2026 21:56:27 -0600 Subject: [PATCH 4/9] Suppress Durable Functions client trace spans Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: b872ac61-5bac-40e3-b9c7-daf295725fde --- CHANGELOG.md | 8 +- azure-functions-durable/CHANGELOG.md | 3 +- .../azure/durable_functions/client.py | 6 +- durabletask/client.py | 143 ++++++++++-------- .../e2e/test_distributed_tracing_e2e.py | 3 +- .../test_client_compat.py | 11 ++ tests/durabletask/test_tracing.py | 65 +++++++- 7 files changed, 165 insertions(+), 74 deletions(-) diff --git a/CHANGELOG.md b/CHANGELOG.md index f138714a..03e21304 100644 --- a/CHANGELOG.md +++ b/CHANGELOG.md @@ -9,10 +9,10 @@ adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html). ADDED -- Added `TaskHubGrpcWorker(..., emit_trace_spans=False)` for hosts that own -Durable Task lifecycle spans. In this propagate-only mode, incoming W3C trace -context remains active while orchestrator, activity, and entity user code runs, -without emitting duplicate worker lifecycle spans. +- 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 1f3c5eb6..b44e127a 100644 --- a/azure-functions-durable/CHANGELOG.md +++ b/azure-functions-durable/CHANGELOG.md @@ -11,7 +11,8 @@ ADDED - Distributed tracing now correlates OpenTelemetry spans created by orchestrator user code with the Durable Functions host trace while avoiding duplicate -orchestration, activity, and entity lifecycle spans from the Python worker. +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 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/durabletask/client.py b/durabletask/client.py index 4fbeb554..f8d1ad69 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() @@ -1005,6 +1014,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, @@ -1128,27 +1143,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._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._stub.StartInstance(req) + return res.instanceId async def get_orchestration_state(self, instance_id: str, *, fetch_payloads: bool = True) -> OrchestrationState | None: @@ -1262,14 +1278,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._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._stub.RaiseEvent(req) async def terminate_orchestration(self, instance_id: str, *, output: Any | None = None, diff --git a/tests/azure-functions-durable/e2e/test_distributed_tracing_e2e.py b/tests/azure-functions-durable/e2e/test_distributed_tracing_e2e.py index 72d2a2d0..9e47e074 100644 --- a/tests/azure-functions-durable/e2e/test_distributed_tracing_e2e.py +++ b/tests/azure-functions-durable/e2e/test_distributed_tracing_e2e.py @@ -66,8 +66,7 @@ def test_user_span_correlates_to_host_without_worker_lifecycle_duplicates( python_lifecycle_spans = [ span for span in spans + if span.get("traceId") == user_span.get("traceId") if span.get("scopeName") == "durabletask" - and _attribute_value( - span, "durabletask.task.instance_id") == instance_id ] assert python_lifecycle_spans == [] diff --git a/tests/azure-functions-durable/test_client_compat.py b/tests/azure-functions-durable/test_client_compat.py index c7c8b503..2cb180c8 100644 --- a/tests/azure-functions-durable/test_client_compat.py +++ b/tests/azure-functions-durable/test_client_compat.py @@ -49,6 +49,17 @@ def test_client_handles_null_max_grpc_message_size(): assert client.maxGrpcMessageSizeInBytes == 0 +async def test_durable_clients_use_propagate_only_tracing(): + async_client = _make_client() + sync_client = df.SyncDurableFunctionsClient(_CLIENT_CONFIG) + try: + assert async_client.emit_trace_spans is False + assert sync_client.emit_trace_spans is False + finally: + await async_client.close() + sync_client.close() + + 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/durabletask/test_tracing.py b/tests/durabletask/test_tracing.py index a38b9e6a..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', @@ -556,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.""" From 24d25ab0375674556d87e3c26427d3bcc61d5bd3 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Tue, 28 Jul 2026 22:29:21 -0600 Subject: [PATCH 5/9] Avoid async channel creation in tracing config test Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: b872ac61-5bac-40e3-b9c7-daf295725fde --- tests/azure-functions-durable/test_client_compat.py | 11 +++++++---- 1 file changed, 7 insertions(+), 4 deletions(-) diff --git a/tests/azure-functions-durable/test_client_compat.py b/tests/azure-functions-durable/test_client_compat.py index 2cb180c8..57105dca 100644 --- a/tests/azure-functions-durable/test_client_compat.py +++ b/tests/azure-functions-durable/test_client_compat.py @@ -49,16 +49,19 @@ def test_client_handles_null_max_grpc_message_size(): assert client.maxGrpcMessageSizeInBytes == 0 -async def test_durable_clients_use_propagate_only_tracing(): - async_client = _make_client() +def test_durable_clients_use_propagate_only_tracing(): sync_client = df.SyncDurableFunctionsClient(_CLIENT_CONFIG) try: - assert async_client.emit_trace_spans is False assert sync_client.emit_trace_spans is False finally: - await async_client.close() 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 From 093df8c60a566841a3f468210eb82adedb7674ea Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Tue, 28 Jul 2026 22:41:11 -0600 Subject: [PATCH 6/9] Seed tracing E2E with application span Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: b872ac61-5bac-40e3-b9c7-daf295725fde --- .../e2e/apps/tracing/function_app.py | 6 ++++-- .../e2e/test_distributed_tracing_e2e.py | 11 ++++++++++- 2 files changed, 14 insertions(+), 3 deletions(-) diff --git a/tests/azure-functions-durable/e2e/apps/tracing/function_app.py b/tests/azure-functions-durable/e2e/apps/tracing/function_app.py index a1f423c9..275d0206 100644 --- a/tests/azure-functions-durable/e2e/apps/tracing/function_app.py +++ b/tests/azure-functions-durable/e2e/apps/tracing/function_app.py @@ -34,8 +34,10 @@ def ping(req: func.HttpRequest) -> func.HttpResponse: async def start_orchestration( req: func.HttpRequest, client: df.DurableFunctionsClient) -> func.HttpResponse: - instance_id = await client.schedule_new_orchestration( - req.route_params["name"]) + 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, diff --git a/tests/azure-functions-durable/e2e/test_distributed_tracing_e2e.py b/tests/azure-functions-durable/e2e/test_distributed_tracing_e2e.py index 9e47e074..4787a709 100644 --- a/tests/azure-functions-durable/e2e/test_distributed_tracing_e2e.py +++ b/tests/azure-functions-durable/e2e/test_distributed_tracing_e2e.py @@ -30,6 +30,7 @@ def test_user_span_correlates_to_host_without_worker_lifecycle_duplicates( 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() @@ -38,6 +39,11 @@ def test_user_span_correlates_to_host_without_worker_lifecycle_duplicates( 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 = [ @@ -45,7 +51,7 @@ def test_user_span_correlates_to_host_without_worker_lifecycle_duplicates( if span.get("traceId") == user_span.get("traceId") and span.get("spanId") == user_span.get("parentSpanId") ] - if parent_spans: + if starter_spans and parent_spans: break time.sleep(0.5) @@ -61,8 +67,11 @@ def test_user_span_correlates_to_host_without_worker_lifecycle_duplicates( ) 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 From 335adcc7a98f7e38ecc34c95f42f444253eb2a82 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Tue, 28 Jul 2026 23:02:32 -0600 Subject: [PATCH 7/9] Stabilize scheduled entity E2E timing Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: b872ac61-5bac-40e3-b9c7-daf295725fde --- tests/azure-functions-durable/e2e/apps/dtask_style/host.json | 3 +-- .../e2e/apps/dtask_style/orchestrators.py | 4 +++- .../e2e/test_dtask_entities_advanced_e2e.py | 5 +++-- 3 files changed, 7 insertions(+), 5 deletions(-) diff --git a/tests/azure-functions-durable/e2e/apps/dtask_style/host.json b/tests/azure-functions-durable/e2e/apps/dtask_style/host.json index a406f335..dd14c74c 100644 --- a/tests/azure-functions-durable/e2e/apps/dtask_style/host.json +++ b/tests/azure-functions-durable/e2e/apps/dtask_style/host.json @@ -12,8 +12,7 @@ "extensions": { "durableTask": { "storageProvider": { - "type": "AzureStorage", - "maxQueuePollingInterval": "00:00:01" + "type": "AzureStorage" } } } diff --git a/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py b/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py index 61a86386..7de12d9c 100644 --- a/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py +++ b/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py @@ -193,7 +193,9 @@ def signal_counter_delayed(ctx: task.OrchestrationContext, key: Any): # Schedules a delayed (future) signal to the counter entity, then completes # immediately. The signal should only be delivered after the delay elapses. entity_id = entities.EntityInstanceId("counter", key) - signal_at = ctx.current_utc_datetime + timedelta(seconds=3) + # Azure Queue visibility delays have one-second precision. The fractional + # offset keeps normal dispatch within Core's 100 ms early-delivery window. + signal_at = ctx.current_utc_datetime + timedelta(seconds=3.1) ctx.signal_entity(entity_id, "add", 4, signal_time=signal_at) return key 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..9de114be 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,9 @@ 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) + # The fractional offset accounts for Azure Queue's one-second visibility + # precision and keeps dispatch within Core's 100 ms early-delivery window. + dtask_app.signal_entity("counter", key, "add", input=9, delay_seconds=3.1) # It must not be delivered immediately. time.sleep(1) From 449e561dddc85f825a401175d626bb4328ffae59 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Tue, 28 Jul 2026 23:27:20 -0600 Subject: [PATCH 8/9] Use supported Node runtime for Azurite CI Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: b872ac61-5bac-40e3-b9c7-daf295725fde --- .github/workflows/durabletask-azurefunctions.yml | 2 +- .github/workflows/durabletask.yml | 4 ++-- .../e2e/apps/dtask_style/orchestrators.py | 4 +--- .../e2e/test_dtask_entities_advanced_e2e.py | 4 +--- 4 files changed, 5 insertions(+), 9 deletions(-) diff --git a/.github/workflows/durabletask-azurefunctions.yml b/.github/workflows/durabletask-azurefunctions.yml index a3260325..ceabb2bb 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: "21.x" - name: Install Azurite and Azure Functions Core Tools run: | diff --git a/.github/workflows/durabletask.yml b/.github/workflows/durabletask.yml index e249a9d2..36b30810 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: '21.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/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py b/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py index 7de12d9c..61a86386 100644 --- a/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py +++ b/tests/azure-functions-durable/e2e/apps/dtask_style/orchestrators.py @@ -193,9 +193,7 @@ def signal_counter_delayed(ctx: task.OrchestrationContext, key: Any): # Schedules a delayed (future) signal to the counter entity, then completes # immediately. The signal should only be delivered after the delay elapses. entity_id = entities.EntityInstanceId("counter", key) - # Azure Queue visibility delays have one-second precision. The fractional - # offset keeps normal dispatch within Core's 100 ms early-delivery window. - signal_at = ctx.current_utc_datetime + timedelta(seconds=3.1) + signal_at = ctx.current_utc_datetime + timedelta(seconds=3) ctx.signal_entity(entity_id, "add", 4, signal_time=signal_at) return key 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 9de114be..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,9 +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)}" - # The fractional offset accounts for Azure Queue's one-second visibility - # precision and keeps dispatch within Core's 100 ms early-delivery window. - dtask_app.signal_entity("counter", key, "add", input=9, delay_seconds=3.1) + dtask_app.signal_entity("counter", key, "add", input=9, delay_seconds=3.0) # It must not be delivered immediately. time.sleep(1) From b8952a6cec4e8397e88770732c6b4eb2e77a4bf1 Mon Sep 17 00:00:00 2001 From: Andy Staples Date: Tue, 28 Jul 2026 23:32:50 -0600 Subject: [PATCH 9/9] Use Node 22 for Azurite pipelines Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com> Copilot-Session: b872ac61-5bac-40e3-b9c7-daf295725fde --- .github/workflows/durabletask-azurefunctions.yml | 2 +- .github/workflows/durabletask.yml | 2 +- 2 files changed, 2 insertions(+), 2 deletions(-) diff --git a/.github/workflows/durabletask-azurefunctions.yml b/.github/workflows/durabletask-azurefunctions.yml index ceabb2bb..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: "21.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 36b30810..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: '21.x' + node-version: '22.x' - name: Cache npm uses: actions/cache@v3