Skip to content

Commit c3d4ac8

Browse files
javier-aliagajavier-aliaga
authored
feat: propagate caller trace context when scheduling workflows (#1783)
* 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 <javier@diagrid.io> * 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 <javier@diagrid.io> --------- Signed-off-by: Javier Aliaga <javier@diagrid.io>
1 parent 3c12975 commit c3d4ac8

6 files changed

Lines changed: 300 additions & 16 deletions

File tree

‎durabletask-client/pom.xml‎

Lines changed: 5 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -90,6 +90,11 @@
9090
<groupId>io.opentelemetry</groupId>
9191
<artifactId>opentelemetry-context</artifactId>
9292
</dependency>
93+
<dependency>
94+
<groupId>io.opentelemetry</groupId>
95+
<artifactId>opentelemetry-sdk</artifactId>
96+
<scope>test</scope>
97+
</dependency>
9398
</dependencies>
9499
<build>
95100
<plugins>

‎durabletask-client/src/main/java/io/dapr/durabletask/DurableTaskGrpcClient.java‎

Lines changed: 68 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -15,6 +15,7 @@
1515

1616
import com.google.protobuf.StringValue;
1717
import com.google.protobuf.Timestamp;
18+
import io.dapr.durabletask.implementation.protobuf.Orchestration;
1819
import io.dapr.durabletask.implementation.protobuf.OrchestratorService;
1920
import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc;
2021
import io.grpc.Channel;
@@ -28,8 +29,13 @@
2829
import io.grpc.netty.GrpcSslContexts;
2930
import io.grpc.netty.NettyChannelBuilder;
3031
import io.netty.handler.ssl.util.InsecureTrustManagerFactory;
31-
import io.opentelemetry.api.GlobalOpenTelemetry;
32+
import io.opentelemetry.api.trace.Span;
33+
import io.opentelemetry.api.trace.SpanKind;
34+
import io.opentelemetry.api.trace.StatusCode;
3235
import io.opentelemetry.api.trace.Tracer;
36+
import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator;
37+
import io.opentelemetry.context.Context;
38+
import io.opentelemetry.context.Scope;
3339

3440
import javax.annotation.Nullable;
3541

@@ -38,6 +44,8 @@
3844
import java.io.InputStream;
3945
import java.time.Duration;
4046
import java.time.Instant;
47+
import java.util.HashMap;
48+
import java.util.Map;
4149
import java.util.Optional;
4250
import java.util.UUID;
4351
import java.util.concurrent.TimeUnit;
@@ -59,6 +67,9 @@ public final class DurableTaskGrpcClient extends DurableTaskClient {
5967
private final DataConverter dataConverter;
6068
private final ManagedChannel managedSidecarChannel;
6169
private final TaskHubSidecarServiceGrpc.TaskHubSidecarServiceBlockingStub sidecarClient;
70+
71+
// Optional. When null, scheduling emits no spans and no trace context is propagated (legacy behavior).
72+
@Nullable
6273
private final Tracer tracer;
6374

6475
DurableTaskGrpcClient(DurableTaskGrpcClientBuilder builder) {
@@ -133,12 +144,7 @@ public final class DurableTaskGrpcClient extends DurableTaskClient {
133144
sidecarGrpcChannel = this.managedSidecarChannel;
134145
}
135146

136-
if (builder.tracer != null) {
137-
this.tracer = builder.tracer;
138-
} else {
139-
//this.tracer = OpenTelemetry.noop().getTracer("DurableTaskGrpcClient");
140-
this.tracer = GlobalOpenTelemetry.getTracer("dapr-workflow");
141-
}
147+
this.tracer = builder.tracer;
142148

143149
this.sidecarClient = TaskHubSidecarServiceGrpc.newBlockingStub(sidecarGrpcChannel);
144150
}
@@ -198,14 +204,68 @@ public String scheduleNewOrchestrationInstance(
198204
builder.setScheduledStartTimestamp(ts);
199205
}
200206

207+
Span span = null;
208+
if (this.tracer != null) {
209+
span = this.tracer.spanBuilder("create_orchestration:" + orchestratorName)
210+
.setSpanKind(SpanKind.CLIENT)
211+
.setAttribute("durabletask.type", "orchestration")
212+
.setAttribute("durabletask.task.name", orchestratorName)
213+
.setAttribute("durabletask.task.instance_id", instanceId)
214+
.startSpan();
215+
Orchestration.TraceContext parentTraceContext = buildTraceContext(span);
216+
if (parentTraceContext != null) {
217+
builder.setParentTraceContext(parentTraceContext);
218+
}
219+
}
220+
201221
AtomicReference<OrchestratorService.CreateInstanceResponse> response = new AtomicReference<>();
202222

203223
OrchestratorService.CreateInstanceRequest request = builder.build();
204-
response.set(this.sidecarClient.startInstance(request));
224+
// Make the span current during the call so instrumentation running underneath
225+
// (e.g. gRPC OpenTelemetry interceptors) attaches to it.
226+
try (Scope ignored = span != null ? span.makeCurrent() : Scope.noop()) {
227+
response.set(this.sidecarClient.startInstance(request));
228+
} catch (RuntimeException e) {
229+
if (span != null) {
230+
span.recordException(e);
231+
span.setStatus(StatusCode.ERROR, "Failed to schedule orchestration instance");
232+
}
233+
throw e;
234+
} finally {
235+
if (span != null) {
236+
span.end();
237+
}
238+
}
205239

206240
return response.get().getInstanceId();
207241
}
208242

243+
/**
244+
* Serializes the span's context to a W3C trace context, so the scheduled orchestration
245+
* (and its activities) are recorded as children of the caller's trace.
246+
* Returns null when the span context is invalid (e.g. a no-op tracer), in which case
247+
* no trace context is attached to the request.
248+
*/
249+
@Nullable
250+
private static Orchestration.TraceContext buildTraceContext(Span span) {
251+
Map<String, String> carrier = new HashMap<>();
252+
W3CTraceContextPropagator.getInstance().inject(Context.current().with(span), carrier, Map::put);
253+
254+
String traceParent = carrier.get("traceparent");
255+
if (traceParent == null || traceParent.isEmpty()) {
256+
return null;
257+
}
258+
259+
Orchestration.TraceContext.Builder traceContext = Orchestration.TraceContext.newBuilder()
260+
.setTraceParent(traceParent);
261+
String traceState = carrier.get("tracestate");
262+
if (traceState != null && !traceState.isEmpty()) {
263+
traceContext.setTraceState(StringValue.of(traceState));
264+
}
265+
266+
return traceContext.build();
267+
}
268+
209269
@Override
210270
public void raiseEvent(String instanceId, String eventName, Object eventPayload) {
211271
Helpers.throwIfArgumentNull(instanceId, "instanceId");

‎durabletask-client/src/test/java/io/dapr/durabletask/DurableTaskGrpcClientTracingTest.java‎

Lines changed: 154 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,154 @@
1+
/*
2+
* Copyright 2026 The Dapr Authors
3+
* Licensed under the Apache License, Version 2.0 (the "License");
4+
* you may not use this file except in compliance with the License.
5+
* You may obtain a copy of the License at
6+
* http://www.apache.org/licenses/LICENSE-2.0
7+
* Unless required by applicable law or agreed to in writing, software
8+
* distributed under the License is distributed on an "AS IS" BASIS,
9+
* WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
10+
* See the License for the specific language governing permissions and
11+
* limitations under the License.
12+
*/
13+
14+
package io.dapr.durabletask;
15+
16+
import io.dapr.durabletask.implementation.protobuf.OrchestratorService;
17+
import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc;
18+
import io.grpc.ManagedChannel;
19+
import io.grpc.Server;
20+
import io.grpc.inprocess.InProcessChannelBuilder;
21+
import io.grpc.inprocess.InProcessServerBuilder;
22+
import io.grpc.stub.StreamObserver;
23+
import io.opentelemetry.api.trace.Span;
24+
import io.opentelemetry.api.trace.SpanContext;
25+
import io.opentelemetry.api.trace.Tracer;
26+
import io.opentelemetry.context.Scope;
27+
import io.opentelemetry.sdk.trace.SdkTracerProvider;
28+
import org.junit.jupiter.api.AfterEach;
29+
import org.junit.jupiter.api.BeforeEach;
30+
import org.junit.jupiter.api.Test;
31+
32+
import java.util.concurrent.TimeUnit;
33+
import java.util.concurrent.atomic.AtomicReference;
34+
35+
import static org.junit.jupiter.api.Assertions.assertEquals;
36+
import static org.junit.jupiter.api.Assertions.assertFalse;
37+
import static org.junit.jupiter.api.Assertions.assertTrue;
38+
39+
/**
40+
* Tests that scheduling an orchestration propagates the caller's trace context
41+
* to the sidecar when a Tracer is configured, and stays untraced when not.
42+
*/
43+
class DurableTaskGrpcClientTracingTest {
44+
45+
private static final String ORCHESTRATION_NAME = "TestOrchestration";
46+
47+
private Server server;
48+
private ManagedChannel channel;
49+
private DurableTaskClient client;
50+
private final AtomicReference<OrchestratorService.CreateInstanceRequest> capturedRequest = new AtomicReference<>();
51+
private final AtomicReference<SpanContext> spanContextDuringCall = new AtomicReference<>();
52+
53+
@BeforeEach
54+
void setUp() throws Exception {
55+
String serverName = InProcessServerBuilder.generateName();
56+
server = InProcessServerBuilder.forName(serverName)
57+
.directExecutor()
58+
.addService(new TaskHubSidecarServiceGrpc.TaskHubSidecarServiceImplBase() {
59+
@Override
60+
public void startInstance(
61+
OrchestratorService.CreateInstanceRequest request,
62+
StreamObserver<OrchestratorService.CreateInstanceResponse> responseObserver) {
63+
capturedRequest.set(request);
64+
// directExecutor() runs this handler on the client thread, so this observes
65+
// the span the client made current for the duration of the call.
66+
spanContextDuringCall.set(Span.current().getSpanContext());
67+
responseObserver.onNext(OrchestratorService.CreateInstanceResponse.newBuilder()
68+
.setInstanceId(request.getInstanceId())
69+
.build());
70+
responseObserver.onCompleted();
71+
}
72+
})
73+
.build()
74+
.start();
75+
channel = InProcessChannelBuilder.forName(serverName).directExecutor().build();
76+
}
77+
78+
@AfterEach
79+
void tearDown() throws Exception {
80+
if (client != null) {
81+
client.close();
82+
}
83+
if (channel != null) {
84+
channel.shutdownNow().awaitTermination(5, TimeUnit.SECONDS);
85+
}
86+
if (server != null) {
87+
server.shutdownNow().awaitTermination(5, TimeUnit.SECONDS);
88+
}
89+
}
90+
91+
@Test
92+
void scheduleWithoutTracerDoesNotSetParentTraceContext() {
93+
client = new DurableTaskGrpcClientBuilder()
94+
.grpcChannel(channel)
95+
.build();
96+
97+
client.scheduleNewOrchestrationInstance(ORCHESTRATION_NAME);
98+
99+
assertFalse(capturedRequest.get().hasParentTraceContext());
100+
}
101+
102+
@Test
103+
void scheduleWithTracerPropagatesCallerTraceContext() {
104+
SdkTracerProvider tracerProvider = SdkTracerProvider.builder().build();
105+
try {
106+
Tracer tracer = tracerProvider.get("test");
107+
client = new DurableTaskGrpcClientBuilder()
108+
.grpcChannel(channel)
109+
.tracer(tracer)
110+
.build();
111+
112+
Span callerSpan = tracer.spanBuilder("caller").startSpan();
113+
try (Scope scope = callerSpan.makeCurrent()) {
114+
client.scheduleNewOrchestrationInstance(ORCHESTRATION_NAME);
115+
} finally {
116+
callerSpan.end();
117+
}
118+
119+
OrchestratorService.CreateInstanceRequest request = capturedRequest.get();
120+
assertTrue(request.hasParentTraceContext());
121+
122+
// traceparent format: 00-<trace-id>-<span-id>-<flags>
123+
String traceParent = request.getParentTraceContext().getTraceParent();
124+
String[] parts = traceParent.split("-");
125+
assertEquals(4, parts.length);
126+
// The scheduled orchestration must join the caller's trace, through a child
127+
// span distinct from the caller's own span.
128+
assertEquals(callerSpan.getSpanContext().getTraceId(), parts[1]);
129+
assertFalse(callerSpan.getSpanContext().getSpanId().equals(parts[2]));
130+
131+
// The scheduling span must be current while the sidecar call runs, so nested
132+
// instrumentation attaches to it. It must match the propagated trace context.
133+
assertEquals(parts[1], spanContextDuringCall.get().getTraceId());
134+
assertEquals(parts[2], spanContextDuringCall.get().getSpanId());
135+
} finally {
136+
tracerProvider.close();
137+
}
138+
}
139+
140+
@Test
141+
void scheduleWithNoOpTracerDoesNotSetParentTraceContext() {
142+
// A tracer that produces invalid span contexts (e.g. OpenTelemetry no-op)
143+
// must not attach an unusable trace context to the request.
144+
Tracer noopTracer = io.opentelemetry.api.OpenTelemetry.noop().getTracer("noop");
145+
client = new DurableTaskGrpcClientBuilder()
146+
.grpcChannel(channel)
147+
.tracer(noopTracer)
148+
.build();
149+
150+
client.scheduleNewOrchestrationInstance(ORCHESTRATION_NAME);
151+
152+
assertFalse(capturedRequest.get().hasParentTraceContext());
153+
}
154+
}

‎sdk-workflows/pom.xml‎

Lines changed: 4 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -42,6 +42,10 @@
4242
<artifactId>durabletask-client</artifactId>
4343
<version>${project.parent.version}</version>
4444
</dependency>
45+
<dependency>
46+
<groupId>io.opentelemetry</groupId>
47+
<artifactId>opentelemetry-api</artifactId>
48+
</dependency>
4549
</dependencies>
4650

4751
<build>

‎sdk-workflows/src/main/java/io/dapr/workflows/client/DaprWorkflowClient.java‎

Lines changed: 32 additions & 8 deletions
Original file line numberDiff line numberDiff line change
@@ -28,6 +28,7 @@
2828
import io.grpc.ManagedChannel;
2929
import io.grpc.Status;
3030
import io.grpc.StatusRuntimeException;
31+
import io.opentelemetry.api.trace.Tracer;
3132

3233
import javax.annotation.Nullable;
3334

@@ -58,7 +59,23 @@ public DaprWorkflowClient() {
5859
* @param properties Properties for the GRPC Channel.
5960
*/
6061
public DaprWorkflowClient(Properties properties) {
61-
this(NetworkUtils.buildGrpcManagedChannel(properties, new ApiTokenClientInterceptor(properties)));
62+
this(properties, (Tracer) null);
63+
}
64+
65+
/**
66+
* Public constructor for DaprWorkflowClient. This layer constructs the GRPC Channel.
67+
*
68+
* <p>When a {@link Tracer} is provided, scheduling a workflow emits a client span as a child of
69+
* the caller's current OpenTelemetry context and propagates that trace context to the Dapr
70+
* sidecar, so the workflow execution is recorded as part of the caller's trace.
71+
*
72+
* @param properties Properties for the GRPC Channel.
73+
* @param tracer OpenTelemetry Tracer used to emit and propagate trace context when
74+
* scheduling workflows. May be null, in which case tracing is disabled
75+
* and this constructor behaves exactly like {@link #DaprWorkflowClient(Properties)}.
76+
*/
77+
public DaprWorkflowClient(Properties properties, @Nullable Tracer tracer) {
78+
this(NetworkUtils.buildGrpcManagedChannel(properties, new ApiTokenClientInterceptor(properties)), tracer);
6279
}
6380

6481
/**
@@ -70,16 +87,17 @@ public DaprWorkflowClient(Properties properties) {
7087
* @param additionalInterceptors extra interceptors appended after the API-token interceptor.
7188
*/
7289
protected DaprWorkflowClient(Properties properties, ClientInterceptor... additionalInterceptors) {
73-
this(buildChannelWithAdditional(properties, additionalInterceptors));
90+
this(buildChannelWithAdditional(properties, additionalInterceptors), null);
7491
}
7592

7693
/**
7794
* Private Constructor that passes a created DurableTaskClient and the new GRPC channel.
7895
*
7996
* @param grpcChannel ManagedChannel for GRPC channel.
97+
* @param tracer optional Tracer used to propagate trace context when scheduling workflows.
8098
*/
81-
private DaprWorkflowClient(ManagedChannel grpcChannel) {
82-
this(createDurableTaskClient(grpcChannel), grpcChannel);
99+
private DaprWorkflowClient(ManagedChannel grpcChannel, @Nullable Tracer tracer) {
100+
this(createDurableTaskClient(grpcChannel, tracer), grpcChannel);
83101
}
84102

85103
/**
@@ -454,12 +472,18 @@ private static ManagedChannel buildChannelWithAdditional(
454472
* Static method to create the DurableTaskClient.
455473
*
456474
* @param grpcChannel ManagedChannel for GRPC.
475+
* @param tracer optional Tracer set on the underlying client; skipped when null.
457476
* @return a new instance of a DurableTaskClient with a GRPC channel.
458477
*/
459-
private static DurableTaskClient createDurableTaskClient(ManagedChannel grpcChannel) {
460-
return new DurableTaskGrpcClientBuilder()
461-
.grpcChannel(grpcChannel)
462-
.build();
478+
private static DurableTaskClient createDurableTaskClient(ManagedChannel grpcChannel, @Nullable Tracer tracer) {
479+
DurableTaskGrpcClientBuilder builder = new DurableTaskGrpcClientBuilder()
480+
.grpcChannel(grpcChannel);
481+
482+
if (tracer != null) {
483+
builder.tracer(tracer);
484+
}
485+
486+
return builder.build();
463487
}
464488

465489
/**

0 commit comments

Comments
 (0)