Skip to content

Add entity tracing support #296

New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

Sign up for GitHub

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

Open
Varshitha Bachu (bachuv) wants to merge 6 commits into
base: main
Choose a base branch
from
Open

Add entity tracing support #296

Show file tree
Hide file tree
Changes from all commits
Commits
Show all changes
6 commits
Select commit Hold shift + click to select a range
3459cb4
initial commit
bachuv Aug 7, 2026
e1d6430
address copilot comment
bachuv Aug 12, 2026
5d9c093
wire entity producer spans and trace-context propagation
bachuv Aug 14, 2026
ac4f9ac
Set requestTime on entity actions and edit comments
bachuv Aug 14, 2026
8ac0fc6
use span.end(endTime)
bachuv Aug 14, 2026
27f6ea2
Potential fix for pull request finding 'Useless parameter'
bachuv Aug 14, 2026
File filter

Filter by extension

Filter by extension
Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
10 changes: 7 additions & 3 deletions client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorker.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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);
Expand Down Expand Up @@ -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;
Expand Down Expand Up @@ -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) {
Expand Down Expand Up @@ -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()) {
Expand Down
17 changes: 17 additions & 0 deletions client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorkerBuilder.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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<>();
Expand Down Expand Up @@ -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
Expand Down
5 changes: 4 additions & 1 deletion client/src/main/java/com/microsoft/durabletask/EntityRunner.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -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();
Expand Down
85 changes: 80 additions & 5 deletions client/src/main/java/com/microsoft/durabletask/TaskEntityExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -6,6 +6,8 @@
import com.google.protobuf.Timestamp;
import com.microsoft.durabletask.implementation.protobuf.OrchestratorService.*;

import io.opentelemetry.api.trace.Span;

import javax.annotation.Nonnull;
import javax.annotation.Nullable;
import java.time.Instant;
Expand All @@ -24,14 +26,17 @@ final class TaskEntityExecutor {
private final HashMap<String, TaskEntityFactory> entityFactories;
private final DataConverter dataConverter;
private final Logger logger;
private final boolean emitTraceSpans;

TaskEntityExecutor(
HashMap<String, TaskEntityFactory> entityFactories,
DataConverter dataConverter,
Logger logger) {
Logger logger,
boolean emitTraceSpans) {
this.entityFactories = entityFactories;
this.dataConverter = dataConverter;
this.logger = logger;
this.emitTraceSpans = emitTraceSpans;
}

/**
Expand Down Expand Up @@ -80,7 +85,7 @@ EntityBatchResult execute(@Nonnull EntityBatchRequest request) {
TaskEntityState entityState = new TaskEntityState(this.dataConverter, initialState);

// Create the concrete context that collects actions
TaskEntityContextImpl context = new TaskEntityContextImpl(entityId, this.dataConverter);
TaskEntityContextImpl context = new TaskEntityContextImpl(entityId, this.dataConverter, this.emitTraceSpans);

// Process each operation
List<OperationResult> results = new ArrayList<>();
Expand Down Expand Up @@ -127,6 +132,23 @@ EntityBatchResult execute(@Nonnull EntityBatchRequest request) {

Instant startTime = Instant.now();

// Entity processing span parented on the incoming operation trace context;
// suppressed when the host emits its own (emitTraceSpans is false).
Span processingSpan = this.emitTraceSpans

@YunchuWang wangbill (YunchuWang) Aug 18, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Should address — make the processing span current while user entity code runs. The span is created and its context is copied manually to Durable actions, but entity.run(operation) is never executed under processingSpan.makeCurrent(). As a result, spans created by user code or automatic HTTP/database instrumentation inside the entity will not be children of this entity processing span and will appear disconnected.

Please mirror the existing activity execution pattern: open a Scope from the processing span around entity.run(...), close it on every success/failure path, and then end the span. Add a test that creates a normal nested span inside an entity operation and verifies its parent is the processing span.

? TracingHelper.startEntityProcessingSpan(
entityName,
operationName,
false,
instanceId,
opRequest.hasTraceContext() ? opRequest.getTraceContext() : null)
: null;

// Signals/orchestrations this operation produces nest under the processing span (or the
// raw incoming context when spans are suppressed), so the host can link them downstream.
context.setCurrentOperationTraceContext(processingSpan != null
? TracingHelper.getCurrentTraceContext(processingSpan)
: (opRequest.hasTraceContext() ? opRequest.getTraceContext() : null));

try {
// Build the operation
TaskEntityOperation operation = new TaskEntityOperation(
Expand Down Expand Up @@ -157,6 +179,8 @@ EntityBatchResult execute(@Nonnull EntityBatchRequest request) {
entityState.commit();
context.commit();

TracingHelper.endEntityProcessingSpan(processingSpan, null);

logger.log(Level.FINE, "Operation '{0}' on entity '{1}' completed successfully.",
new Object[]{operationName, instanceId});

Expand Down Expand Up @@ -188,6 +212,8 @@ EntityBatchResult execute(@Nonnull EntityBatchRequest request) {
// Rollback state and actions on failure
entityState.rollback();
context.rollback();

TracingHelper.endEntityProcessingSpan(processingSpan, e.getMessage());
}
}

Expand Down Expand Up @@ -219,12 +245,20 @@ private static Timestamp toTimestamp(Instant instant) {
private static class TaskEntityContextImpl extends TaskEntityContext {
private final EntityInstanceId entityId;
private final DataConverter dataConverter;
private final boolean emitTraceSpans;
private final List<PendingAction> pendingActions = new ArrayList<>();
private int committedActionCount = 0;
@Nullable
private TraceContext currentOperationTraceContext;

TaskEntityContextImpl(EntityInstanceId entityId, DataConverter dataConverter) {
TaskEntityContextImpl(EntityInstanceId entityId, DataConverter dataConverter, boolean emitTraceSpans) {
this.entityId = entityId;
this.dataConverter = dataConverter;
this.emitTraceSpans = emitTraceSpans;
}

void setCurrentOperationTraceContext(@Nullable TraceContext traceContext) {
this.currentOperationTraceContext = traceContext;
}

@Nonnull
Expand All @@ -244,7 +278,8 @@ public void signalEntity(

SendSignalAction.Builder signalBuilder = SendSignalAction.newBuilder()
.setInstanceId(targetEntityId.toString())
.setName(operationName);
.setName(operationName)
.setRequestTime(toTimestamp(Instant.now()));

if (input != null) {
String serializedInput = this.dataConverter.serialize(input);
Expand All @@ -261,6 +296,26 @@ public void signalEntity(
.build());
}

if (this.currentOperationTraceContext != null) {
Comment thread
bachuv marked this conversation as resolved.

@YunchuWang wangbill (YunchuWang) Aug 18, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Must address — propagate the PRODUCER span context, not its parent processing context. The action is stamped with currentOperationTraceContext before the PRODUCER span is created, and the new PRODUCER span's context is discarded. A downstream CONSUMER/orchestration span will therefore be parented directly to the processing span, making it a sibling of the PRODUCER instead of its child. The .NET dispatcher starts the producer activity and puts that activity's context on the outgoing message.

After creating the committed PRODUCER span, set the action's parentTraceContext from that span's context (for both signal and start-orchestration actions). Add an assertion that the action's propagated span ID equals the exported PRODUCER span ID, not the enclosing SERVER span ID.

signalBuilder.setParentTraceContext(this.currentOperationTraceContext);
}

if (this.emitTraceSpans && this.currentOperationTraceContext != null) {
String signalScheduledTime = (options != null && options.getScheduledTime() != null)
? options.getScheduledTime().toString() : null;
Span producerSpan = TracingHelper.startEntitySignalProducerSpan(
targetEntityId.getName(),
operationName,
targetEntityId.toString(),
this.entityId.toString(),
this.currentOperationTraceContext,
null,
signalScheduledTime);
if (producerSpan != null) {

@YunchuWang wangbill (YunchuWang) Aug 18, 2026

Copy link
Copy Markdown
Member

Choose a reason for hiding this comment

The reason will be displayed to describe this comment to others. Learn more.

Must address — do not export spans for actions that are later rolled back. This PRODUCER span is ended immediately inside signalEntity() (and similarly in startNewOrchestration()), before the surrounding entity operation commits. If the entity signals and then throws, context.rollback() correctly removes the pending action, but this already-finished span remains exported and appears to show work that was never sent.

Please defer emission until the action is known to be committed, and add a regression test where an entity queues an action and then throws: the result should contain zero actions and export zero corresponding PRODUCER spans. Keep the existing success assertion to prove committed actions still emit exactly once.

producerSpan.end();
}
}

this.pendingActions.add(new PendingAction(PendingAction.Type.SEND_SIGNAL, signalBuilder.build(), null));
}

Expand All @@ -278,7 +333,8 @@ public String startNewOrchestration(

StartNewOrchestrationAction.Builder orchBuilder = StartNewOrchestrationAction.newBuilder()
.setInstanceId(instanceId)
.setName(name);
.setName(name)
.setRequestTime(toTimestamp(Instant.now()));

if (input != null) {
String serializedInput = this.dataConverter.serialize(input);
Expand All @@ -300,6 +356,25 @@ public String startNewOrchestration(
}
}

if (this.currentOperationTraceContext != null) {
orchBuilder.setParentTraceContext(this.currentOperationTraceContext);
}

if (this.emitTraceSpans && this.currentOperationTraceContext != null) {
String orchScheduledTime = (options != null && options.getStartTime() != null)
? options.getStartTime().toString() : null;
Span producerSpan = TracingHelper.startEntityStartOrchestrationSpan(
this.entityId.getName(),
this.entityId.toString(),
instanceId,
this.currentOperationTraceContext,
null,
orchScheduledTime);
if (producerSpan != null) {
producerSpan.end();
}
}

this.pendingActions.add(new PendingAction(
PendingAction.Type.START_NEW_ORCHESTRATION, null, orchBuilder.build()));

Expand Down
59 changes: 59 additions & 0 deletions client/src/main/java/com/microsoft/durabletask/TaskOrchestrationExecutor.java
View file
Open in desktop
Original file line number Diff line number Diff line change
Expand Up @@ -14,6 +14,8 @@
import com.microsoft.durabletask.implementation.protobuf.OrchestratorService.ScheduleTaskAction.Builder;
import com.microsoft.durabletask.util.UUIDGenerator;

import io.opentelemetry.api.trace.Span;

import javax.annotation.Nullable;
import java.time.Duration;
import java.time.Instant;
Expand All @@ -39,6 +41,7 @@ final class TaskOrchestrationExecutor {
private final DurableTaskGrpcWorkerVersioningOptions versioningOptions;
private final ExceptionPropertiesProvider exceptionPropertiesProvider;
private final boolean useNativeEntityActions;
private final boolean emitTraceSpans;

public TaskOrchestrationExecutor(
HashMap<String, TaskOrchestrationFactory> orchestrationFactories,
Expand Down Expand Up @@ -67,13 +70,27 @@ public TaskOrchestrationExecutor(
DurableTaskGrpcWorkerVersioningOptions versioningOptions,
boolean useNativeEntityActions,
ExceptionPropertiesProvider exceptionPropertiesProvider) {
this(orchestrationFactories, dataConverter, maximumTimerInterval, logger, versioningOptions,
useNativeEntityActions, exceptionPropertiesProvider, false);
}

public TaskOrchestrationExecutor(
HashMap<String, TaskOrchestrationFactory> orchestrationFactories,
DataConverter dataConverter,
Duration maximumTimerInterval,
Logger logger,
DurableTaskGrpcWorkerVersioningOptions versioningOptions,
boolean useNativeEntityActions,
ExceptionPropertiesProvider exceptionPropertiesProvider,
boolean emitTraceSpans) {
this.orchestrationFactories = orchestrationFactories;
this.dataConverter = dataConverter;
this.maximumTimerInterval = maximumTimerInterval;
this.logger = logger;
this.versioningOptions = versioningOptions;
this.useNativeEntityActions = useNativeEntityActions;
this.exceptionPropertiesProvider = exceptionPropertiesProvider;
this.emitTraceSpans = emitTraceSpans;
}

public TaskOrchestratorResult execute(
Expand Down Expand Up @@ -446,6 +463,24 @@ public UUID newUUID() {

// region Entity integration methods (Phase 4)

// Writes the orchestration trace context into the legacy DTFx RequestMessage JSON as
// parentTraceContext (DistributedTraceContext, PascalCase members) so the Azure Functions
// host can link its entity spans. The orchestration context is deterministic (from history).
private void addLegacyEntityParentTraceContext(ObjectNode requestMessage) {
TraceContext propagatedCtx = this.orchestrationSpanContext != null
? this.orchestrationSpanContext : this.parentTraceContext;
if (propagatedCtx == null || propagatedCtx.getTraceParent() == null
|| propagatedCtx.getTraceParent().isEmpty()) {
return;
}
ObjectNode ptc = requestMessage.putObject("parentTraceContext");
ptc.put("TraceParent", propagatedCtx.getTraceParent());
if (propagatedCtx.hasTraceState() && propagatedCtx.getTraceState().getValue() != null
&& !propagatedCtx.getTraceState().getValue().isEmpty()) {
ptc.put("TraceState", propagatedCtx.getTraceState().getValue());
}
}

@Override
public void signalEntity(EntityInstanceId entityId, String operationName, Object input, SignalEntityOptions options) {
Helpers.throwIfOrchestratorComplete(this.isComplete);
Expand Down Expand Up @@ -490,6 +525,7 @@ public void signalEntity(EntityInstanceId entityId, String operationName, Object
requestMessage.put("due", scheduledTimeStr);
eventName = "op@" + scheduledTimeStr;
}
this.addLegacyEntityParentTraceContext(requestMessage);
this.pendingActions.put(id, OrchestratorAction.newBuilder()
.setId(id)
.setSendEvent(SendEventAction.newBuilder()
Expand All @@ -500,6 +536,28 @@ public void signalEntity(EntityInstanceId entityId, String operationName, Object
.build());
}

// PRODUCER span for the signal so standalone/DTS workers record the client side.
// Suppressed under Azure Functions, where the host emits it.
if (TaskOrchestrationExecutor.this.emitTraceSpans && !this.isReplaying) {
TraceContext signalParentCtx = this.orchestrationSpanContext != null
? this.orchestrationSpanContext : this.parentTraceContext;
if (signalParentCtx != null) {
String signalScheduledTime = (options != null && options.getScheduledTime() != null)
? options.getScheduledTime().toString() : null;
Span signalSpan = TracingHelper.startEntitySignalProducerSpan(
entityId.getName(),
operationName,
entityId.toString(),
this.instanceId,
signalParentCtx,
null,
signalScheduledTime);
if (signalSpan != null) {
signalSpan.end();
}
}
}

if (!this.isReplaying) {
this.logger.fine(() -> String.format(
"%s: signaling entity '%s' operation '%s' (#%d)",
Expand Down Expand Up @@ -567,6 +625,7 @@ public <V> Task<V> callEntity(EntityInstanceId entityId, String operationName, O
if (this.executionId != null) {
requestMessage.put("parentExecution", this.executionId);
}
this.addLegacyEntityParentTraceContext(requestMessage);
this.pendingActions.put(id, OrchestratorAction.newBuilder()
.setId(id)
.setSendEvent(SendEventAction.newBuilder()
Expand Down
Loading
Loading