Add entity tracing support

client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorker.java

@@ -59,6 +59,7 @@ public final class DurableTaskGrpcWorker implements AutoCloseable {
     private final boolean supportsLargePayloads;
     private final int maxChunkSizeBytes;
     private final int largePayloadThresholdBytes;
+    private final boolean emitTraceSpans;
 
     DurableTaskGrpcWorker(DurableTaskGrpcWorkerBuilder builder, WorkItemFilter workItemFilter) {
         this.orchestrationFactories.putAll(builder.orchestrationFactories);
@@ -95,6 +96,7 @@ public final class DurableTaskGrpcWorker implements AutoCloseable {
         this.supportsLargePayloads = builder.supportsLargePayloads;
         this.maxChunkSizeBytes = builder.maxChunkSizeBytes;
         this.largePayloadThresholdBytes = builder.largePayloadThresholdBytes;
+        this.emitTraceSpans = builder.emitTraceSpans;
         this.dataConverter = builder.dataConverter != null ? builder.dataConverter : new JacksonDataConverter();
         this.maximumTimerInterval = builder.maximumTimerInterval != null ? builder.maximumTimerInterval : DEFAULT_MAXIMUM_TIMER_INTERVAL;
         this.versioningOptions = builder.versioningOptions;
@@ -175,15 +177,17 @@ public void startAndBlock() {
                 logger,
                 this.versioningOptions,
                 true,
-                this.exceptionPropertiesProvider);
+                this.exceptionPropertiesProvider,
+                this.emitTraceSpans);
         TaskActivityExecutor taskActivityExecutor = new TaskActivityExecutor(
                 this.activityFactories,
                 this.dataConverter,
                 logger);
         TaskEntityExecutor taskEntityExecutor = new TaskEntityExecutor(
                 this.entityFactories,
                 this.dataConverter,
-                logger);
+                logger,
+                this.emitTraceSpans);
 
         // TODO: How do we interrupt manually?
         while (true) {
@@ -450,7 +454,7 @@ public void startAndBlock() {
                         EntityRequest entityRequestV2 = workItem.getEntityRequestV2();
                         this.workItemExecutor.submit(() -> {
                             try {
-                                // Convert V2 (history-based) format to V1 (flat) format
+                                // Convert V2 (history-based) format to V1 (flat) format.
                                 EntityBatchRequest.Builder batchBuilder = EntityBatchRequest.newBuilder()
                                         .setInstanceId(entityRequestV2.getInstanceId());
                                 if (entityRequestV2.hasEntityState()) {

client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorkerBuilder.java

@@ -28,6 +28,7 @@ public final class DurableTaskGrpcWorkerBuilder {
     ExceptionPropertiesProvider exceptionPropertiesProvider;
     int maxConcurrentEntityWorkItems = 1;
     int maxWorkItemThreads;
+    boolean emitTraceSpans = true;
     private WorkItemFilter workItemFilter;
     private boolean autoGenerateWorkItemFilters;
     final List<ClientInterceptor> interceptors = new ArrayList<>();
@@ -452,6 +453,22 @@ public DurableTaskGrpcWorkerBuilder setMaxChunkSizeBytes(int maxChunkSizeBytes)
         return this;
     }
 
+    /**
+     * Sets whether this worker emits its own OpenTelemetry spans for orchestrations, activities, and
+     * entities. Defaults to {@code true}.
+     * <p>
+     * Set to {@code false} when running under a host that already emits Durable Task spans (for
+     * example, the Azure Functions Durable extension, which emits {@code DurableTask.Core} spans), to
+     * avoid a duplicate worker-side span layer. Trace-context propagation is unaffected either way.
+     *
+     * @param emitTraceSpans whether the worker emits its own spans
+     * @return this builder object
+     */
+    public DurableTaskGrpcWorkerBuilder setEmitTraceSpans(boolean emitTraceSpans) {
+        this.emitTraceSpans = emitTraceSpans;
+        return this;
+    }
+
     /**
      * Initializes a new {@link DurableTaskGrpcWorker} object with the settings specified in the current builder object.
      * @return a new {@link DurableTaskGrpcWorker} object

client/src/main/java/com/microsoft/durabletask/EntityRunner.java

@@ -90,7 +90,10 @@ public static byte[] loadAndRun(byte[] entityRequestBytes, TaskEntityFactory ent
         TaskEntityExecutor executor = new TaskEntityExecutor(
                 factories,
                 new JacksonDataConverter(),
-                logger);
+                logger,
+                // EntityRunner is the Azure Functions entry point; the Durable extension host already
+                // emits DurableTask.Core entity spans, so the worker suppresses its own to avoid duplicates.
+                false);
 
         EntityBatchResult result = executor.execute(request);
         return result.toByteArray();

client/src/main/java/com/microsoft/durabletask/GrpcDurableEntityClient.java

@@ -7,6 +7,9 @@
 import com.microsoft.durabletask.implementation.protobuf.OrchestratorService.*;
 import com.microsoft.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc.TaskHubSidecarServiceBlockingStub;
 
+import io.opentelemetry.api.trace.Span;
+import io.opentelemetry.context.Scope;
+
 import javax.annotation.Nullable;
 import java.time.Instant;
 import java.util.ArrayList;
@@ -39,10 +42,12 @@ public void signalEntity(
         Helpers.throwIfArgumentNull(entityId, "entityId");
         Helpers.throwIfArgumentNull(operationName, "operationName");
 
+        Instant requestTime = Instant.now();
         SignalEntityRequest.Builder builder = SignalEntityRequest.newBuilder()
                 .setInstanceId(entityId.toString())
                 .setName(operationName)
-                .setRequestId(UUID.randomUUID().toString());
+            .setRequestId(UUID.randomUUID().toString())
+            .setRequestTime(DataConverter.getTimestampFromInstant(requestTime));
 
         if (input != null) {
             String serializedInput = this.dataConverter.serialize(input);
@@ -58,11 +63,34 @@ public void signalEntity(
 
         // Capture and propagate distributed trace context (matching .NET SDK pattern)
         TraceContext traceContext = TracingHelper.getCurrentTraceContext();
-        if (traceContext != null) {
-            builder.setParentTraceContext(traceContext);
+        String scheduledTime = options != null && options.getScheduledTime() != null
+                ? options.getScheduledTime().toString() : null;
+        Span producerSpan = TracingHelper.startEntitySignalProducerSpan(
+                entityId.getName(),
+                operationName,
+                entityId.toString(),
+                null,
+                traceContext,
+                requestTime,
+                scheduledTime);
+        TraceContext producerTraceContext = TracingHelper.getCurrentTraceContext(producerSpan);
+        TraceContext propagatedTraceContext = producerTraceContext != null
+                ? producerTraceContext : traceContext;
+        if (propagatedTraceContext != null) {
+            builder.setParentTraceContext(propagatedTraceContext);
         }
 
-        this.sidecarClient.signalEntity(builder.build());
+        Scope producerScope = producerTraceContext != null ? producerSpan.makeCurrent() : null;
+        try {
+            this.sidecarClient.signalEntity(builder.build());
+        } finally {
+            if (producerScope != null) {
+                producerScope.close();
+            }
+            if (producerSpan != null) {
+                producerSpan.end();
+            }
+        }
     }
 
     @Override