Suppress Durable Functions client trace spans

Co-authored-by: Copilot App <223556219+Copilot@users.noreply.github.com>

Copilot-Session: b872ac61-5bac-40e3-b9c7-daf295725fde

Author: andystaples

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

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

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

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,

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 == []

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

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