Add propagate-only distributed tracing for Durable Functions

.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: |

.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: |

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

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.

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":

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]

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"
 ]

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,