adding tracing

Author: salaboy

client/build.gradle

@@ -11,12 +11,14 @@ plugins {
 }
 
 group 'io.dapr'
-version = '1.5.10'
+version = '1.5.11-SNAPSHOT'
 archivesBaseName = 'durabletask-client'
 
 def grpcVersion = '1.69.0'
 def protocVersion = '3.25.5'
 def jacksonVersion = '2.15.3'
+def micrometerVersion = '1.16.0'
+def otelVersion = '1.56.0'
 // Java 11 is now the minimum required version for both compilation and testing
 
 dependencies {
@@ -25,6 +27,8 @@ dependencies {
     implementation "io.grpc:grpc-protobuf:${grpcVersion}"
     implementation "io.grpc:grpc-stub:${grpcVersion}"
     implementation 'com.google.protobuf:protobuf-java:3.25.5'
+    implementation "io.micrometer:micrometer-observation:${micrometerVersion}"
+    implementation "io.opentelemetry:opentelemetry-api:${otelVersion}"
     runtimeOnly "io.grpc:grpc-netty-shaded:${grpcVersion}"
 
     compileOnly "org.apache.tomcat:annotations-api:6.0.53"

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

@@ -4,6 +4,8 @@
 
 import com.google.protobuf.StringValue;
 import com.google.protobuf.Timestamp;
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.api.trace.Tracer;
 import io.dapr.durabletask.implementation.protobuf.OrchestratorService.*;
 import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc;
 import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc.*;
@@ -38,6 +40,7 @@ public final class DurableTaskGrpcClient extends DurableTaskClient {
     private final DataConverter dataConverter;
     private final ManagedChannel managedSidecarChannel;
     private final TaskHubSidecarServiceBlockingStub sidecarClient;
+    private final Tracer tracer;
 
     DurableTaskGrpcClient(DurableTaskGrpcClientBuilder builder) {
         this.dataConverter = builder.dataConverter != null ? builder.dataConverter : new JacksonDataConverter();
@@ -109,8 +112,9 @@ public final class DurableTaskGrpcClient extends DurableTaskClient {
             // Need to keep track of this channel so we can dispose it on close()
             this.managedSidecarChannel = channelBuilder.build();
             sidecarGrpcChannel = this.managedSidecarChannel;
-        }
 
+        }
+        this.tracer = builder.tracer;
         this.sidecarClient = TaskHubSidecarServiceGrpc.newBlockingStub(sidecarGrpcChannel);
     }
 
@@ -169,6 +173,23 @@ public String scheduleNewOrchestrationInstance(
             builder.setScheduledStartTimestamp(ts);
         }
 
+        if (tracer != null) {
+            Span span = tracer.spanBuilder("dapr.workflow.grpc.startInstance")
+              .startSpan();
+            try {
+              TraceContext.Builder traceContextBuilder = TraceContext.newBuilder();
+              traceContextBuilder.setTraceParent(span.getSpanContext().getTraceId());
+              traceContextBuilder.setSpanID(span.getSpanContext().getSpanId());
+              builder.setParentTraceContext(traceContextBuilder.build());
+
+              CreateInstanceRequest request = builder.build();
+              CreateInstanceResponse response = this.sidecarClient.startInstance(request);
+              return response.getInstanceId();
+            } finally {
+              span.end();
+            }
+        }
+
         CreateInstanceRequest request = builder.build();
         CreateInstanceResponse response = this.sidecarClient.startInstance(request);
         return response.getInstanceId();
@@ -180,11 +201,27 @@ public void raiseEvent(String instanceId, String eventName, Object eventPayload)
         Helpers.throwIfArgumentNull(eventName, "eventName");
 
         RaiseEventRequest.Builder builder = RaiseEventRequest.newBuilder()
-                .setInstanceId(instanceId)
-                .setName(eventName);
+            .setInstanceId(instanceId)
+            .setName(eventName);
         if (eventPayload != null) {
-            String serializedPayload = this.dataConverter.serialize(eventPayload);
-            builder.setInput(StringValue.of(serializedPayload));
+          String serializedPayload = this.dataConverter.serialize(eventPayload);
+          builder.setInput(StringValue.of(serializedPayload));
+        }
+
+        if (tracer != null) {
+          Span span = tracer.spanBuilder("dapr.workflow.grpc.raiseEvent")
+              .startSpan();
+          try {
+            TraceContext.Builder traceContextBuilder = TraceContext.newBuilder();
+            traceContextBuilder.setTraceParent(span.getSpanContext().getTraceId());
+            traceContextBuilder.setSpanID(span.getSpanContext().getSpanId());
+            builder.setParentTraceContext(traceContextBuilder.build());
+
+            RaiseEventRequest request = builder.build();
+            this.sidecarClient.raiseEvent(request);
+          } finally {
+            span.end();
+          }
         }
 
         RaiseEventRequest request = builder.build();

client/src/main/java/io/dapr/durabletask/DurableTaskGrpcClientBuilder.java

@@ -3,6 +3,7 @@
 package io.dapr.durabletask;
 
 import io.grpc.Channel;
+import io.opentelemetry.api.trace.Tracer;
 
 /**
  * Builder class for constructing new {@link DurableTaskClient} objects that communicate with a sidecar process
@@ -16,6 +17,7 @@ public final class DurableTaskGrpcClientBuilder {
     String tlsCertPath;
     String tlsKeyPath;
     boolean insecure;
+    Tracer tracer;
 
     /**
      * Sets the {@link DataConverter} to use for converting serializable data payloads.
@@ -57,6 +59,17 @@ public DurableTaskGrpcClientBuilder port(int port) {
         return this;
     }
 
+  /**
+   * Sets the Tracer object to be used by DurableTaskClient to emit traces
+   *
+   * @param tracer to  be used by the DurableTaskClient
+   * @return this builder object
+   */
+  public DurableTaskGrpcClientBuilder tracer(Tracer tracer) {
+    this.tracer = tracer;
+    return this;
+  }
+
     /**
      * Sets the path to the TLS CA certificate file for server authentication.
      * If not set, the system's default CA certificates will be used.

client/src/main/java/io/dapr/durabletask/DurableTaskGrpcWorker.java

@@ -200,6 +200,7 @@ public void startAndBlock() {
                                         activityRequest.getName(),
                                         activityRequest.getInput().getValue(),
                                         activityRequest.getTaskExecutionId(),
+                                        activityRequest.getParentTraceContext().getTraceParent(),
                                         activityRequest.getTaskId());
                             } catch (Throwable e) {
                                 failureDetails = TaskFailureDetails.newBuilder()

client/src/main/java/io/dapr/durabletask/TaskActivityContext.java

@@ -34,4 +34,10 @@ public interface TaskActivityContext {
      * @return the task id of the current task activity
      */
     int getTaskId();
+
+    /**
+     * Gets the trace parent id for the current workflow execution.
+     * @return trace parent id
+     */
+     String getTraceParentId();
 }

client/src/main/java/io/dapr/durabletask/TaskActivityExecutor.java

@@ -19,7 +19,7 @@ public TaskActivityExecutor(
         this.logger = logger;
     }
 
-    public String execute(String taskName, String input, String taskExecutionId, int taskId) throws Throwable {
+    public String execute(String taskName, String input, String taskExecutionId, String parentTraceId, int taskId) throws Throwable {
         TaskActivityFactory factory = this.activityFactories.get(taskName);
         if (factory == null) {
             throw new IllegalStateException(
@@ -32,7 +32,7 @@ public String execute(String taskName, String input, String taskExecutionId, int
                     String.format("The task factory '%s' returned a null TaskActivity object.", taskName));
         }
 
-        TaskActivityContextImpl context = new TaskActivityContextImpl(taskName, input, taskExecutionId, taskId);
+        TaskActivityContextImpl context = new TaskActivityContextImpl(taskName, input, taskExecutionId, parentTraceId, taskId);
 
         // Unhandled exceptions are allowed to escape
         Object output = activity.run(context);
@@ -48,13 +48,17 @@ private class TaskActivityContextImpl implements TaskActivityContext {
         private final String rawInput;
         private final String taskExecutionId;
         private final int taskId;
+        private final String traceParentId;
 
         private final DataConverter dataConverter = TaskActivityExecutor.this.dataConverter;
 
-        public TaskActivityContextImpl(String activityName, String rawInput, String taskExecutionId, int taskId) {
+        public TaskActivityContextImpl(String activityName, String rawInput, String taskExecutionId,
+                                       String traceParentId,
+                                       int taskId) {
             this.name = activityName;
             this.rawInput = rawInput;
             this.taskExecutionId = taskExecutionId;
+            this.traceParentId = traceParentId;
             this.taskId = taskId;
         }
 
@@ -81,5 +85,10 @@ public String getTaskExecutionId() {
         public int getTaskId() {
             return this.taskId;
         }
+
+        @Override
+        public String getTraceParentId() {
+          return traceParentId;
+        }
     }
 }

internal/durabletask-protobuf/PROTO_SOURCE_COMMIT_HASH

@@ -1 +1 @@
-4b86756497d875b97f9a91051781b5711c1e4fa6
+474626b111db2d10e6d75b80dd277cf5993543ef

internal/durabletask-protobuf/protos/orchestrator_service.proto

@@ -477,6 +477,17 @@ message PurgeInstancesRequest {
         PurgeInstanceFilter purgeInstanceFilter = 2;
     }
     bool recursive = 3;
+
+    // force will force a purge of a workflow, regardless of its current
+    // runtime state, or whether an active worker can process it, the backend
+    // will attempt to delete it anyway. This neccessarily means the purging is
+    // executed out side of the workflow state machine, and therefore, can lead
+    // to corrupt state or broken workflow execution. Usage of this should
+    // _only_ be used when the client knows the workflow is not being currently
+    // processed. It is highly recommended to avoid using this flag unless
+    // absolutely necessary.
+    // Defaults to false.
+    optional bool force = 4;
 }
 
 message PurgeInstanceFilter {
@@ -739,6 +750,9 @@ service TaskHubSidecarService {
 
     // Rerun a Workflow from a specific event ID of a workflow instance.
     rpc RerunWorkflowFromEvent(RerunWorkflowFromEventRequest) returns (RerunWorkflowFromEventResponse);
+
+    rpc ListInstanceIDs (ListInstanceIDsRequest) returns (ListInstanceIDsResponse);
+    rpc GetInstanceHistory (GetInstanceHistoryRequest) returns (GetInstanceHistoryResponse);
 }
 
 message GetWorkItemsRequest {
@@ -820,3 +834,36 @@ message RerunWorkflowFromEventRequest {
 message RerunWorkflowFromEventResponse {
     string newInstanceID = 1;
 }
+
+// ListInstanceIDsRequest is used to list all orchestration instances.
+message ListInstanceIDsRequest {
+    // continuationToken is the continuation token to use for pagination. This
+    // is the token which the next page should start from. If not given, the
+    // first page will be returned.
+    optional string continuationToken = 1;
+
+    // pageSize is the maximum number of instances to return for this page. If
+    // not given, all instances will be attempted to be returned.
+    optional uint32 pageSize = 2;
+}
+
+// ListInstanceIDsResponse is the response to executing ListInstanceIDs.
+message ListInstanceIDsResponse {
+  // instanceIds is the list of instance IDs returned.
+  repeated string instanceIds = 1;
+
+  // continuationToken is the continuation token to use for pagination. If
+  // there are no more pages, this will be null.
+  optional string continuationToken = 2;
+}
+
+// GetInstanceHistoryRequest is used to get the full history of an
+// orchestration instance.
+message GetInstanceHistoryRequest {
+    string instanceId = 1;
+}
+
+// GetInstanceHistoryResponse is the response to executing GetInstanceHistory.
+message GetInstanceHistoryResponse {
+    repeated HistoryEvent events = 1;
+}