diff --git a/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorker.java b/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorker.java index 086b1cf..0bd158c 100644 --- a/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorker.java +++ b/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,7 +177,8 @@ public void startAndBlock() { logger, this.versioningOptions, true, - this.exceptionPropertiesProvider); + this.exceptionPropertiesProvider, + this.emitTraceSpans); TaskActivityExecutor taskActivityExecutor = new TaskActivityExecutor( this.activityFactories, this.dataConverter, @@ -183,7 +186,8 @@ public void startAndBlock() { 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()) { diff --git a/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorkerBuilder.java b/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorkerBuilder.java index 3df0a28..9a6ea80 100644 --- a/client/src/main/java/com/microsoft/durabletask/DurableTaskGrpcWorkerBuilder.java +++ b/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 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}. + *

+ * 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 diff --git a/client/src/main/java/com/microsoft/durabletask/EntityRunner.java b/client/src/main/java/com/microsoft/durabletask/EntityRunner.java index 680cced..1630e01 100644 --- a/client/src/main/java/com/microsoft/durabletask/EntityRunner.java +++ b/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(); diff --git a/client/src/main/java/com/microsoft/durabletask/GrpcDurableEntityClient.java b/client/src/main/java/com/microsoft/durabletask/GrpcDurableEntityClient.java index 343e385..ece6dfd 100644 --- a/client/src/main/java/com/microsoft/durabletask/GrpcDurableEntityClient.java +++ b/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 diff --git a/client/src/main/java/com/microsoft/durabletask/TaskEntityExecutor.java b/client/src/main/java/com/microsoft/durabletask/TaskEntityExecutor.java index e2352da..f22e383 100644 --- a/client/src/main/java/com/microsoft/durabletask/TaskEntityExecutor.java +++ b/client/src/main/java/com/microsoft/durabletask/TaskEntityExecutor.java @@ -6,6 +6,9 @@ import com.google.protobuf.Timestamp; import com.microsoft.durabletask.implementation.protobuf.OrchestratorService.*; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.context.Scope; + import javax.annotation.Nonnull; import javax.annotation.Nullable; import java.time.Instant; @@ -24,14 +27,17 @@ final class TaskEntityExecutor { private final HashMap entityFactories; private final DataConverter dataConverter; private final Logger logger; + private final boolean emitTraceSpans; TaskEntityExecutor( HashMap entityFactories, DataConverter dataConverter, - Logger logger) { + Logger logger, + boolean emitTraceSpans) { this.entityFactories = entityFactories; this.dataConverter = dataConverter; this.logger = logger; + this.emitTraceSpans = emitTraceSpans; } /** @@ -80,7 +86,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 results = new ArrayList<>(); @@ -127,13 +133,27 @@ EntityBatchResult execute(@Nonnull EntityBatchRequest request) { Instant startTime = Instant.now(); + // The dispatcher owns the processing span and propagates its context with the operation. + // Child spans and actions created by user code use that context without re-emitting it. + TraceContext operationTraceContext = opRequest.hasTraceContext() + ? opRequest.getTraceContext() : null; + context.setCurrentOperationTraceContext(operationTraceContext); + try { // Build the operation TaskEntityOperation operation = new TaskEntityOperation( operationName, serializedInput, context, entityState, this.dataConverter); // Execute - Object result = entity.run(operation); + Object result; + Scope processingScope = TracingHelper.makeTraceContextCurrent(operationTraceContext); + try { + result = entity.run(operation); + } finally { + if (processingScope != null) { + processingScope.close(); + } + } Instant endTime = Instant.now(); @@ -219,12 +239,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 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 @@ -242,9 +270,11 @@ public void signalEntity( Objects.requireNonNull(targetEntityId, "targetEntityId must not be null"); Objects.requireNonNull(operationName, "operationName must not be null"); + Instant requestTime = Instant.now(); SendSignalAction.Builder signalBuilder = SendSignalAction.newBuilder() .setInstanceId(targetEntityId.toString()) - .setName(operationName); + .setName(operationName) + .setRequestTime(toTimestamp(requestTime)); if (input != null) { String serializedInput = this.dataConverter.serialize(input); @@ -261,7 +291,21 @@ public void signalEntity( .build()); } - this.pendingActions.add(new PendingAction(PendingAction.Type.SEND_SIGNAL, signalBuilder.build(), null)); + if (this.currentOperationTraceContext != null) { + signalBuilder.setParentTraceContext(this.currentOperationTraceContext); + } + + String signalScheduledTime = (options != null && options.getScheduledTime() != null) + ? options.getScheduledTime().toString() : null; + this.pendingActions.add(PendingAction.forSignal( + signalBuilder, + targetEntityId.getName(), + operationName, + targetEntityId.toString(), + this.entityId.toString(), + this.currentOperationTraceContext, + requestTime, + signalScheduledTime)); } @Nonnull @@ -276,9 +320,11 @@ public String startNewOrchestration( ? options.getInstanceId() : UUID.randomUUID().toString(); + Instant requestTime = Instant.now(); StartNewOrchestrationAction.Builder orchBuilder = StartNewOrchestrationAction.newBuilder() .setInstanceId(instanceId) - .setName(name); + .setName(name) + .setRequestTime(toTimestamp(requestTime)); if (input != null) { String serializedInput = this.dataConverter.serialize(input); @@ -300,8 +346,20 @@ public String startNewOrchestration( } } - this.pendingActions.add(new PendingAction( - PendingAction.Type.START_NEW_ORCHESTRATION, null, orchBuilder.build())); + if (this.currentOperationTraceContext != null) { + orchBuilder.setParentTraceContext(this.currentOperationTraceContext); + } + + String orchScheduledTime = (options != null && options.getStartTime() != null) + ? options.getStartTime().toString() : null; + this.pendingActions.add(PendingAction.forOrchestration( + orchBuilder, + this.entityId.getName(), + this.entityId.toString(), + instanceId, + this.currentOperationTraceContext, + requestTime, + orchScheduledTime)); return instanceId; } @@ -310,6 +368,9 @@ public String startNewOrchestration( * Marks the current set of pending actions as committed (snapshot for rollback). */ void commit() { + for (int i = this.committedActionCount; i < this.pendingActions.size(); i++) { + this.pendingActions.get(i).commit(this.emitTraceSpans); + } this.committedActionCount = this.pendingActions.size(); } @@ -336,9 +397,9 @@ List getCommittedActions(int startId) { OperationAction.Builder actionBuilder = OperationAction.newBuilder() .setId(id++); if (pending.type == PendingAction.Type.SEND_SIGNAL) { - actionBuilder.setSendSignal(pending.sendSignal); + actionBuilder.setSendSignal(pending.sendSignal.build()); } else { - actionBuilder.setStartNewOrchestration(pending.startNewOrchestration); + actionBuilder.setStartNewOrchestration(pending.startNewOrchestration.build()); } actions.add(actionBuilder.build()); } @@ -352,13 +413,125 @@ private static class PendingAction { enum Type { SEND_SIGNAL, START_NEW_ORCHESTRATION } final Type type; - final SendSignalAction sendSignal; - final StartNewOrchestrationAction startNewOrchestration; - - PendingAction(Type type, SendSignalAction sendSignal, StartNewOrchestrationAction startNewOrchestration) { + final SendSignalAction.Builder sendSignal; + final StartNewOrchestrationAction.Builder startNewOrchestration; + final String sourceEntityName; + final String sourceEntityInstanceId; + final String targetName; + final String operationName; + final String targetInstanceId; + final TraceContext parentTraceContext; + final Instant requestTime; + final String scheduledTime; + + private PendingAction( + Type type, + SendSignalAction.Builder sendSignal, + StartNewOrchestrationAction.Builder startNewOrchestration, + String sourceEntityName, + String sourceEntityInstanceId, + String targetName, + String operationName, + String targetInstanceId, + TraceContext parentTraceContext, + Instant requestTime, + String scheduledTime) { this.type = type; this.sendSignal = sendSignal; this.startNewOrchestration = startNewOrchestration; + this.sourceEntityName = sourceEntityName; + this.sourceEntityInstanceId = sourceEntityInstanceId; + this.targetName = targetName; + this.operationName = operationName; + this.targetInstanceId = targetInstanceId; + this.parentTraceContext = parentTraceContext; + this.requestTime = requestTime; + this.scheduledTime = scheduledTime; + } + + static PendingAction forSignal( + SendSignalAction.Builder signal, + String targetEntityName, + String operationName, + String targetEntityInstanceId, + String sourceEntityInstanceId, + TraceContext parentTraceContext, + Instant requestTime, + String scheduledTime) { + return new PendingAction( + Type.SEND_SIGNAL, + signal, + null, + null, + sourceEntityInstanceId, + targetEntityName, + operationName, + targetEntityInstanceId, + parentTraceContext, + requestTime, + scheduledTime); + } + + static PendingAction forOrchestration( + StartNewOrchestrationAction.Builder orchestration, + String sourceEntityName, + String sourceEntityInstanceId, + String targetOrchestrationInstanceId, + TraceContext parentTraceContext, + Instant requestTime, + String scheduledTime) { + return new PendingAction( + Type.START_NEW_ORCHESTRATION, + null, + orchestration, + sourceEntityName, + sourceEntityInstanceId, + null, + null, + targetOrchestrationInstanceId, + parentTraceContext, + requestTime, + scheduledTime); + } + + void commit(boolean emitTraceSpans) { + if (!emitTraceSpans || this.parentTraceContext == null) { + return; + } + + Span producerSpan; + if (this.type == Type.SEND_SIGNAL) { + producerSpan = TracingHelper.startEntitySignalProducerSpan( + this.targetName, + this.operationName, + this.targetInstanceId, + this.sourceEntityInstanceId, + this.parentTraceContext, + this.requestTime, + this.scheduledTime); + } else { + producerSpan = TracingHelper.startEntityStartOrchestrationSpan( + this.sourceEntityName, + this.sourceEntityInstanceId, + this.targetInstanceId, + this.parentTraceContext, + this.requestTime, + this.scheduledTime); + } + + if (producerSpan == null) { + return; + } + + TraceContext producerTraceContext = TracingHelper.getCurrentTraceContext(producerSpan); + if (producerTraceContext != null) { + if (this.type == Type.SEND_SIGNAL) { + this.sendSignal.setParentTraceContext(producerTraceContext); + } else { + this.startNewOrchestration.setParentTraceContext(producerTraceContext); + } + } + producerSpan.end(); } } } diff --git a/client/src/main/java/com/microsoft/durabletask/TaskOrchestrationExecutor.java b/client/src/main/java/com/microsoft/durabletask/TaskOrchestrationExecutor.java index 5bdbad2..82c7b62 100644 --- a/client/src/main/java/com/microsoft/durabletask/TaskOrchestrationExecutor.java +++ b/client/src/main/java/com/microsoft/durabletask/TaskOrchestrationExecutor.java @@ -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; @@ -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 orchestrationFactories, @@ -67,6 +70,19 @@ public TaskOrchestrationExecutor( DurableTaskGrpcWorkerVersioningOptions versioningOptions, boolean useNativeEntityActions, ExceptionPropertiesProvider exceptionPropertiesProvider) { + this(orchestrationFactories, dataConverter, maximumTimerInterval, logger, versioningOptions, + useNativeEntityActions, exceptionPropertiesProvider, false); + } + + public TaskOrchestrationExecutor( + HashMap orchestrationFactories, + DataConverter dataConverter, + Duration maximumTimerInterval, + Logger logger, + DurableTaskGrpcWorkerVersioningOptions versioningOptions, + boolean useNativeEntityActions, + ExceptionPropertiesProvider exceptionPropertiesProvider, + boolean emitTraceSpans) { this.orchestrationFactories = orchestrationFactories; this.dataConverter = dataConverter; this.maximumTimerInterval = maximumTimerInterval; @@ -74,6 +90,7 @@ public TaskOrchestrationExecutor( this.versioningOptions = versioningOptions; this.useNativeEntityActions = useNativeEntityActions; this.exceptionPropertiesProvider = exceptionPropertiesProvider; + this.emitTraceSpans = emitTraceSpans; } public TaskOrchestratorResult execute( @@ -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); @@ -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() @@ -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)", @@ -567,6 +625,7 @@ public Task 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() diff --git a/client/src/main/java/com/microsoft/durabletask/TracingHelper.java b/client/src/main/java/com/microsoft/durabletask/TracingHelper.java index 79f9dcd..9f77410 100644 --- a/client/src/main/java/com/microsoft/durabletask/TracingHelper.java +++ b/client/src/main/java/com/microsoft/durabletask/TracingHelper.java @@ -16,6 +16,7 @@ import io.opentelemetry.api.trace.Tracer; import io.opentelemetry.api.GlobalOpenTelemetry; import io.opentelemetry.context.Context; +import io.opentelemetry.context.Scope; import javax.annotation.Nullable; import java.lang.reflect.Field; @@ -76,6 +77,13 @@ final class TracingHelper { static final String ATTR_FIRE_AT = "durabletask.fire_at"; static final String ATTR_EVENT_TARGET_INSTANCE_ID = "durabletask.event.target_instance_id"; + // Entity span constants matching .NET SDK schema. Entity spans deliberately do NOT set + // durabletask.task.name/version/task_id; the entity name already appears in the span name. + static final String TYPE_ENTITY = "entity"; + static final String OP_SIGNAL_ENTITY = "signal_entity"; + static final String ATTR_OPERATION = "durabletask.task.operation"; + static final String ATTR_SCHEDULED_TIME = "durabletask.task.scheduled_time"; + private TracingHelper() { // Static utility class } @@ -233,6 +241,13 @@ static Context extractTraceContext(@Nullable TraceContext protoCtx) { return Context.current().with(Span.wrap(remoteContext)); } + /** Makes a propagated trace context current until the returned scope is closed. */ + @Nullable + static Scope makeTraceContextCurrent(@Nullable TraceContext traceContext) { + Context context = extractTraceContext(traceContext); + return context != null ? context.makeCurrent() : null; + } + /** * Starts a new span as a child of the given trace context. * @@ -476,4 +491,89 @@ static void emitEventSpan( spanBuilder.startSpan().end(); } + + // region Entity spans + + /** Builds an entity span name: {@code entity::}. */ + static String createEntitySpanName(String entityName, String operation) { + return TYPE_ENTITY + ":" + entityName + ":" + operation; + } + + /** Builds an entity-starts-orchestration span name: {@code :create_orchestration}. */ + static String createEntityStartOrchestrationSpanName(String entityName) { + return entityName + ":" + TYPE_CREATE_ORCHESTRATION; + } + + /** + * Starts a {@link SpanKind#PRODUCER} span for signaling an entity. Used by orchestration signals, + * entity-to-entity signals, and the external client signal path. Returns {@code null} when the + * parent context is absent or invalid. Short-lived callers end the span immediately; the client + * path ends it in a {@code finally} block after the gRPC call. + */ + @Nullable + static Span startEntitySignalProducerSpan( + String targetEntityName, + String operation, + String targetEntityInstanceId, + @Nullable String sourceEntityInstanceId, + @Nullable TraceContext parentContext, + @Nullable java.time.Instant startTime, + @Nullable String scheduledTime) { + Context parentCtx = extractTraceContext(parentContext); + if (parentCtx == null) { + return null; + } + Tracer tracer = GlobalOpenTelemetry.getTracer(TRACER_NAME); + SpanBuilder spanBuilder = tracer.spanBuilder(createEntitySpanName(targetEntityName, operation)) + .setSpanKind(SpanKind.PRODUCER) + .setParent(parentCtx) + .setAttribute(ATTR_TYPE, TYPE_ENTITY) + .setAttribute(ATTR_OPERATION, OP_SIGNAL_ENTITY) + .setAttribute(ATTR_EVENT_TARGET_INSTANCE_ID, targetEntityInstanceId); + if (sourceEntityInstanceId != null) { + spanBuilder.setAttribute(ATTR_INSTANCE_ID, sourceEntityInstanceId); + } + if (scheduledTime != null) { + spanBuilder.setAttribute(ATTR_SCHEDULED_TIME, scheduledTime); + } + if (startTime != null) { + spanBuilder.setStartTimestamp(startTime); + } + return spanBuilder.startSpan(); + } + + /** + * Starts a {@link SpanKind#PRODUCER} span for an entity starting an orchestration. The span name + * is {@code :create_orchestration}. Returns {@code null} when the parent context is + * absent or invalid. + */ + @Nullable + static Span startEntityStartOrchestrationSpan( + String sourceEntityName, + String sourceEntityInstanceId, + String targetOrchestrationInstanceId, + @Nullable TraceContext parentContext, + @Nullable java.time.Instant startTime, + @Nullable String scheduledTime) { + Context parentCtx = extractTraceContext(parentContext); + if (parentCtx == null) { + return null; + } + Tracer tracer = GlobalOpenTelemetry.getTracer(TRACER_NAME); + SpanBuilder spanBuilder = tracer.spanBuilder(createEntityStartOrchestrationSpanName(sourceEntityName)) + .setSpanKind(SpanKind.PRODUCER) + .setParent(parentCtx) + .setAttribute(ATTR_TYPE, TYPE_ENTITY) + .setAttribute(ATTR_EVENT_TARGET_INSTANCE_ID, targetOrchestrationInstanceId) + .setAttribute(ATTR_INSTANCE_ID, sourceEntityInstanceId); + if (scheduledTime != null) { + spanBuilder.setAttribute(ATTR_SCHEDULED_TIME, scheduledTime); + } + if (startTime != null) { + spanBuilder.setStartTimestamp(startTime); + } + return spanBuilder.startSpan(); + } + + // endregion } diff --git a/client/src/test/java/com/microsoft/durabletask/GrpcDurableEntityClientTracingTest.java b/client/src/test/java/com/microsoft/durabletask/GrpcDurableEntityClientTracingTest.java new file mode 100644 index 0000000..7ec68e7 --- /dev/null +++ b/client/src/test/java/com/microsoft/durabletask/GrpcDurableEntityClientTracingTest.java @@ -0,0 +1,114 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. +package com.microsoft.durabletask; + +import com.microsoft.durabletask.implementation.protobuf.OrchestratorService.SignalEntityRequest; +import com.microsoft.durabletask.implementation.protobuf.OrchestratorService.SignalEntityResponse; +import com.microsoft.durabletask.implementation.protobuf.TaskHubSidecarServiceGrpc; + +import io.grpc.ManagedChannel; +import io.grpc.Server; +import io.grpc.inprocess.InProcessChannelBuilder; +import io.grpc.inprocess.InProcessServerBuilder; +import io.grpc.stub.StreamObserver; +import io.opentelemetry.api.GlobalOpenTelemetry; +import io.opentelemetry.api.trace.Span; +import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.context.Scope; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter; +import io.opentelemetry.sdk.trace.SdkTracerProvider; +import io.opentelemetry.sdk.trace.data.SpanData; +import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.concurrent.atomic.AtomicReference; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +public class GrpcDurableEntityClientTracingTest { + + private final AtomicReference capturedRequest = new AtomicReference<>(); + private InMemorySpanExporter spanExporter; + private OpenTelemetrySdk openTelemetry; + private Server inProcessServer; + private ManagedChannel inProcessChannel; + private DurableTaskClient client; + + @BeforeEach + void setUp() throws Exception { + GlobalOpenTelemetry.resetForTest(); + this.spanExporter = InMemorySpanExporter.create(); + SdkTracerProvider tracerProvider = SdkTracerProvider.builder() + .addSpanProcessor(SimpleSpanProcessor.create(this.spanExporter)) + .build(); + this.openTelemetry = OpenTelemetrySdk.builder() + .setTracerProvider(tracerProvider) + .buildAndRegisterGlobal(); + + String serverName = InProcessServerBuilder.generateName(); + this.inProcessServer = InProcessServerBuilder.forName(serverName) + .directExecutor() + .addService(new TaskHubSidecarServiceGrpc.TaskHubSidecarServiceImplBase() { + @Override + public void signalEntity( + SignalEntityRequest request, + StreamObserver responseObserver) { + capturedRequest.set(request); + responseObserver.onNext(SignalEntityResponse.getDefaultInstance()); + responseObserver.onCompleted(); + } + }) + .build() + .start(); + this.inProcessChannel = InProcessChannelBuilder.forName(serverName).directExecutor().build(); + this.client = new DurableTaskGrpcClientBuilder().grpcChannel(this.inProcessChannel).build(); + } + + @AfterEach + void tearDown() { + if (this.inProcessChannel != null) { + this.inProcessChannel.shutdownNow(); + } + if (this.inProcessServer != null) { + this.inProcessServer.shutdownNow(); + } + if (this.openTelemetry != null) { + this.openTelemetry.close(); + } + GlobalOpenTelemetry.resetForTest(); + } + + @Test + void signalEntity_emitsProducerSpanAndPropagatesItsContext() { + Span parentSpan = GlobalOpenTelemetry.getTracer("test").spanBuilder("parent").startSpan(); + try (Scope ignored = parentSpan.makeCurrent()) { + this.client.getEntities().signalEntity( + new EntityInstanceId("Counter", "c1"), + "add", + 5); + } finally { + parentSpan.end(); + } + + SignalEntityRequest request = this.capturedRequest.get(); + assertNotNull(request); + assertTrue(request.hasRequestTime()); + assertTrue(request.getRequestTime().getSeconds() > 0); + + SpanData producer = this.spanExporter.getFinishedSpanItems().stream() + .filter(span -> span.getKind() == SpanKind.PRODUCER) + .findFirst() + .orElse(null); + assertNotNull(producer, "expected external entity signal PRODUCER span"); + assertEquals("entity:counter:add", producer.getName()); + assertEquals(parentSpan.getSpanContext().getSpanId(), producer.getParentSpanId()); + assertEquals(producer.getSpanId(), + request.getParentTraceContext().getTraceParent().split("-")[2]); + } +} \ No newline at end of file diff --git a/client/src/test/java/com/microsoft/durabletask/TaskEntityExecutorTest.java b/client/src/test/java/com/microsoft/durabletask/TaskEntityExecutorTest.java index 7b678df..df3d137 100644 --- a/client/src/test/java/com/microsoft/durabletask/TaskEntityExecutorTest.java +++ b/client/src/test/java/com/microsoft/durabletask/TaskEntityExecutorTest.java @@ -116,7 +116,7 @@ public Object run(TaskEntityOperation operation) throws Exception { private TaskEntityExecutor createExecutor(String entityName, TaskEntityFactory factory) { HashMap factories = new HashMap<>(); factories.put(entityName.toLowerCase(java.util.Locale.ROOT), factory); - return new TaskEntityExecutor(factories, dataConverter, logger); + return new TaskEntityExecutor(factories, dataConverter, logger, true); } private OperationRequest buildOperationRequest(String operationName, Object input, String requestId) { @@ -358,6 +358,9 @@ void execute_entitySignalsOther_actionsIncluded() { SendSignalAction signalAction = result.getActions(0).getSendSignal(); assertEquals("@counter@target1", signalAction.getInstanceId()); assertEquals("add", signalAction.getName()); + assertTrue(signalAction.hasRequestTime(), "SendSignalAction should set requestTime"); + assertTrue(signalAction.getRequestTime().getSeconds() > 0, + "requestTime should be a real timestamp, not the unset Unix epoch"); } @Test @@ -378,6 +381,72 @@ void execute_entityStartsOrchestration_actionsIncluded() { StartNewOrchestrationAction orchAction = result.getActions(0).getStartNewOrchestration(); assertEquals("MyOrchestration", orchAction.getName()); + assertTrue(orchAction.hasRequestTime(), "StartNewOrchestrationAction should set requestTime"); + assertTrue(orchAction.getRequestTime().getSeconds() > 0, + "requestTime should be a real timestamp, not the unset Unix epoch"); + } + + @Test + void execute_entitySignalsOther_propagatesOperationTraceContext() { + TaskEntityExecutor executor = createExecutor("Signaler", SignalingEntity::new); + + TraceContext opTraceContext = TraceContext.newBuilder() + .setTraceParent("00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01") + .setTraceState(StringValue.of("rojo=00f067aa0ba902b7")) + .build(); + OperationRequest op = OperationRequest.newBuilder() + .setOperation("signalOther") + .setRequestId("req-signalOther") + .setTraceContext(opTraceContext) + .build(); + + EntityBatchResult result = executor.execute(buildBatchRequest("Signaler", "s1", null, op)); + + assertEquals(1, result.getActionsCount()); + assertTrue(result.getActions(0).hasSendSignal()); + SendSignalAction signalAction = result.getActions(0).getSendSignal(); + assertTrue(signalAction.hasParentTraceContext(), + "SendSignalAction should carry the operation's trace context"); + assertEquals("00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01", + signalAction.getParentTraceContext().getTraceParent()); + assertEquals("rojo=00f067aa0ba902b7", signalAction.getParentTraceContext().getTraceState().getValue()); + } + + @Test + void execute_entityStartsOrchestration_propagatesOperationTraceContext() { + TaskEntityExecutor executor = createExecutor("OrchStarter", OrchestrationStartingEntity::new); + + TraceContext opTraceContext = TraceContext.newBuilder() + .setTraceParent("00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01") + .build(); + OperationRequest op = OperationRequest.newBuilder() + .setOperation("startOrch") + .setRequestId("req-startOrch") + .setTraceContext(opTraceContext) + .build(); + + EntityBatchResult result = executor.execute(buildBatchRequest("OrchStarter", "o1", null, op)); + + assertEquals(1, result.getActionsCount()); + assertTrue(result.getActions(0).hasStartNewOrchestration()); + StartNewOrchestrationAction orchAction = result.getActions(0).getStartNewOrchestration(); + assertTrue(orchAction.hasParentTraceContext(), + "StartNewOrchestrationAction should carry the operation's trace context"); + assertEquals("00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01", + orchAction.getParentTraceContext().getTraceParent()); + } + + @Test + void execute_entitySignalsOther_noTraceContext_noParentTraceContext() { + TaskEntityExecutor executor = createExecutor("Signaler", SignalingEntity::new); + + EntityBatchResult result = executor.execute( + buildBatchRequest("Signaler", "s1", null, buildOperationRequest("signalOther"))); + + assertEquals(1, result.getActionsCount()); + assertTrue(result.getActions(0).hasSendSignal()); + assertFalse(result.getActions(0).getSendSignal().hasParentTraceContext(), + "SendSignalAction should not carry a trace context when the operation had none"); } @Test diff --git a/client/src/test/java/com/microsoft/durabletask/TaskEntityExecutorTracingTest.java b/client/src/test/java/com/microsoft/durabletask/TaskEntityExecutorTracingTest.java new file mode 100644 index 0000000..505417b --- /dev/null +++ b/client/src/test/java/com/microsoft/durabletask/TaskEntityExecutorTracingTest.java @@ -0,0 +1,223 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. +package com.microsoft.durabletask; + +import com.google.protobuf.StringValue; +import com.microsoft.durabletask.implementation.protobuf.OrchestratorService.EntityBatchRequest; +import com.microsoft.durabletask.implementation.protobuf.OrchestratorService.EntityBatchResult; +import com.microsoft.durabletask.implementation.protobuf.OrchestratorService.OperationRequest; +import com.microsoft.durabletask.implementation.protobuf.OrchestratorService.TraceContext; + +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter; +import io.opentelemetry.sdk.trace.SdkTracerProvider; +import io.opentelemetry.sdk.trace.data.SpanData; +import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.util.HashMap; +import java.util.List; +import java.util.logging.Logger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Verifies entity action spans and propagation from the dispatcher-owned processing context. + */ +public class TaskEntityExecutorTracingTest { + + private static final Logger logger = Logger.getLogger(TaskEntityExecutorTracingTest.class.getName()); + private static final DataConverter dataConverter = new JacksonDataConverter(); + private static final String TRACE_ID = "0af7651916cd43dd8448eb211c80319c"; + private static final String PARENT_SPAN_ID = "b7ad6b7169203331"; + + private InMemorySpanExporter spanExporter; + private OpenTelemetrySdk openTelemetry; + + @BeforeEach + void setUp() { + io.opentelemetry.api.GlobalOpenTelemetry.resetForTest(); + spanExporter = InMemorySpanExporter.create(); + SdkTracerProvider tracerProvider = SdkTracerProvider.builder() + .addSpanProcessor(SimpleSpanProcessor.create(spanExporter)) + .build(); + openTelemetry = OpenTelemetrySdk.builder() + .setTracerProvider(tracerProvider) + .buildAndRegisterGlobal(); + } + + @AfterEach + void tearDown() { + openTelemetry.close(); + io.opentelemetry.api.GlobalOpenTelemetry.resetForTest(); + } + + /** Minimal counter entity used for the batch under test. */ + static class CounterEntity extends AbstractTaskEntity { + public void add(int amount) { + this.state += amount; + } + + public void signalOther(int amount) { + this.context.signalEntity(new EntityInstanceId("counter", "c2"), "add", amount); + } + + public void signalThenThrow(int amount) { + this.context.signalEntity(new EntityInstanceId("counter", "c2"), "add", amount); + throw new IllegalStateException("operation failed"); + } + + public void createNestedSpan(int amount) { + io.opentelemetry.api.GlobalOpenTelemetry.getTracer("test") + .spanBuilder("entity-user-code") + .setAttribute("amount", amount) + .startSpan() + .end(); + } + + public void startOrch() { + this.context.startNewOrchestration("DownstreamOrch", null); + } + + @Override + protected Integer initializeState(TaskEntityOperation operation) { + return 0; + } + + @Override + protected Class getStateType() { + return Integer.class; + } + } + + private TaskEntityExecutor createExecutor(boolean emitTraceSpans) { + HashMap factories = new HashMap<>(); + factories.put("counter", CounterEntity::new); + return new TaskEntityExecutor(factories, dataConverter, logger, emitTraceSpans); + } + + private EntityBatchRequest requestWith(@javax.annotation.Nullable TraceContext traceContext) { + return requestWithOp("add", 5, traceContext); + } + + private EntityBatchRequest requestWithOp( + String operation, int input, @javax.annotation.Nullable TraceContext traceContext) { + OperationRequest.Builder op = OperationRequest.newBuilder() + .setOperation(operation) + .setRequestId("req-1") + .setInput(StringValue.of(dataConverter.serialize(input))); + if (traceContext != null) { + op.setTraceContext(traceContext); + } + return EntityBatchRequest.newBuilder() + .setInstanceId("@counter@c1") + .setEntityState(StringValue.of(dataConverter.serialize(10))) + .addOperations(op.build()) + .build(); + } + + private static TraceContext parentTraceContext() { + return TraceContext.newBuilder() + .setTraceParent("00-" + TRACE_ID + "-" + PARENT_SPAN_ID + "-01") + .build(); + } + + @Test + void execute_doesNotDuplicateDispatcherProcessingSpan() { + TaskEntityExecutor executor = createExecutor(true); + + EntityBatchResult result = executor.execute(requestWith(parentTraceContext())); + assertTrue(result.getResults(0).hasSuccess()); + assertTrue(spanExporter.getFinishedSpanItems().isEmpty()); + } + + @Test + void execute_emitTraceSpansDisabled_suppressesSpan() { + TaskEntityExecutor executor = createExecutor(false); + + EntityBatchResult result = executor.execute(requestWith(parentTraceContext())); + assertTrue(result.getResults(0).hasSuccess()); + + assertTrue(spanExporter.getFinishedSpanItems().isEmpty()); + } + + @Test + void execute_noParentTraceContext_emitsNoSpan() { + TaskEntityExecutor executor = createExecutor(true); + + EntityBatchResult result = executor.execute(requestWith(null)); + assertTrue(result.getResults(0).hasSuccess()); + + assertTrue(spanExporter.getFinishedSpanItems().isEmpty()); + } + + @Test + void execute_entitySignalsEntity_emitsProducerSpanUnderProcessingContext() { + TaskEntityExecutor executor = createExecutor(true); + + EntityBatchResult result = executor.execute(requestWithOp("signalOther", 3, parentTraceContext())); + assertTrue(result.getResults(0).hasSuccess()); + + List spans = spanExporter.getFinishedSpanItems(); + SpanData producer = spans.stream().filter(s -> s.getKind() == SpanKind.PRODUCER).findFirst().orElse(null); + assertNotNull(producer, "expected PRODUCER signal span"); + assertEquals("entity:counter:add", producer.getName()); + assertEquals("signal_entity", + producer.getAttributes().get(AttributeKey.stringKey("durabletask.task.operation"))); + assertEquals(TRACE_ID, producer.getTraceId()); + assertEquals(PARENT_SPAN_ID, producer.getParentSpanId()); + assertEquals(producer.getSpanId(), result.getActions(0).getSendSignal() + .getParentTraceContext().getTraceParent().split("-")[2]); + } + + @Test + void execute_entityStartsOrchestration_emitsProducerSpanUnderProcessingContext() { + TaskEntityExecutor executor = createExecutor(true); + + EntityBatchResult result = executor.execute(requestWithOp("startOrch", 0, parentTraceContext())); + assertTrue(result.getResults(0).hasSuccess()); + + List spans = spanExporter.getFinishedSpanItems(); + SpanData producer = spans.stream().filter(s -> s.getKind() == SpanKind.PRODUCER).findFirst().orElse(null); + assertNotNull(producer, "expected PRODUCER create_orchestration span"); + assertEquals("counter:create_orchestration", producer.getName()); + assertEquals("entity", producer.getAttributes().get(AttributeKey.stringKey("durabletask.type"))); + assertEquals(TRACE_ID, producer.getTraceId()); + assertEquals(PARENT_SPAN_ID, producer.getParentSpanId()); + assertEquals(producer.getSpanId(), result.getActions(0).getStartNewOrchestration() + .getParentTraceContext().getTraceParent().split("-")[2]); + } + + @Test + void execute_makesProcessingSpanCurrentForUserCode() { + TaskEntityExecutor executor = createExecutor(true); + + EntityBatchResult result = executor.execute(requestWithOp("createNestedSpan", 3, parentTraceContext())); + + assertTrue(result.getResults(0).hasSuccess()); + List spans = spanExporter.getFinishedSpanItems(); + SpanData nested = spans.stream().filter(s -> s.getName().equals("entity-user-code")).findFirst().orElse(null); + assertNotNull(nested, "expected span created by entity user code"); + assertEquals(TRACE_ID, nested.getTraceId()); + assertEquals(PARENT_SPAN_ID, nested.getParentSpanId()); + } + + @Test + void execute_entityActionRollsBack_emitsNoProducerSpan() { + TaskEntityExecutor executor = createExecutor(true); + + EntityBatchResult result = executor.execute(requestWithOp("signalThenThrow", 3, parentTraceContext())); + + assertTrue(result.getResults(0).hasFailure()); + assertEquals(0, result.getActionsCount()); + assertTrue(spanExporter.getFinishedSpanItems().stream() + .noneMatch(span -> span.getKind() == SpanKind.PRODUCER)); + } +} diff --git a/client/src/test/java/com/microsoft/durabletask/TaskOrchestrationEntityEventTest.java b/client/src/test/java/com/microsoft/durabletask/TaskOrchestrationEntityEventTest.java index b07ac43..a83033e 100644 --- a/client/src/test/java/com/microsoft/durabletask/TaskOrchestrationEntityEventTest.java +++ b/client/src/test/java/com/microsoft/durabletask/TaskOrchestrationEntityEventTest.java @@ -309,6 +309,39 @@ private boolean hasLockRequestAction(Collection actions) thr // region signalEntity tests + @Test + void signalEntity_legacyPath_propagatesParentTraceContextInJson() throws Exception { + final String orchestratorName = "SignalEntityTraceOrchestration"; + EntityInstanceId entityId = new EntityInstanceId("Counter", "c1"); + + TaskOrchestrationExecutor executor = createExecutor(orchestratorName, ctx -> { + ctx.signalEntity(entityId, "add", 5); + ctx.complete("done"); + }); + + String traceParent = "00-0af7651916cd43dd8448eb211c80319c-b7ad6b7169203331-01"; + TraceContext orchCtx = TraceContext.newBuilder().setTraceParent(traceParent).build(); + + List pastEvents = Arrays.asList( + orchestratorStarted(), + executionStarted(orchestratorName, "null")); + List newEvents = Collections.singletonList(orchestratorCompleted()); + + TaskOrchestratorResult result = executor.execute(pastEvents, newEvents, orchCtx); + + boolean found = false; + for (OrchestratorAction action : result.getActions()) { + if (action.hasSendEvent() + && action.getSendEvent().getInstance().getInstanceId().contains("@counter@c1")) { + JsonNode json = JSON_MAPPER.readTree(action.getSendEvent().getData().getValue()); + assertTrue(json.has("parentTraceContext"), "expected parentTraceContext in signal JSON"); + assertEquals(traceParent, json.get("parentTraceContext").get("TraceParent").asText()); + found = true; + } + } + assertTrue(found, "expected a SendEvent signal action carrying parentTraceContext"); + } + @Test void signalEntity_producesSendEventAction() throws Exception { final String orchestratorName = "SignalEntityOrchestration"; diff --git a/client/src/test/java/com/microsoft/durabletask/TaskOrchestrationEntityTracingTest.java b/client/src/test/java/com/microsoft/durabletask/TaskOrchestrationEntityTracingTest.java new file mode 100644 index 0000000..2f31fdb --- /dev/null +++ b/client/src/test/java/com/microsoft/durabletask/TaskOrchestrationEntityTracingTest.java @@ -0,0 +1,166 @@ +// Copyright (c) Microsoft Corporation. All rights reserved. +// Licensed under the MIT License. +package com.microsoft.durabletask; + +import com.google.protobuf.StringValue; +import com.google.protobuf.Timestamp; +import com.microsoft.durabletask.implementation.protobuf.OrchestratorService.ExecutionStartedEvent; +import com.microsoft.durabletask.implementation.protobuf.OrchestratorService.HistoryEvent; +import com.microsoft.durabletask.implementation.protobuf.OrchestratorService.OrchestrationInstance; +import com.microsoft.durabletask.implementation.protobuf.OrchestratorService.OrchestratorStartedEvent; +import com.microsoft.durabletask.implementation.protobuf.OrchestratorService.TraceContext; + +import io.opentelemetry.api.common.AttributeKey; +import io.opentelemetry.api.trace.SpanKind; +import io.opentelemetry.sdk.OpenTelemetrySdk; +import io.opentelemetry.sdk.testing.exporter.InMemorySpanExporter; +import io.opentelemetry.sdk.trace.SdkTracerProvider; +import io.opentelemetry.sdk.trace.data.SpanData; +import io.opentelemetry.sdk.trace.export.SimpleSpanProcessor; + +import org.junit.jupiter.api.AfterEach; +import org.junit.jupiter.api.BeforeEach; +import org.junit.jupiter.api.Test; + +import java.time.Duration; +import java.util.Arrays; +import java.util.Collections; +import java.util.HashMap; +import java.util.List; +import java.util.logging.Logger; + +import static org.junit.jupiter.api.Assertions.assertEquals; +import static org.junit.jupiter.api.Assertions.assertNotNull; +import static org.junit.jupiter.api.Assertions.assertTrue; + +/** + * Verifies that {@link TaskOrchestrationExecutor} emits the orchestration-initiated entity signal + * PRODUCER span when {@code emitTraceSpans} is enabled (standalone/DTS worker) and suppresses it + * otherwise (Azure Functions, where the host emits the entity spans). + */ +public class TaskOrchestrationEntityTracingTest { + + private static final Logger logger = Logger.getLogger(TaskOrchestrationEntityTracingTest.class.getName()); + private static final String TRACE_ID = "0af7651916cd43dd8448eb211c80319c"; + private static final String ORCH_SPAN_ID = "b7ad6b7169203331"; + + private InMemorySpanExporter spanExporter; + private OpenTelemetrySdk openTelemetry; + + @BeforeEach + void setUp() { + io.opentelemetry.api.GlobalOpenTelemetry.resetForTest(); + spanExporter = InMemorySpanExporter.create(); + SdkTracerProvider tracerProvider = SdkTracerProvider.builder() + .addSpanProcessor(SimpleSpanProcessor.create(spanExporter)) + .build(); + openTelemetry = OpenTelemetrySdk.builder() + .setTracerProvider(tracerProvider) + .buildAndRegisterGlobal(); + } + + @AfterEach + void tearDown() { + openTelemetry.close(); + io.opentelemetry.api.GlobalOpenTelemetry.resetForTest(); + } + + private TaskOrchestrationExecutor createExecutor( + String orchestratorName, TaskOrchestration orchestration, boolean emitTraceSpans) { + HashMap factories = new HashMap<>(); + factories.put(orchestratorName, new TaskOrchestrationFactory() { + @Override + public String getName() { + return orchestratorName; + } + + @Override + public TaskOrchestration create() { + return orchestration; + } + }); + return new TaskOrchestrationExecutor( + factories, + new JacksonDataConverter(), + Duration.ofDays(1), + logger, + null, + true, + null, + emitTraceSpans); + } + + private HistoryEvent orchestratorStarted() { + return HistoryEvent.newBuilder() + .setEventId(-1) + .setTimestamp(Timestamp.getDefaultInstance()) + .setOrchestratorStarted(OrchestratorStartedEvent.getDefaultInstance()) + .build(); + } + + private HistoryEvent executionStarted(String name) { + return HistoryEvent.newBuilder() + .setEventId(-1) + .setTimestamp(Timestamp.getDefaultInstance()) + .setExecutionStarted(ExecutionStartedEvent.newBuilder() + .setName(name) + .setVersion(StringValue.of("")) + .setInput(StringValue.of("null")) + .setOrchestrationInstance(OrchestrationInstance.newBuilder() + .setInstanceId("test-instance-id") + .build()) + .build()) + .build(); + } + + private static TraceContext orchestrationContext() { + return TraceContext.newBuilder() + .setTraceParent("00-" + TRACE_ID + "-" + ORCH_SPAN_ID + "-01") + .build(); + } + + @Test + void signalEntity_emitsProducerSpanUnderOrchestrationContext() { + String orchestratorName = "SignalOrch"; + EntityInstanceId entityId = new EntityInstanceId("Counter", "c1"); + TaskOrchestrationExecutor executor = createExecutor(orchestratorName, ctx -> { + ctx.signalEntity(entityId, "add", 5); + ctx.complete("done"); + }, true); + + executor.execute( + Collections.emptyList(), + Arrays.asList(orchestratorStarted(), executionStarted(orchestratorName)), + orchestrationContext()); + + List spans = spanExporter.getFinishedSpanItems(); + SpanData producer = spans.stream() + .filter(s -> s.getKind() == SpanKind.PRODUCER).findFirst().orElse(null); + assertNotNull(producer, "expected PRODUCER signal span"); + assertEquals("entity:counter:add", producer.getName()); + assertEquals("signal_entity", + producer.getAttributes().get(AttributeKey.stringKey("durabletask.task.operation"))); + assertEquals(TRACE_ID, producer.getTraceId()); + assertEquals(ORCH_SPAN_ID, producer.getParentSpanId()); + } + + @Test + void signalEntity_emitTraceSpansDisabled_suppressesProducerSpan() { + String orchestratorName = "SignalOrchDisabled"; + EntityInstanceId entityId = new EntityInstanceId("Counter", "c1"); + TaskOrchestrationExecutor executor = createExecutor(orchestratorName, ctx -> { + ctx.signalEntity(entityId, "add", 5); + ctx.complete("done"); + }, false); + + executor.execute( + Collections.emptyList(), + Arrays.asList(orchestratorStarted(), executionStarted(orchestratorName)), + orchestrationContext()); + + List spans = spanExporter.getFinishedSpanItems(); + assertTrue(spans.stream().noneMatch(s -> s.getKind() == SpanKind.PRODUCER), + "expected no PRODUCER span when emitTraceSpans is disabled"); + } + +} diff --git a/client/src/test/java/com/microsoft/durabletask/TracingHelperTest.java b/client/src/test/java/com/microsoft/durabletask/TracingHelperTest.java index c7e0a03..79aa9f7 100644 --- a/client/src/test/java/com/microsoft/durabletask/TracingHelperTest.java +++ b/client/src/test/java/com/microsoft/durabletask/TracingHelperTest.java @@ -358,4 +358,82 @@ void emitEventSpan_fromClient_createsProducerSpan() { assertEquals("event", sd.getAttributes().get(io.opentelemetry.api.common.AttributeKey.stringKey("durabletask.type"))); assertEquals("target-orch-1", sd.getAttributes().get(io.opentelemetry.api.common.AttributeKey.stringKey("durabletask.event.target_instance_id"))); } + + // region Entity spans + + private static final String TRACE_ID = "0af7651916cd43dd8448eb211c80319c"; + private static final String PARENT_SPAN_ID = "b7ad6b7169203331"; + + private static TraceContext parentCtx() { + return TraceContext.newBuilder() + .setTraceParent("00-" + TRACE_ID + "-" + PARENT_SPAN_ID + "-01") + .build(); + } + + private static String attr(SpanData sd, String key) { + return sd.getAttributes().get(io.opentelemetry.api.common.AttributeKey.stringKey(key)); + } + + @Test + void createEntitySpanName_usesEntityAndOperation() { + assertEquals("entity:Counter:add", TracingHelper.createEntitySpanName("Counter", "add")); + } + + @Test + void createEntityStartOrchestrationSpanName_isInverted() { + assertEquals("Counter:create_orchestration", + TracingHelper.createEntityStartOrchestrationSpanName("Counter")); + } + + @Test + void startEntitySignalProducerSpan_setsTargetAndSource() { + Span span = TracingHelper.startEntitySignalProducerSpan( + "Audit", "record", "@audit@a1", "@counter@c1", parentCtx(), null, null); + assertNotNull(span); + span.end(); + + SpanData sd = spanExporter.getFinishedSpanItems().get(0); + assertEquals("entity:Audit:record", sd.getName()); + assertEquals(SpanKind.PRODUCER, sd.getKind()); + assertEquals("signal_entity", attr(sd, "durabletask.task.operation")); + assertEquals("@audit@a1", attr(sd, "durabletask.event.target_instance_id")); + assertEquals("@counter@c1", attr(sd, "durabletask.task.instance_id")); + } + + @Test + void startEntitySignalProducerSpan_scheduledTime_setsAttribute() { + Span span = TracingHelper.startEntitySignalProducerSpan( + "Audit", "record", "@audit@a1", null, parentCtx(), null, "2026-01-01T00:00:00Z"); + span.end(); + + SpanData sd = spanExporter.getFinishedSpanItems().get(0); + assertEquals("2026-01-01T00:00:00Z", attr(sd, "durabletask.task.scheduled_time")); + assertNull(attr(sd, "durabletask.task.instance_id")); + } + + @Test + void startEntityStartOrchestrationSpan_setsInvertedNameAndAttributes() { + Span span = TracingHelper.startEntityStartOrchestrationSpan( + "Counter", "@counter@c1", "orch-2", parentCtx(), null, null); + assertNotNull(span); + span.end(); + + SpanData sd = spanExporter.getFinishedSpanItems().get(0); + assertEquals("Counter:create_orchestration", sd.getName()); + assertEquals(SpanKind.PRODUCER, sd.getKind()); + assertEquals("entity", attr(sd, "durabletask.type")); + assertEquals("orch-2", attr(sd, "durabletask.event.target_instance_id")); + assertEquals("@counter@c1", attr(sd, "durabletask.task.instance_id")); + } + + @Test + void entityProducerSpans_missingParent_returnNull() { + assertNull(TracingHelper.startEntitySignalProducerSpan( + "Audit", "record", "@audit@a1", null, null, null, null)); + assertNull(TracingHelper.startEntityStartOrchestrationSpan( + "Counter", "@counter@c1", "orch-2", null, null, null)); + assertTrue(spanExporter.getFinishedSpanItems().isEmpty()); + } + + // endregion }