chore: Refactor durable task and introduce telemetry

Signed-off-by: Javier Aliaga <[email protected]>

Author: javier-aliaga

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

@@ -16,7 +16,6 @@
 import com.google.protobuf.StringValue;
 import com.google.protobuf.Timestamp;
 import io.dapr.durabletask.implementation.protobuf.OrchestratorService;
-import io.dapr.durabletask.implementation.protobuf.OrchestratorService.TraceContext;
 import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc;
 import io.grpc.Channel;
 import io.grpc.ChannelCredentials;
@@ -30,7 +29,6 @@
 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.Tracer;
 
 import javax.annotation.Nullable;
@@ -46,7 +44,6 @@
 import java.util.concurrent.TimeUnit;
 import java.util.concurrent.TimeoutException;
 import java.util.concurrent.atomic.AtomicReference;
-import java.util.function.Consumer;
 import java.util.logging.Logger;
 
 /**
@@ -204,27 +201,30 @@ public String scheduleNewOrchestrationInstance(
 
     AtomicReference<OrchestratorService.CreateInstanceResponse> response = new AtomicReference<>();
 
-    this.observe("dapr.workflow.grpc.startInstance", (TraceContext context) -> {
-      builder.setParentTraceContext(context);
-      OrchestratorService.CreateInstanceRequest request = builder.build();
-      response.set(this.sidecarClient.startInstance(request));
-    });
+    OrchestratorService.CreateInstanceRequest request = builder.build();
+    response.set(this.sidecarClient.startInstance(request));
+
+    //    this.observe("dapr.workflow.grpc.startInstance", (TraceContext context) -> {
+    //      builder.setParentTraceContext(context);
+    //      OrchestratorService.CreateInstanceRequest request = builder.build();
+    //      response.set(this.sidecarClient.startInstance(request));
+    //    });
 
     return response.get().getInstanceId();
   }
 
-  private void observe(String spanName, Consumer<TraceContext> fn) {
-    Span span = tracer.spanBuilder(spanName).startSpan();
-    try {
-      TraceContext.Builder traceContextBuilder = TraceContext.newBuilder();
-      traceContextBuilder.setTraceParent(span.getSpanContext().getTraceId());
-      var context = traceContextBuilder.build();
-
-      fn.accept(context);
-    } finally {
-      span.end();
-    }
-  }
+  //  private void observe(String spanName, Consumer<TraceContext> fn) {
+  //    Span span = tracer.spanBuilder(spanName).startSpan();
+  //    try {
+  //      TraceContext.Builder traceContextBuilder = TraceContext.newBuilder();
+  //      traceContextBuilder.setTraceParent(span.getSpanContext().getTraceId());
+  //      var context = traceContextBuilder.build();
+  //
+  //      fn.accept(context);
+  //    } finally {
+  //      span.end();
+  //    }
+  //  }
 
   @Override
   public void raiseEvent(String instanceId, String eventName, Object eventPayload) {
@@ -239,10 +239,8 @@ public void raiseEvent(String instanceId, String eventName, Object eventPayload)
       builder.setInput(StringValue.of(serializedPayload));
     }
 
-    this.observe("dapr.workflow.grpc.raiseEvent", (TraceContext context) -> {
-      OrchestratorService.RaiseEventRequest request = builder.build();
-      this.sidecarClient.raiseEvent(request);
-    });
+    OrchestratorService.RaiseEventRequest request = builder.build();
+    this.sidecarClient.raiseEvent(request);
 
   }
 

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

@@ -14,14 +14,16 @@
 package io.dapr.durabletask;
 
 import com.google.protobuf.StringValue;
+import io.dapr.durabletask.activity.ActivityRunner;
 import io.dapr.durabletask.implementation.protobuf.OrchestratorService;
-import io.dapr.durabletask.implementation.protobuf.OrchestratorService.TaskFailureDetails;
 import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc;
 import io.grpc.Channel;
 import io.grpc.ManagedChannel;
 import io.grpc.ManagedChannelBuilder;
 import io.grpc.Status;
 import io.grpc.StatusRuntimeException;
+import io.opentelemetry.api.GlobalOpenTelemetry;
+import io.opentelemetry.api.trace.Tracer;
 import io.opentelemetry.api.trace.propagation.W3CTraceContextPropagator;
 import io.opentelemetry.context.Context;
 import io.opentelemetry.context.propagation.TextMapGetter;
@@ -54,6 +56,7 @@ public final class DurableTaskGrpcWorker implements AutoCloseable {
   private final Duration maximumTimerInterval;
   private final ExecutorService workerPool;
   private final String appId; // App ID for cross-app routing
+  private final Tracer tracer;
 
   private final TaskHubSidecarServiceGrpc.TaskHubSidecarServiceBlockingStub sidecarClient;
   private final boolean isExecutorServiceManaged;
@@ -85,6 +88,8 @@ public final class DurableTaskGrpcWorker implements AutoCloseable {
       sidecarGrpcChannel = this.managedSidecarChannel;
     }
 
+    this.tracer = GlobalOpenTelemetry.getTracer("dapr-workflow");
+
     this.sidecarClient = TaskHubSidecarServiceGrpc.newBlockingStub(sidecarGrpcChannel);
     this.dataConverter = builder.dataConverter != null ? builder.dataConverter : new JacksonDataConverter();
     this.maximumTimerInterval = builder.maximumTimerInterval != null ? builder.maximumTimerInterval
@@ -95,6 +100,8 @@ public final class DurableTaskGrpcWorker implements AutoCloseable {
     this.workerPool = Context.taskWrapping(rawExecutor);
 
     this.isExecutorServiceManaged = builder.executorService == null;
+
+
   }
 
   /**
@@ -217,59 +224,8 @@ public void startAndBlock() {
                     activityRequest.getOrchestrationInstance().getInstanceId(),
                     Context.current()));
 
-            // Extract trace context from the ActivityRequest and set it as current
-            Context traceContext = extractTraceContext(activityRequest);
-
-            // TODO: Error handling
-            this.workerPool.submit(() -> {
-              String output = null;
-              TaskFailureDetails failureDetails = null;
-              try {
-                output = taskActivityExecutor.execute(
-                    activityRequest.getName(),
-                    activityRequest.getInput().getValue(),
-                    activityRequest.getTaskExecutionId(),
-                    activityRequest.getTaskId(),
-                    activityRequest.getParentTraceContext().getTraceParent());
-              } catch (Throwable e) {
-                failureDetails = TaskFailureDetails.newBuilder()
-                    .setErrorType(e.getClass().getName())
-                    .setErrorMessage(e.getMessage())
-                    .setStackTrace(StringValue.of(FailureDetails.getFullStackTrace(e)))
-                    .build();
-              }
-
-              OrchestratorService.ActivityResponse.Builder responseBuilder = OrchestratorService.ActivityResponse
-                  .newBuilder()
-                  .setInstanceId(activityRequest.getOrchestrationInstance().getInstanceId())
-                  .setTaskId(activityRequest.getTaskId())
-                  .setCompletionToken(workItem.getCompletionToken());
-
-              if (output != null) {
-                responseBuilder.setResult(StringValue.of(output));
-              }
-
-              if (failureDetails != null) {
-                responseBuilder.setFailureDetails(failureDetails);
-              }
+            this.workerPool.submit(new ActivityRunner(workItem, taskActivityExecutor, sidecarClient, tracer));
 
-              try {
-                this.sidecarClient.completeActivityTask(responseBuilder.build());
-              } catch (StatusRuntimeException e) {
-                if (e.getStatus().getCode() == Status.Code.UNAVAILABLE) {
-                  logger.log(Level.WARNING,
-                      "The sidecar at address {0} is unavailable while completing the activity task.",
-                      this.getSidecarAddress());
-                } else if (e.getStatus().getCode() == Status.Code.CANCELLED) {
-                  logger.log(Level.WARNING,
-                      "Durable Task worker has disconnected from {0} while completing the activity task.",
-                      this.getSidecarAddress());
-                } else {
-                  logger.log(Level.WARNING, "Unexpected failure completing the activity task at {0}.",
-                      this.getSidecarAddress());
-                }
-              }
-            });
           } else if (requestType == OrchestratorService.WorkItem.RequestCase.HEALTHPING) {
             // No-op
           } else {

durabletask-client/src/main/java/io/dapr/durabletask/FailureDetails.java

@@ -124,7 +124,13 @@ public boolean isCausedBy(Class<? extends Exception> exceptionClass) {
     }
   }
 
-  static String getFullStackTrace(Throwable e) {
+  /**
+   * Gets the full stack trace of the specified exception.
+   *
+   * @param e the exception
+   * @return the full stack trace of the exception
+   */
+  public static String getFullStackTrace(Throwable e) {
     StackTraceElement[] elements = e.getStackTrace();
 
     // Plan for 256 characters per stack frame (which is likely on the high-end)

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

@@ -16,11 +16,18 @@
 import java.util.HashMap;
 import java.util.logging.Logger;
 
-final class TaskActivityExecutor {
+public final class TaskActivityExecutor {
   private final HashMap<String, TaskActivityFactory> activityFactories;
   private final DataConverter dataConverter;
   private final Logger logger;
 
+  /**
+   * Constructor.
+   *
+   * @param activityFactories the activity factories to use for creating activities
+   * @param dataConverter     the data converter to use for serializing and deserializing activity inputs and outputs
+   * @param logger            the logger to use for logging
+   */
   public TaskActivityExecutor(
       HashMap<String, TaskActivityFactory> activityFactories,
       DataConverter dataConverter,
@@ -30,6 +37,17 @@ public TaskActivityExecutor(
     this.logger = logger;
   }
 
+  /**
+   * Executes an activity task.
+   *
+   * @param taskName        the name of the activity task to execute
+   * @param input           the serialized input payload for the activity task
+   * @param taskExecutionId Unique ID for the task execution.
+   * @param taskId          Auto-incrementing ID for the task.
+   * @param traceParent     The traceparent header value.
+   * @return the serialized output payload for the activity task, or null if the activity task returned null.
+   * @throws Throwable if an unhandled exception occurs during activity task execution.
+   */
   public String execute(String taskName, String input,
                         String taskExecutionId, int taskId, String traceParent) throws Throwable {
     TaskActivityFactory factory = this.activityFactories.get(taskName);