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..4f27bdef94 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,13 @@ 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 io.opentelemetry.context.Scope; import javax.annotation.Nullable; @@ -38,6 +44,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 +67,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 +144,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 +204,68 @@ 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)); + // 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; + } 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..ce7ee5fc48 --- /dev/null +++ b/durabletask-client/src/test/java/io/dapr/durabletask/DurableTaskGrpcClientTracingTest.java @@ -0,0 +1,154 @@ +/* + * 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.SpanContext; +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<>(); + private final AtomicReference spanContextDuringCall = 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); + // 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()); + 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])); + + // 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(); + } + } + + @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 bb4e63d76d..69b59163c0 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 @@ -26,6 +26,7 @@ import io.dapr.workflows.runtime.DefaultWorkflowState; import io.grpc.ClientInterceptor; import io.grpc.ManagedChannel; +import io.opentelemetry.api.trace.Tracer; import javax.annotation.Nullable; @@ -55,7 +56,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); } /** @@ -67,16 +84,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); } /** @@ -444,12 +462,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(); } private static NewOrchestrationInstanceOptions fromNewWorkflowOptions(NewWorkflowOptions options) { 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 2faf1e1080..eb718c5dfc 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; @@ -21,10 +23,12 @@ import io.dapr.workflows.WorkflowContext; import io.dapr.workflows.WorkflowStub; import io.grpc.ManagedChannel; +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; @@ -38,6 +42,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; @@ -83,6 +89,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();