From 3dde86f2b9333138baff7a5f3c0375822fb28584 Mon Sep 17 00:00:00 2001 From: Shyju Krishnankutty Date: Wed, 25 Feb 2026 13:15:54 -0800 Subject: [PATCH] OTEL trace support. --- .../Workflows/DurableActivityExecutor.cs | 61 ++++++++++++++++--- .../Workflows/DurableActivityInput.cs | 6 ++ .../Workflows/DurableExecutorDispatcher.cs | 3 +- .../Workflows/DurableStreamingWorkflowRun.cs | 5 +- .../Workflows/DurableWorkflowClient.cs | 34 +++++++++-- .../Workflows/DurableWorkflowInput.cs | 7 +++ .../DurableWorkflowInstrumentation.cs | 6 ++ .../Workflows/DurableWorkflowRun.cs | 54 +++++++++------- .../Workflows/DurableWorkflowRunner.cs | 22 ++----- 9 files changed, 144 insertions(+), 54 deletions(-) diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableActivityExecutor.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableActivityExecutor.cs index 526a0f00d4..efc906f04e 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableActivityExecutor.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableActivityExecutor.cs @@ -1,5 +1,6 @@ // Copyright (c) Microsoft. All rights reserved. +using System.Diagnostics; using System.Diagnostics.CodeAnalysis; using System.Text.Json; using Microsoft.Agents.AI.Workflows; @@ -40,18 +41,58 @@ internal static class DurableActivityExecutor string executorInput = inputWithState?.Input ?? input; Dictionary sharedState = inputWithState?.State ?? []; - Executor executor = await binding.FactoryAsync(binding.Id).ConfigureAwait(false); - Type inputType = ResolveInputType(inputWithState?.InputTypeName, executor.InputTypes); - object typedInput = DeserializeInput(executorInput, inputType); + // Restore the orchestrator's trace context (workflow.run span) as the parent + // for spans created in the activity worker (e.g., executor.process). + // The Durable Task SDK propagates its own trace context (from scheduling time), + // not the orchestrator's Activity.Current. We bridge this gap by explicitly + // passing the traceparent through the activity input. + Activity? parentBridge = RestoreParentTraceContext(inputWithState?.TraceParent); - DurableWorkflowContext workflowContext = new(sharedState, executor); - object? result = await executor.ExecuteAsync( - typedInput, - new TypeId(inputType), - workflowContext, - cancellationToken).ConfigureAwait(false); + try + { + Executor executor = await binding.FactoryAsync(binding.Id).ConfigureAwait(false); + Type inputType = ResolveInputType(inputWithState?.InputTypeName, executor.InputTypes); + object typedInput = DeserializeInput(executorInput, inputType); - return SerializeActivityOutput(result, workflowContext); + DurableWorkflowContext workflowContext = new(sharedState, executor); + object? result = await executor.ExecuteAsync( + typedInput, + new TypeId(inputType), + workflowContext, + cancellationToken).ConfigureAwait(false); + + return SerializeActivityOutput(result, workflowContext); + } + finally + { + parentBridge?.Dispose(); + } + } + + /// + /// Restores the orchestrator's trace context so that spans created in the activity worker + /// (like executor.process) appear as children of workflow.run in the trace hierarchy. + /// + /// A bridge activity that should be disposed when execution completes, or null if no context was provided. + private static Activity? RestoreParentTraceContext(string? traceParent) + { + if (traceParent is null) + { + return null; + } + + if (!ActivityContext.TryParse(traceParent, null, out ActivityContext parentContext)) + { + return null; + } + + // StartActivity with an explicit parent context creates a sampled span whose parent + // is workflow.run. All subsequent spans (executor.process, message.send) created while + // this is Activity.Current will nest under it in the trace. + return DurableWorkflowInstrumentation.ActivitySource.StartActivity( + "executor.dispatch", + ActivityKind.Internal, + parentContext); } private static string SerializeActivityOutput(object? result, DurableWorkflowContext context) diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableActivityInput.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableActivityInput.cs index b49306bf9e..b3cd6ca52e 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableActivityInput.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableActivityInput.cs @@ -21,4 +21,10 @@ internal sealed class DurableActivityInput /// Gets or sets the shared state dictionary (scope-prefixed key -> serialized value). /// public Dictionary State { get; set; } = []; + + /// + /// Gets or sets the W3C traceparent of the orchestrator's workflow.run span, + /// used to establish parent-child trace hierarchy from the activity worker. + /// + public string? TraceParent { get; set; } } diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableExecutorDispatcher.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableExecutorDispatcher.cs index 2b94f23c93..4682d0464a 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableExecutorDispatcher.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableExecutorDispatcher.cs @@ -68,7 +68,8 @@ internal static class DurableExecutorDispatcher { Input = input, InputTypeName = inputTypeName, - State = sharedState + State = sharedState, + TraceParent = DurableWorkflowInstrumentation.WorkflowRunTraceParent.Value }; string serializedInput = JsonSerializer.Serialize(activityInput, DurableWorkflowJsonContext.Default.DurableActivityInput); diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableStreamingWorkflowRun.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableStreamingWorkflowRun.cs index 57a44fc06b..bfe85b6b2a 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableStreamingWorkflowRun.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableStreamingWorkflowRun.cs @@ -23,6 +23,7 @@ namespace Microsoft.Agents.AI.DurableTask.Workflows; internal sealed class DurableStreamingWorkflowRun : IStreamingWorkflowRun { private readonly DurableTaskClient _client; + private readonly Activity? _runActivity; /// /// Initializes a new instance of the class. @@ -30,11 +31,13 @@ internal sealed class DurableStreamingWorkflowRun : IStreamingWorkflowRun /// The durable task client for orchestration operations. /// The unique instance ID for this orchestration run. /// The workflow being executed. - internal DurableStreamingWorkflowRun(DurableTaskClient client, string instanceId, Workflow workflow) + /// The workflow.run activity to stop when the workflow completes. + internal DurableStreamingWorkflowRun(DurableTaskClient client, string instanceId, Workflow workflow, Activity? runActivity = null) { this._client = client; this.RunId = instanceId; this.WorkflowName = workflow.Name ?? string.Empty; + this._runActivity = runActivity; } /// diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowClient.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowClient.cs index 5944d578ef..c3d6a3be19 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowClient.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowClient.cs @@ -1,5 +1,6 @@ // Copyright (c) Microsoft. All rights reserved. +using System.Diagnostics; using Microsoft.Agents.AI.Workflows; using Microsoft.DurableTask; using Microsoft.DurableTask.Client; @@ -40,7 +41,20 @@ internal sealed class DurableWorkflowClient : IWorkflowClient throw new ArgumentException("Workflow must have a valid Name property.", nameof(workflow)); } - DurableWorkflowInput workflowInput = new() { Input = input }; + // Start workflow.run at the client level so its span ID is stable across orchestrator + // replays. The Durable Task orchestrator re-executes from the top on each replay, + // creating new Activity objects that are abandoned when the method suspends. By creating + // workflow.run here and propagating its context via TraceParent, activity worker spans + // consistently reference a properly exported parent span. + Activity? runActivity = DurableWorkflowInstrumentation.ActivitySource.StartActivity("workflow.run"); + runActivity?.SetTag("workflow.id", workflow.StartExecutorId) + .SetTag("workflow.name", workflow.Name); + + DurableWorkflowInput workflowInput = new() + { + Input = input, + TraceParent = runActivity?.Id ?? Activity.Current?.Id + }; string instanceId = await this._client.ScheduleNewOrchestrationInstanceAsync( orchestratorName: WorkflowNamingHelper.ToOrchestrationFunctionName(workflow.Name), @@ -48,7 +62,9 @@ internal sealed class DurableWorkflowClient : IWorkflowClient options: runId is not null ? new StartOrchestrationOptions(runId) : null, cancellation: cancellationToken).ConfigureAwait(false); - return new DurableWorkflowRun(this._client, instanceId, workflow.Name); + runActivity?.SetTag("run.id", instanceId); + + return new DurableWorkflowRun(this._client, instanceId, workflow.Name, runActivity); } /// @@ -74,7 +90,15 @@ internal sealed class DurableWorkflowClient : IWorkflowClient throw new ArgumentException("Workflow must have a valid Name property.", nameof(workflow)); } - DurableWorkflowInput workflowInput = new() { Input = input }; + Activity? runActivity = DurableWorkflowInstrumentation.ActivitySource.StartActivity("workflow.run"); + runActivity?.SetTag("workflow.id", workflow.StartExecutorId) + .SetTag("workflow.name", workflow.Name); + + DurableWorkflowInput workflowInput = new() + { + Input = input, + TraceParent = runActivity?.Id ?? Activity.Current?.Id + }; string instanceId = await this._client.ScheduleNewOrchestrationInstanceAsync( orchestratorName: WorkflowNamingHelper.ToOrchestrationFunctionName(workflow.Name), @@ -82,7 +106,9 @@ internal sealed class DurableWorkflowClient : IWorkflowClient options: runId is not null ? new StartOrchestrationOptions(runId) : null, cancellation: cancellationToken).ConfigureAwait(false); - return new DurableStreamingWorkflowRun(this._client, instanceId, workflow); + runActivity?.SetTag("run.id", instanceId); + + return new DurableStreamingWorkflowRun(this._client, instanceId, workflow, runActivity); } /// diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowInput.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowInput.cs index bd6f42f501..bb9a143a5b 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowInput.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowInput.cs @@ -13,4 +13,11 @@ internal sealed class DurableWorkflowInput /// Gets the workflow input data. /// public required TInput Input { get; init; } + + /// + /// Gets or sets the W3C traceparent of the client-side workflow.run span. + /// Propagated through the orchestration to activity workers so that executor spans + /// appear as children of workflow.run in the trace hierarchy. + /// + public string? TraceParent { get; set; } } diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowInstrumentation.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowInstrumentation.cs index 9f76b18a4c..83385887d3 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowInstrumentation.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowInstrumentation.cs @@ -13,4 +13,10 @@ internal static class DurableWorkflowInstrumentation /// The shared used by all durable workflow components. /// internal static readonly ActivitySource ActivitySource = new("Microsoft.Agents.AI.DurableTask.Workflows"); + + /// + /// Carries the W3C traceparent of the client-side workflow.run span through the + /// orchestrator's async call chain so it can be included in activity inputs. + /// + internal static readonly AsyncLocal WorkflowRunTraceParent = new(); } diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowRun.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowRun.cs index aeb42f4fb6..47944310a5 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowRun.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowRun.cs @@ -15,6 +15,7 @@ internal sealed class DurableWorkflowRun : IAwaitableWorkflowRun { private readonly DurableTaskClient _client; private readonly List _eventSink = []; + private Activity? _runActivity; private int _lastBookmark; /// @@ -23,11 +24,13 @@ internal sealed class DurableWorkflowRun : IAwaitableWorkflowRun /// The durable task client for orchestration operations. /// The unique instance ID for this orchestration run. /// The name of the workflow being executed. - internal DurableWorkflowRun(DurableTaskClient client, string instanceId, string workflowName) + /// The workflow.run activity to stop when the workflow completes. + internal DurableWorkflowRun(DurableTaskClient client, string instanceId, string workflowName, Activity? runActivity = null) { this._client = client; this.RunId = instanceId; this.WorkflowName = workflowName; + this._runActivity = runActivity; } /// @@ -48,33 +51,42 @@ internal sealed class DurableWorkflowRun : IAwaitableWorkflowRun /// Thrown when the workflow was terminated or ended with an unexpected status. public async ValueTask WaitForCompletionAsync(CancellationToken cancellationToken = default) { - OrchestrationMetadata metadata = await this._client.WaitForInstanceCompletionAsync( - this.RunId, - getInputsAndOutputs: true, - cancellation: cancellationToken).ConfigureAwait(false); - - if (metadata.RuntimeStatus == OrchestrationRuntimeStatus.Completed) + try { - return DurableStreamingWorkflowRun.ExtractResult(metadata.SerializedOutput); - } + OrchestrationMetadata metadata = await this._client.WaitForInstanceCompletionAsync( + this.RunId, + getInputsAndOutputs: true, + cancellation: cancellationToken).ConfigureAwait(false); - if (metadata.RuntimeStatus == OrchestrationRuntimeStatus.Failed) - { - if (metadata.FailureDetails is not null) + if (metadata.RuntimeStatus == OrchestrationRuntimeStatus.Completed) { - // Use TaskFailedException to preserve full failure details including stack trace and inner exceptions - throw new TaskFailedException( - taskName: this.WorkflowName, - taskId: 0, - failureDetails: metadata.FailureDetails); + this._runActivity?.AddEvent(new ActivityEvent("workflow.completed")); + return DurableStreamingWorkflowRun.ExtractResult(metadata.SerializedOutput); + } + + if (metadata.RuntimeStatus == OrchestrationRuntimeStatus.Failed) + { + if (metadata.FailureDetails is not null) + { + // Use TaskFailedException to preserve full failure details including stack trace and inner exceptions + throw new TaskFailedException( + taskName: this.WorkflowName, + taskId: 0, + failureDetails: metadata.FailureDetails); + } + + throw new InvalidOperationException( + $"Workflow '{this.WorkflowName}' (RunId: {this.RunId}) failed without failure details."); } throw new InvalidOperationException( - $"Workflow '{this.WorkflowName}' (RunId: {this.RunId}) failed without failure details."); + $"Workflow '{this.WorkflowName}' (RunId: {this.RunId}) ended with unexpected status: {metadata.RuntimeStatus}"); + } + finally + { + this._runActivity?.Dispose(); + this._runActivity = null; } - - throw new InvalidOperationException( - $"Workflow '{this.WorkflowName}' (RunId: {this.RunId}) ended with unexpected status: {metadata.RuntimeStatus}"); } /// diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowRunner.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowRunner.cs index f809643fd2..269b196334 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowRunner.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowRunner.cs @@ -44,7 +44,6 @@ // Superstep 5 — loop exits (no pending messages) // GetFinalResult returns resultE -using System.Diagnostics; using System.Diagnostics.CodeAnalysis; using System.Text.Json; using Microsoft.Agents.AI.DurableTask.Workflows.EdgeRouters; @@ -116,23 +115,12 @@ internal sealed class DurableWorkflowRunner // Extract input - the start executor determines the expected input type from its own InputTypes object input = workflowInput.Input; - string result = await RunSuperstepLoopAsync(context, workflow, edgeMap, input, logger).ConfigureAwait(true); + // Store the client-side workflow.run traceparent so activity dispatchers can include it + // in activity inputs. This avoids creating workflow.run in the orchestrator, where the + // replay model causes Activities to be abandoned on each re-invocation. + DurableWorkflowInstrumentation.WorkflowRunTraceParent.Value = workflowInput.TraceParent; - // Emit the workflow.run span after the superstep loop completes. - // Durable Task orchestrations replay the method from the top, where context.IsReplaying - // is true. It only transitions to false during forward progress (after all cached activity - // results are replayed). By emitting here — after RunSuperstepLoopAsync — we are guaranteed - // to be in non-replay mode, so the span is always captured exactly once. - if (!context.IsReplaying) - { - using Activity? activity = DurableWorkflowInstrumentation.ActivitySource.StartActivity("workflow.run"); - activity?.SetTag("workflow.id", workflow.StartExecutorId) - .SetTag("workflow.name", workflowName) - .SetTag("run.id", instanceId) - .AddEvent(new ActivityEvent("workflow.completed")); - } - - return result; + return await RunSuperstepLoopAsync(context, workflow, edgeMap, input, logger).ConfigureAwait(true); } private Workflow GetWorkflowOrThrow(string orchestrationName)