From e90f470d60e6bafc1e5f99801ecae1b7ffac86fa Mon Sep 17 00:00:00 2001 From: Javier Aliaga Date: Mon, 6 Jul 2026 18:02:45 +0200 Subject: [PATCH 1/2] feat: propagate caller trace context when scheduling workflows Upstream half of workflow trace propagation (follow-up to #1619). DaprWorkflowClient gains a DaprWorkflowClient(Properties, Tracer) constructor that passes the tracer to the internal DurableTaskGrpcClientBuilder. DurableTaskGrpcClient now uses the configured tracer when scheduling: it emits a create_orchestration CLIENT span as a child of the caller's current OpenTelemetry context and stamps its W3C trace context into CreateInstanceRequest.parentTraceContext, so the workflow execution nests under the caller's trace instead of starting a separate one. Tracing stays opt-in: without a tracer the request is unchanged and no spans are emitted. The unused GlobalOpenTelemetry fallback is removed so the global instance is never pulled implicitly. Signed-off-by: Javier Aliaga --- durabletask-client/pom.xml | 5 + .../durabletask/DurableTaskGrpcClient.java | 72 ++++++++- .../DurableTaskGrpcClientTracingTest.java | 144 ++++++++++++++++++ sdk-workflows/pom.xml | 4 + .../workflows/client/DaprWorkflowClient.java | 40 ++++- .../client/DaprWorkflowClientTest.java | 37 +++++ 6 files changed, 286 insertions(+), 16 deletions(-) create mode 100644 durabletask-client/src/test/java/io/dapr/durabletask/DurableTaskGrpcClientTracingTest.java diff --git a/durabletask-client/pom.xml b/durabletask-client/pom.xml index bc8f12fea5..9d0e4e8aea 100644 --- a/durabletask-client/pom.xml +++ b/durabletask-client/pom.xml @@ -90,6 +90,11 @@ io.opentelemetry opentelemetry-context + + io.opentelemetry + opentelemetry-sdk + test + diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcClient.java b/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcClient.java index e66e6f3084..bbbe0566e8 100644 --- a/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcClient.java +++ b/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcClient.java @@ -15,6 +15,7 @@ import com.google.protobuf.StringValue; import com.google.protobuf.Timestamp; +import io.dapr.durabletask.implementation.protobuf.Orchestration; import io.dapr.durabletask.implementation.protobuf.OrchestratorService; import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc; import io.grpc.Channel; @@ -28,8 +29,12 @@ import io.grpc.netty.GrpcSslContexts; import io.grpc.netty.NettyChannelBuilder; import io.netty.handler.ssl.util.InsecureTrustManagerFactory; -import io.opentelemetry.api.GlobalOpenTelemetry; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.api.trace.StatusCode; import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator; +import io.opentelemetry.context.Context; import javax.annotation.Nullable; @@ -38,6 +43,8 @@ import java.io.InputStream; import java.time.Duration; import java.time.Instant; +import java.util.HashMap; +import java.util.Map; import java.util.Optional; import java.util.UUID; import java.util.concurrent.TimeUnit; @@ -59,6 +66,9 @@ public final class DurableTaskGrpcClient extends DurableTaskClient { private final DataConverter dataConverter; private final ManagedChannel managedSidecarChannel; private final TaskHubSidecarServiceGrpc.TaskHubSidecarServiceBlockingStub sidecarClient; + + // Optional. When null, scheduling emits no spans and no trace context is propagated (legacy behavior). + @Nullable private final Tracer tracer; DurableTaskGrpcClient(DurableTaskGrpcClientBuilder builder) { @@ -133,12 +143,7 @@ public final class DurableTaskGrpcClient extends DurableTaskClient { sidecarGrpcChannel = this.managedSidecarChannel; } - if (builder.tracer != null) { - this.tracer = builder.tracer; - } else { - //this.tracer = OpenTelemetry.noop().getTracer("DurableTaskGrpcClient"); - this.tracer = GlobalOpenTelemetry.getTracer("dapr-workflow"); - } + this.tracer = builder.tracer; this.sidecarClient = TaskHubSidecarServiceGrpc.newBlockingStub(sidecarGrpcChannel); } @@ -198,14 +203,65 @@ public String scheduleNewOrchestrationInstance( builder.setScheduledStartTimestamp(ts); } + Span span = null; + if (this.tracer != null) { + span = this.tracer.spanBuilder("create_orchestration:" + orchestratorName) + .setSpanKind(SpanKind.CLIENT) + .setAttribute("durabletask.type", "orchestration") + .setAttribute("durabletask.task.name", orchestratorName) + .setAttribute("durabletask.task.instance_id", instanceId) + .startSpan(); + Orchestration.TraceContext parentTraceContext = buildTraceContext(span); + if (parentTraceContext != null) { + builder.setParentTraceContext(parentTraceContext); + } + } + AtomicReference response = new AtomicReference<>(); OrchestratorService.CreateInstanceRequest request = builder.build(); - response.set(this.sidecarClient.startInstance(request)); + try { + response.set(this.sidecarClient.startInstance(request)); + } catch (RuntimeException e) { + if (span != null) { + span.setStatus(StatusCode.ERROR, "Failed to schedule orchestration instance"); + } + throw e; + } finally { + if (span != null) { + span.end(); + } + } return response.get().getInstanceId(); } + /** + * Serializes the span's context to a W3C trace context, so the scheduled orchestration + * (and its activities) are recorded as children of the caller's trace. + * Returns null when the span context is invalid (e.g. a no-op tracer), in which case + * no trace context is attached to the request. + */ + @Nullable + private static Orchestration.TraceContext buildTraceContext(Span span) { + Map carrier = new HashMap<>(); + W3CTraceContextPropagator.getInstance().inject(Context.current().with(span), carrier, Map::put); + + String traceParent = carrier.get("traceparent"); + if (traceParent == null || traceParent.isEmpty()) { + return null; + } + + Orchestration.TraceContext.Builder traceContext = Orchestration.TraceContext.newBuilder() + .setTraceParent(traceParent); + String traceState = carrier.get("tracestate"); + if (traceState != null && !traceState.isEmpty()) { + traceContext.setTraceState(StringValue.of(traceState)); + } + + return traceContext.build(); + } + @Override public void raiseEvent(String instanceId, String eventName, Object eventPayload) { Helpers.throwIfArgumentNull(instanceId, "instanceId"); diff --git a/durabletask-client/src/test/java/io/dapr/durabletask/DurableTaskGrpcClientTracingTest.java b/durabletask-client/src/test/java/io/dapr/durabletask/DurableTaskGrpcClientTracingTest.java new file mode 100644 index 0000000000..1b87cf05c4 --- /dev/null +++ b/durabletask-client/src/test/java/io/dapr/durabletask/DurableTaskGrpcClientTracingTest.java @@ -0,0 +1,144 @@ +/* + * Copyright 2026 The Dapr Authors + * Licensed under the Apache License, Version 2.0 (the "License"); + * you may not use this file except in compliance with the License. + * You may obtain a copy of the License at + * http://www.apache.org/licenses/LICENSE-2.0 + * Unless required by applicable law or agreed to in writing, software + * distributed under the License is distributed on an "AS IS" BASIS, + * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. + * See the License for the specific language governing permissions and + * limitations under the License. + */ + +package io.dapr.durabletask; + +import io.dapr.durabletask.implementation.protobuf.OrchestratorService; +import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc; +import io.grpc.ManagedChannel; +import io.grpc.Server; +import io.grpc.inprocess.InProcessChannelBuilder; +import io.grpc.inprocess.InProcessServerBuilder; +import io.grpc.stub.StreamObserver; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.Tracer; +import io.opentelemetry.context.Scope; +import io.opentelemetry.sdk.trace.SdkTracerProvider; +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.concurrent.TimeUnit; +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertFalse; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Tests that scheduling an orchestration propagates the caller's trace context + * to the sidecar when a Tracer is configured, and stays untraced when not. + */ +class DurableTaskGrpcClientTracingTest { + + private static final String ORCHESTRATION_NAME = "TestOrchestration"; + + private Server server; + private ManagedChannel channel; + private DurableTaskClient client; + private final AtomicReference capturedRequest = new AtomicReference<>(); + + @BeforeEach + void setUp() throws Exception { + String serverName = InProcessServerBuilder.generateName(); + server = InProcessServerBuilder.forName(serverName) + .directExecutor() + .addService(new TaskHubSidecarServiceGrpc.TaskHubSidecarServiceImplBase() { + @Override + public void startInstance( + OrchestratorService.CreateInstanceRequest request, + StreamObserver responseObserver) { + capturedRequest.set(request); + responseObserver.onNext(OrchestratorService.CreateInstanceResponse.newBuilder() + .setInstanceId(request.getInstanceId()) + .build()); + responseObserver.onCompleted(); + } + }) + .build() + .start(); + channel = InProcessChannelBuilder.forName(serverName).directExecutor().build(); + } + + @AfterEach + void tearDown() throws Exception { + if (client != null) { + client.close(); + } + if (channel != null) { + channel.shutdownNow().awaitTermination(5, TimeUnit.SECONDS); + } + if (server != null) { + server.shutdownNow().awaitTermination(5, TimeUnit.SECONDS); + } + } + + @Test + void scheduleWithoutTracerDoesNotSetParentTraceContext() { + client = new DurableTaskGrpcClientBuilder() + .grpcChannel(channel) + .build(); + + client.scheduleNewOrchestrationInstance(ORCHESTRATION_NAME); + + assertFalse(capturedRequest.get().hasParentTraceContext()); + } + + @Test + void scheduleWithTracerPropagatesCallerTraceContext() { + SdkTracerProvider tracerProvider = SdkTracerProvider.builder().build(); + try { + Tracer tracer = tracerProvider.get("test"); + client = new DurableTaskGrpcClientBuilder() + .grpcChannel(channel) + .tracer(tracer) + .build(); + + Span callerSpan = tracer.spanBuilder("caller").startSpan(); + try (Scope scope = callerSpan.makeCurrent()) { + client.scheduleNewOrchestrationInstance(ORCHESTRATION_NAME); + } finally { + callerSpan.end(); + } + + OrchestratorService.CreateInstanceRequest request = capturedRequest.get(); + assertTrue(request.hasParentTraceContext()); + + // traceparent format: 00--- + String traceParent = request.getParentTraceContext().getTraceParent(); + String[] parts = traceParent.split("-"); + assertEquals(4, parts.length); + // The scheduled orchestration must join the caller's trace, through a child + // span distinct from the caller's own span. + assertEquals(callerSpan.getSpanContext().getTraceId(), parts[1]); + assertFalse(callerSpan.getSpanContext().getSpanId().equals(parts[2])); + } finally { + tracerProvider.close(); + } + } + + @Test + void scheduleWithNoOpTracerDoesNotSetParentTraceContext() { + // A tracer that produces invalid span contexts (e.g. OpenTelemetry no-op) + // must not attach an unusable trace context to the request. + Tracer noopTracer = io.opentelemetry.api.OpenTelemetry.noop().getTracer("noop"); + client = new DurableTaskGrpcClientBuilder() + .grpcChannel(channel) + .tracer(noopTracer) + .build(); + + client.scheduleNewOrchestrationInstance(ORCHESTRATION_NAME); + + assertFalse(capturedRequest.get().hasParentTraceContext()); + } +} diff --git a/sdk-workflows/pom.xml b/sdk-workflows/pom.xml index 36c0e07403..0d52f5d9b1 100644 --- a/sdk-workflows/pom.xml +++ b/sdk-workflows/pom.xml @@ -42,6 +42,10 @@ durabletask-client ${project.parent.version} + + io.opentelemetry + opentelemetry-api + diff --git a/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java b/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java index de9fe466e1..d97b1e288b 100644 --- a/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java +++ b/sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java @@ -28,6 +28,7 @@ import io.grpc.ManagedChannel; import io.grpc.Status; import io.grpc.StatusRuntimeException; +import io.opentelemetry.api.trace.Tracer; import javax.annotation.Nullable; @@ -58,7 +59,23 @@ public DaprWorkflowClient() { * @param properties Properties for the GRPC Channel. */ public DaprWorkflowClient(Properties properties) { - this(NetworkUtils.buildGrpcManagedChannel(properties, new ApiTokenClientInterceptor(properties))); + this(properties, (Tracer) null); + } + + /** + * Public constructor for DaprWorkflowClient. This layer constructs the GRPC Channel. + * + *

When a {@link Tracer} is provided, scheduling a workflow emits a client span as a child of + * the caller's current OpenTelemetry context and propagates that trace context to the Dapr + * sidecar, so the workflow execution is recorded as part of the caller's trace. + * + * @param properties Properties for the GRPC Channel. + * @param tracer OpenTelemetry Tracer used to emit and propagate trace context when + * scheduling workflows. May be null, in which case tracing is disabled + * and this constructor behaves exactly like {@link #DaprWorkflowClient(Properties)}. + */ + public DaprWorkflowClient(Properties properties, @Nullable Tracer tracer) { + this(NetworkUtils.buildGrpcManagedChannel(properties, new ApiTokenClientInterceptor(properties)), tracer); } /** @@ -70,16 +87,17 @@ public DaprWorkflowClient(Properties properties) { * @param additionalInterceptors extra interceptors appended after the API-token interceptor. */ protected DaprWorkflowClient(Properties properties, ClientInterceptor... additionalInterceptors) { - this(buildChannelWithAdditional(properties, additionalInterceptors)); + this(buildChannelWithAdditional(properties, additionalInterceptors), null); } /** * Private Constructor that passes a created DurableTaskClient and the new GRPC channel. * * @param grpcChannel ManagedChannel for GRPC channel. + * @param tracer optional Tracer used to propagate trace context when scheduling workflows. */ - private DaprWorkflowClient(ManagedChannel grpcChannel) { - this(createDurableTaskClient(grpcChannel), grpcChannel); + private DaprWorkflowClient(ManagedChannel grpcChannel, @Nullable Tracer tracer) { + this(createDurableTaskClient(grpcChannel, tracer), grpcChannel); } /** @@ -454,12 +472,18 @@ private static ManagedChannel buildChannelWithAdditional( * Static method to create the DurableTaskClient. * * @param grpcChannel ManagedChannel for GRPC. + * @param tracer optional Tracer set on the underlying client; skipped when null. * @return a new instance of a DurableTaskClient with a GRPC channel. */ - private static DurableTaskClient createDurableTaskClient(ManagedChannel grpcChannel) { - return new DurableTaskGrpcClientBuilder() - .grpcChannel(grpcChannel) - .build(); + private static DurableTaskClient createDurableTaskClient(ManagedChannel grpcChannel, @Nullable Tracer tracer) { + DurableTaskGrpcClientBuilder builder = new DurableTaskGrpcClientBuilder() + .grpcChannel(grpcChannel); + + if (tracer != null) { + builder.tracer(tracer); + } + + return builder.build(); } /** diff --git a/sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientTest.java b/sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientTest.java index 208cd66dc3..f88a7dbcc0 100644 --- a/sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientTest.java +++ b/sdk-workflows/src/test/java/io/dapr/workflows/client/DaprWorkflowClientTest.java @@ -13,7 +13,9 @@ package io.dapr.workflows.client; +import io.dapr.config.Properties; import io.dapr.durabletask.DurableTaskClient; +import io.dapr.durabletask.DurableTaskGrpcClientBuilder; import io.dapr.durabletask.NewOrchestrationInstanceOptions; import io.dapr.durabletask.OrchestrationMetadata; import io.dapr.durabletask.OrchestrationRuntimeStatus; @@ -23,10 +25,12 @@ import io.grpc.ManagedChannel; import io.grpc.Status; import io.grpc.StatusRuntimeException; +import io.opentelemetry.api.trace.Tracer; import org.junit.jupiter.api.BeforeAll; import org.junit.jupiter.api.BeforeEach; import org.junit.jupiter.api.Test; import org.mockito.ArgumentCaptor; +import org.mockito.MockedConstruction; import java.lang.reflect.Constructor; import java.time.Duration; @@ -42,6 +46,8 @@ import static org.mockito.ArgumentMatchers.any; import static org.mockito.ArgumentMatchers.eq; import static org.mockito.Mockito.mock; +import static org.mockito.Mockito.mockConstruction; +import static org.mockito.Mockito.never; import static org.mockito.Mockito.times; import static org.mockito.Mockito.verify; import static org.mockito.Mockito.when; @@ -87,6 +93,37 @@ public void EmptyConstructor() { assertDoesNotThrow(() -> new DaprWorkflowClient()); } + @Test + public void tracerConstructorPassesTracerToDurableTaskClientBuilder() throws Exception { + Tracer tracer = mock(Tracer.class); + + try (MockedConstruction construction = + mockConstruction(DurableTaskGrpcClientBuilder.class, (builder, context) -> { + when(builder.grpcChannel(any())).thenReturn(builder); + when(builder.tracer(any())).thenReturn(builder); + when(builder.build()).thenReturn(mockInnerClient); + })) { + new DaprWorkflowClient(new Properties(), tracer).close(); + + DurableTaskGrpcClientBuilder builder = construction.constructed().get(0); + verify(builder, times(1)).tracer(tracer); + } + } + + @Test + public void nullTracerIsNotPassedToDurableTaskClientBuilder() throws Exception { + try (MockedConstruction construction = + mockConstruction(DurableTaskGrpcClientBuilder.class, (builder, context) -> { + when(builder.grpcChannel(any())).thenReturn(builder); + when(builder.build()).thenReturn(mockInnerClient); + })) { + new DaprWorkflowClient(new Properties(), (Tracer) null).close(); + + DurableTaskGrpcClientBuilder builder = construction.constructed().get(0); + verify(builder, never()).tracer(any()); + } + } + @Test public void scheduleNewWorkflowWithArgName() { String expectedName = TestWorkflow.class.getCanonicalName(); From bff244dd4faf3062d004752aba2dcec06b67abe8 Mon Sep 17 00:00:00 2001 From: Javier Aliaga Date: Tue, 7 Jul 2026 10:22:15 +0200 Subject: [PATCH 2/2] fix: make scheduling span current during startInstance and record exceptions Addresses review feedback: the create_orchestration span is now current while the sidecar call runs, so nested instrumentation attaches to it, and failures record the exception on the span in addition to the error status. Signed-off-by: Javier Aliaga --- .../io/dapr/durabletask/DurableTaskGrpcClient.java | 6 +++++- .../durabletask/DurableTaskGrpcClientTracingTest.java | 10 ++++++++++ 2 files changed, 15 insertions(+), 1 deletion(-) diff --git a/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcClient.java b/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcClient.java index bbbe0566e8..4f27bdef94 100644 --- a/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcClient.java +++ b/durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcClient.java @@ -35,6 +35,7 @@ import io.opentelemetry.api.trace.Tracer; import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator; import io.opentelemetry.context.Context; +import io.opentelemetry.context.Scope; import javax.annotation.Nullable; @@ -220,10 +221,13 @@ public String scheduleNewOrchestrationInstance( AtomicReference response = new AtomicReference<>(); OrchestratorService.CreateInstanceRequest request = builder.build(); - try { + // Make the span current during the call so instrumentation running underneath + // (e.g. gRPC OpenTelemetry interceptors) attaches to it. + try (Scope ignored = span != null ? span.makeCurrent() : Scope.noop()) { response.set(this.sidecarClient.startInstance(request)); } catch (RuntimeException e) { if (span != null) { + span.recordException(e); span.setStatus(StatusCode.ERROR, "Failed to schedule orchestration instance"); } throw e; diff --git a/durabletask-client/src/test/java/io/dapr/durabletask/DurableTaskGrpcClientTracingTest.java b/durabletask-client/src/test/java/io/dapr/durabletask/DurableTaskGrpcClientTracingTest.java index 1b87cf05c4..ce7ee5fc48 100644 --- a/durabletask-client/src/test/java/io/dapr/durabletask/DurableTaskGrpcClientTracingTest.java +++ b/durabletask-client/src/test/java/io/dapr/durabletask/DurableTaskGrpcClientTracingTest.java @@ -21,6 +21,7 @@ import io.grpc.inprocess.InProcessServerBuilder; import io.grpc.stub.StreamObserver; import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanContext; import io.opentelemetry.api.trace.Tracer; import io.opentelemetry.context.Scope; import io.opentelemetry.sdk.trace.SdkTracerProvider; @@ -47,6 +48,7 @@ class DurableTaskGrpcClientTracingTest { private ManagedChannel channel; private DurableTaskClient client; private final AtomicReference capturedRequest = new AtomicReference<>(); + private final AtomicReference spanContextDuringCall = new AtomicReference<>(); @BeforeEach void setUp() throws Exception { @@ -59,6 +61,9 @@ public void startInstance( OrchestratorService.CreateInstanceRequest request, StreamObserver responseObserver) { capturedRequest.set(request); + // directExecutor() runs this handler on the client thread, so this observes + // the span the client made current for the duration of the call. + spanContextDuringCall.set(Span.current().getSpanContext()); responseObserver.onNext(OrchestratorService.CreateInstanceResponse.newBuilder() .setInstanceId(request.getInstanceId()) .build()); @@ -122,6 +127,11 @@ void scheduleWithTracerPropagatesCallerTraceContext() { // span distinct from the caller's own span. assertEquals(callerSpan.getSpanContext().getTraceId(), parts[1]); assertFalse(callerSpan.getSpanContext().getSpanId().equals(parts[2])); + + // The scheduling span must be current while the sidecar call runs, so nested + // instrumentation attaches to it. It must match the propagated trace context. + assertEquals(parts[1], spanContextDuringCall.get().getTraceId()); + assertEquals(parts[2], spanContextDuringCall.get().getSpanId()); } finally { tracerProvider.close(); }