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,28 +201,12 @@ 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));
 
     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();
-    }
-  }
-
   @Override
   public void raiseEvent(String instanceId, String eventName, Object eventPayload) {
     Helpers.throwIfArgumentNull(instanceId, "instanceId");
@@ -239,10 +220,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

@@ -13,15 +13,17 @@
 
 package io.dapr.durabletask;
 
-import com.google.protobuf.StringValue;
 import io.dapr.durabletask.implementation.protobuf.OrchestratorService;
-import io.dapr.durabletask.implementation.protobuf.OrchestratorService.TaskFailureDetails;
 import io.dapr.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc;
+import io.dapr.durabletask.runner.ActivityRunner;
+import io.dapr.durabletask.runner.OrchestratorRunner;
 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;
+
+
   }
 
   /**
@@ -167,109 +174,25 @@ public void startAndBlock() {
         while (workItemStream.hasNext()) {
           OrchestratorService.WorkItem workItem = workItemStream.next();
           OrchestratorService.WorkItem.RequestCase requestType = workItem.getRequestCase();
+
           if (requestType == OrchestratorService.WorkItem.RequestCase.ORCHESTRATORREQUEST) {
             OrchestratorService.OrchestratorRequest orchestratorRequest = workItem.getOrchestratorRequest();
-            logger.log(Level.FINEST,
+            logger.log(Level.INFO,
                 String.format("Processing orchestrator request for instance: {0}",
                     orchestratorRequest.getInstanceId()));
 
-            // TODO: Error handling
-            this.workerPool.submit(() -> {
-              TaskOrchestratorResult taskOrchestratorResult = taskOrchestrationExecutor.execute(
-                  orchestratorRequest.getPastEventsList(),
-                  orchestratorRequest.getNewEventsList());
-
-              OrchestratorService.OrchestratorResponse response = OrchestratorService.OrchestratorResponse.newBuilder()
-                  .setInstanceId(orchestratorRequest.getInstanceId())
-                  .addAllActions(taskOrchestratorResult.getActions())
-                  .setCustomStatus(StringValue.of(taskOrchestratorResult.getCustomStatus()))
-                  .setCompletionToken(workItem.getCompletionToken())
-                  .build();
-
-              try {
-                this.sidecarClient.completeOrchestratorTask(response);
-                logger.log(Level.FINEST,
-                    "Completed orchestrator request for instance: {0}",
-                    orchestratorRequest.getInstanceId());
-              } catch (StatusRuntimeException e) {
-                if (e.getStatus().getCode() == Status.Code.UNAVAILABLE) {
-                  logger.log(Level.WARNING,
-                      "The sidecar at address {0} is unavailable while completing the orchestrator 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 orchestrator task.",
-                      this.getSidecarAddress());
-                } else {
-                  logger.log(Level.WARNING,
-                      "Unexpected failure completing the orchestrator task at {0}.",
-                      this.getSidecarAddress());
-                }
-              }
-            });
+            this.workerPool.submit(new OrchestratorRunner(workItem, taskOrchestrationExecutor, sidecarClient, tracer));
           } else if (requestType == OrchestratorService.WorkItem.RequestCase.ACTIVITYREQUEST) {
             OrchestratorService.ActivityRequest activityRequest = workItem.getActivityRequest();
 
-
             logger.log(Level.INFO,
                 String.format("Processing activity request: %s for instance: %s, gRPC thread context: %s",
                     activityRequest.getName(),
                     activityRequest.getOrchestrationInstance().getInstanceId(),
                     Context.current()));
 
-            // Extract trace context from the ActivityRequest and set it as current
-            Context traceContext = extractTraceContext(activityRequest);
+            this.workerPool.submit(new ActivityRunner(workItem, taskActivityExecutor, sidecarClient, tracer));
 
-            // 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);
-              }
-
-              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);

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

@@ -44,7 +44,7 @@
 import java.util.function.IntFunction;
 import java.util.logging.Logger;
 
-final class TaskOrchestrationExecutor {
+public final class TaskOrchestrationExecutor {
 
   private static final String EMPTY_STRING = "";
   private final HashMap<String, TaskOrchestrationFactory> orchestrationFactories;
@@ -53,6 +53,15 @@ final class TaskOrchestrationExecutor {
   private final Duration maximumTimerInterval;
   private final String appId;
 
+  /**
+   * Creates a new TaskOrchestrationExecutor.
+   *
+   * @param orchestrationFactories map of orchestration names to their factories
+   * @param dataConverter          converter for serializing/deserializing data
+   * @param maximumTimerInterval   maximum duration for timer intervals
+   * @param logger                 logger for orchestration execution
+   * @param appId                  application ID for cross-app routing
+   */
   public TaskOrchestrationExecutor(
       HashMap<String, TaskOrchestrationFactory> orchestrationFactories,
       DataConverter dataConverter,
@@ -66,6 +75,13 @@ public TaskOrchestrationExecutor(
     this.appId = appId; // extracted from router
   }
 
+  /**
+   * Executes the orchestration with the given past and new events.
+   *
+   * @param pastEvents list of past history events
+   * @param newEvents  list of new history events
+   * @return the result of the orchestrator execution
+   */
   public TaskOrchestratorResult execute(List<OrchestratorService.HistoryEvent> pastEvents,
                                         List<OrchestratorService.HistoryEvent> newEvents) {
     ContextImplTask context = new ContextImplTask(pastEvents, newEvents);

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

@@ -18,12 +18,18 @@
 import java.util.Collection;
 import java.util.Collections;
 
-final class TaskOrchestratorResult {
+public final class TaskOrchestratorResult {
 
   private final Collection<OrchestratorService.OrchestratorAction> actions;
 
   private final String customStatus;
 
+  /**
+   * Creates a new TaskOrchestratorResult.
+   *
+   * @param actions      the collection of orchestrator actions
+   * @param customStatus the custom status of the orchestrator
+   */
   public TaskOrchestratorResult(Collection<OrchestratorService.OrchestratorAction> actions, String customStatus) {
     this.actions = Collections.unmodifiableCollection(actions);
     ;