diff --git a/dotnet/agent-framework-dotnet.slnx b/dotnet/agent-framework-dotnet.slnx index 2c5ea815c5..452a418d32 100644 --- a/dotnet/agent-framework-dotnet.slnx +++ b/dotnet/agent-framework-dotnet.slnx @@ -54,6 +54,7 @@ + @@ -409,7 +410,6 @@ - @@ -417,6 +417,7 @@ + @@ -428,8 +429,8 @@ - + @@ -439,8 +440,8 @@ - + @@ -454,13 +455,13 @@ - + diff --git a/dotnet/samples/Durable/Workflow/ConsoleApps/09_Observability/09_Observability.csproj b/dotnet/samples/Durable/Workflow/ConsoleApps/09_Observability/09_Observability.csproj new file mode 100644 index 0000000000..d30d3d9526 --- /dev/null +++ b/dotnet/samples/Durable/Workflow/ConsoleApps/09_Observability/09_Observability.csproj @@ -0,0 +1,35 @@ + + + net10.0 + Exe + enable + enable + DurableWorkflowObservability + DurableWorkflowObservability + + + + + + + + + + + + + + + + + + + + + + diff --git a/dotnet/samples/Durable/Workflow/ConsoleApps/09_Observability/Program.cs b/dotnet/samples/Durable/Workflow/ConsoleApps/09_Observability/Program.cs new file mode 100644 index 0000000000..3a7ff6cc12 --- /dev/null +++ b/dotnet/samples/Durable/Workflow/ConsoleApps/09_Observability/Program.cs @@ -0,0 +1,140 @@ +// Copyright (c) Microsoft. All rights reserved. + +// This sample demonstrates how to enable OpenTelemetry observability for durable workflows. +// Traces are sent to an Aspire Dashboard via OTLP, and optionally to Azure Monitor +// if an Application Insights connection string is provided. +// +// The workflow is a simple text processing pipeline: +// UppercaseExecutor -> ReverseTextExecutor +// +// For input "Hello, World!", the workflow produces "!DLROW ,OLLEH". +// +// OpenTelemetry captures traces at the workflow level (executor dispatch, edge routing) +// and at the Durable Task level (orchestration replay, activity execution). +// +// Learn how to set up an Aspire dashboard here: +// https://learn.microsoft.com/en-us/dotnet/aspire/fundamentals/dashboard/standalone?tabs=bash + +using System.Diagnostics; +using Azure.Monitor.OpenTelemetry.Exporter; +using DurableWorkflowObservability; +using Microsoft.Agents.AI.DurableTask; +using Microsoft.Agents.AI.DurableTask.Workflows; +using Microsoft.Agents.AI.Workflows; +using Microsoft.DurableTask.Client.AzureManaged; +using Microsoft.DurableTask.Worker.AzureManaged; +using Microsoft.Extensions.DependencyInjection; +using Microsoft.Extensions.Hosting; +using Microsoft.Extensions.Logging; +using OpenTelemetry; +using OpenTelemetry.Resources; +using OpenTelemetry.Trace; + +// ── Configuration ──────────────────────────────────────────────────────────── +string sourceName = "DurableWorkflow.ObservabilitySample"; +ActivitySource activitySource = new(sourceName); + +string dtsConnectionString = Environment.GetEnvironmentVariable("DURABLE_TASK_SCHEDULER_CONNECTION_STRING") + ?? "Endpoint=http://localhost:8080;TaskHub=default;Authentication=None"; +string? applicationInsightsConnectionString = Environment.GetEnvironmentVariable("APPLICATIONINSIGHTS_CONNECTION_STRING"); +string otlpEndpoint = Environment.GetEnvironmentVariable("OTLP_ENDPOINT") ?? "http://localhost:4317"; + +// ── OpenTelemetry Setup ────────────────────────────────────────────────────── +ResourceBuilder resourceBuilder = ResourceBuilder + .CreateDefault() + .AddService("DurableWorkflowObservabilitySample"); + +TracerProviderBuilder traceProviderBuilder = Sdk.CreateTracerProviderBuilder() + .SetResourceBuilder(resourceBuilder) + .AddSource("Microsoft.Agents.AI.Workflows*") // Workflow-level telemetry (executors, edges) + .AddSource("Microsoft.Agents.AI.DurableTask*") // Durable workflow telemetry (orchestration, dispatch, routing) + .AddSource(sourceName) // Application-level telemetry + .AddOtlpExporter(options => options.Endpoint = new Uri(otlpEndpoint)); + +// Optionally add Azure Monitor exporter if connection string is provided +if (!string.IsNullOrWhiteSpace(applicationInsightsConnectionString)) +{ + traceProviderBuilder.AddAzureMonitorTraceExporter( + options => options.ConnectionString = applicationInsightsConnectionString); +} + +using TracerProvider? traceProvider = traceProviderBuilder.Build(); + +// Start a root activity so all workflow spans are correlated under one trace +using Activity? rootActivity = activitySource.StartActivity("main"); +Console.WriteLine($"Operation/Trace ID: {Activity.Current?.TraceId}"); + +// ── Define executors and build the workflow ────────────────────────────────── +UppercaseExecutor uppercase = new(); +ReverseTextExecutor reverse = new(); + +Workflow textProcessing = new WorkflowBuilder(uppercase) + .WithName("TextProcessing") + .WithDescription("Convert text to uppercase then reverse it") + .AddEdge(uppercase, reverse) + .Build(); + +// ── Configure the host with durable workflow support ───────────────────────── +IHost host = Host.CreateDefaultBuilder(args) + .ConfigureLogging(logging => logging.SetMinimumLevel(LogLevel.Warning)) + .ConfigureServices(services => + { + services.ConfigureDurableWorkflows( + workflowOptions => workflowOptions.AddWorkflow(textProcessing), + workerBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString), + clientBuilder: builder => builder.UseDurableTaskScheduler(dtsConnectionString)); + }) + .Build(); + +await host.StartAsync(); + +IWorkflowClient workflowClient = host.Services.GetRequiredService(); + +// ── Interactive loop ───────────────────────────────────────────────────────── +Console.WriteLine(); +Console.WriteLine("Durable Workflow Observability Sample"); +Console.WriteLine("Workflow: UppercaseExecutor -> ReverseTextExecutor"); +Console.WriteLine("Traces are exported via OTLP to: " + otlpEndpoint); +if (!string.IsNullOrWhiteSpace(applicationInsightsConnectionString)) +{ + Console.WriteLine("Traces are also exported to Azure Monitor (Application Insights)."); +} + +Console.WriteLine(); +Console.WriteLine("Enter text to process (or 'exit' to quit):"); + +while (true) +{ + Console.Write("> "); + string? input = Console.ReadLine(); + if (string.IsNullOrWhiteSpace(input) || input.Equals("exit", StringComparison.OrdinalIgnoreCase)) + { + break; + } + + try + { + // Create a child activity for each workflow invocation + using Activity? invocationActivity = activitySource.StartActivity("ProcessText"); + invocationActivity?.SetTag("input.text", input); + + Console.WriteLine($"Starting workflow for input: \"{input}\"..."); + + IAwaitableWorkflowRun run = (IAwaitableWorkflowRun)await workflowClient.RunAsync(textProcessing, input); + Console.WriteLine($"Run ID: {run.RunId}"); + + Console.WriteLine("Waiting for workflow to complete..."); + string? result = await run.WaitForCompletionAsync(); + + invocationActivity?.SetTag("output.text", result); + Console.WriteLine($"Result: {result}"); + } + catch (Exception ex) + { + Console.WriteLine($"Error: {ex.Message}"); + } + + Console.WriteLine(); +} + +await host.StopAsync(); diff --git a/dotnet/samples/Durable/Workflow/ConsoleApps/09_Observability/README.md b/dotnet/samples/Durable/Workflow/ConsoleApps/09_Observability/README.md new file mode 100644 index 0000000000..facee333ea --- /dev/null +++ b/dotnet/samples/Durable/Workflow/ConsoleApps/09_Observability/README.md @@ -0,0 +1,100 @@ +# Observability Sample + +This sample demonstrates how to enable **OpenTelemetry observability** for durable workflows. It shows how to capture and export traces from workflow execution, including executor dispatch, edge routing, and Durable Task orchestration replays. + +## Key Concepts Demonstrated + +- Configuring OpenTelemetry tracing for durable workflows +- Subscribing to `Microsoft.Agents.AI.Workflows*` activity sources for workflow-level telemetry +- Exporting traces via OTLP (for Aspire Dashboard) +- Optionally exporting traces to Azure Monitor (Application Insights) +- Correlating all workflow spans under a single trace ID + +## Overview + +The sample implements a simple text processing pipeline that runs as a durable workflow: + +``` +UppercaseExecutor --> ReverseTextExecutor +``` + +| Executor | Description | +|----------|-------------| +| UppercaseExecutor | Converts input text to uppercase | +| ReverseTextExecutor | Reverses the text | + +For input `"Hello, World!"`, the workflow produces `"!DLROW ,OLLEH"`. + +## Observability Setup + +The sample configures OpenTelemetry to capture traces from: + +1. **Workflow-level telemetry** (`Microsoft.Agents.AI.Workflows*`): Captures executor execution, edge routing, and workflow lifecycle events. +2. **Durable workflow telemetry** (`Microsoft.Agents.AI.DurableTask*`): Captures durable orchestration lifecycle, executor dispatch, and edge routing within the durable execution environment. +3. **Application-level telemetry**: Custom spans for each workflow invocation with input/output tags. + +Traces are exported to: + +- **Aspire Dashboard** (default): Via OTLP exporter to `http://localhost:4317` +- **Azure Monitor** (optional): If `APPLICATIONINSIGHTS_CONNECTION_STRING` is set + +## Environment Setup + +See the [README.md](../README.md) file in the parent directory for information on configuring the Durable Task Scheduler. + +### Aspire Dashboard + +To visualize traces, start an Aspire Dashboard: + +```bash +docker run --rm -it -d -p 18888:18888 -p 4317:18889 --name aspire-dashboard mcr.microsoft.com/dotnet/aspire-dashboard:9.0 +``` + +Then open `http://localhost:18888` in your browser. + +Learn more: [Aspire Dashboard Standalone](https://learn.microsoft.com/en-us/dotnet/aspire/fundamentals/dashboard/standalone?tabs=bash) + +## Environment Variables + +| Variable | Required | Description | +|----------|----------|-------------| +| `DURABLE_TASK_SCHEDULER_CONNECTION_STRING` | No | Connection string for the Durable Task Scheduler. Defaults to local emulator. | +| `OTLP_ENDPOINT` | No | OTLP exporter endpoint. Defaults to `http://localhost:4317`. | +| `APPLICATIONINSIGHTS_CONNECTION_STRING` | No | Application Insights connection string. If set, traces are also sent to Azure Monitor. | + +## Running the Sample + +```bash +cd dotnet/samples/Durable/Workflow/ConsoleApps/09_Observability +dotnet run --framework net10.0 +``` + +### Sample Output + +```text +Operation/Trace ID: abc123def456... + +Durable Workflow Observability Sample +Workflow: UppercaseExecutor -> ReverseTextExecutor +Traces are exported via OTLP to: http://localhost:4317 + +Enter text to process (or 'exit' to quit): +> Hello, World! +Starting workflow for input: "Hello, World!"... +Run ID: xyz789... +Waiting for workflow to complete... + [UppercaseExecutor] Processing: "Hello, World!" + [UppercaseExecutor] Result: "HELLO, WORLD!" + [ReverseTextExecutor] Processing: "HELLO, WORLD!" + [ReverseTextExecutor] Result: "!DLROW ,OLLEH" +Result: !DLROW ,OLLEH + +> exit +``` + +After running, open the Aspire Dashboard to view the trace. You will see spans for: + +- The root `main` activity +- Individual `ProcessText` activities for each invocation +- Workflow executor spans (UppercaseExecutor, ReverseTextExecutor) +- Edge routing spans between executors diff --git a/dotnet/samples/Durable/Workflow/ConsoleApps/09_Observability/TextProcessingExecutors.cs b/dotnet/samples/Durable/Workflow/ConsoleApps/09_Observability/TextProcessingExecutors.cs new file mode 100644 index 0000000000..cd70ac8f46 --- /dev/null +++ b/dotnet/samples/Durable/Workflow/ConsoleApps/09_Observability/TextProcessingExecutors.cs @@ -0,0 +1,59 @@ +// Copyright (c) Microsoft. All rights reserved. + +using Microsoft.Agents.AI.Workflows; + +namespace DurableWorkflowObservability; + +/// +/// First executor: converts input text to uppercase. +/// +internal sealed class UppercaseExecutor() : Executor("UppercaseExecutor") +{ + /// + /// Processes the input message by converting it to uppercase. + /// + /// The input text to convert + /// Workflow context for accessing workflow services and adding events + /// The to monitor for cancellation requests. + /// The default is . + /// The input text converted to uppercase + public override async ValueTask HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.WriteLine($" [UppercaseExecutor] Processing: \"{message}\""); + + // Simulate some processing time + await Task.Delay(TimeSpan.FromMilliseconds(100), cancellationToken); + + string result = message.ToUpperInvariant(); + Console.WriteLine($" [UppercaseExecutor] Result: \"{result}\""); + + return result; + } +} + +/// +/// Second executor: reverses the input text. +/// +internal sealed class ReverseTextExecutor() : Executor("ReverseTextExecutor") +{ + /// + /// Processes the input message by reversing the text. + /// + /// The input text to reverse + /// Workflow context for accessing workflow services and adding events + /// The to monitor for cancellation requests. + /// The default is . + /// The input text reversed + public override async ValueTask HandleAsync(string message, IWorkflowContext context, CancellationToken cancellationToken = default) + { + Console.WriteLine($" [ReverseTextExecutor] Processing: \"{message}\""); + + // Simulate some processing time + await Task.Delay(TimeSpan.FromMilliseconds(100), cancellationToken); + + string result = new(message.Reverse().ToArray()); + Console.WriteLine($" [ReverseTextExecutor] Result: \"{result}\""); + + return result; + } +} diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableExecutorDispatcher.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableExecutorDispatcher.cs index a0257c6d91..2b94f23c93 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableExecutorDispatcher.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableExecutorDispatcher.cs @@ -43,6 +43,9 @@ internal static class DurableExecutorDispatcher { logger.LogDispatchingExecutor(executorInfo.ExecutorId, executorInfo.IsAgenticExecutor); + // Note: executor.process telemetry is emitted by core Executor.ExecuteAsync inside the + // activity worker. We don't duplicate it here because orchestration replay semantics + // make wrapping spans unreliable (context.IsReplaying is true at method entry). if (executorInfo.IsAgenticExecutor) { return await ExecuteAgentAsync(context, executorInfo, logger, envelope.Message).ConfigureAwait(true); diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableRunStatus.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableRunStatus.cs index 4ed2049dc9..cff00a84ca 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableRunStatus.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableRunStatus.cs @@ -1,4 +1,4 @@ -// Copyright (c) Microsoft. All rights reserved. +// Copyright (c) Microsoft. All rights reserved. namespace Microsoft.Agents.AI.DurableTask.Workflows; diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowContext.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowContext.cs index 9ddb337561..3b2ef34eeb 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowContext.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowContext.cs @@ -1,4 +1,4 @@ -// Copyright (c) Microsoft. All rights reserved. +// Copyright (c) Microsoft. All rights reserved. using System.Diagnostics; using System.Diagnostics.CodeAnalysis; @@ -33,10 +33,12 @@ internal sealed class DurableWorkflowContext : IWorkflowContext /// /// The shared state passed from the orchestration. /// The executor running in this context. - internal DurableWorkflowContext(Dictionary? initialState, Executor executor) + /// Optional trace context for correlation. + internal DurableWorkflowContext(Dictionary? initialState, Executor executor, IReadOnlyDictionary? traceContext = null) { this._executor = executor; this._initialState = initialState ?? []; + this.TraceContext = traceContext; } /// @@ -87,6 +89,8 @@ internal sealed class DurableWorkflowContext : IWorkflowContext { if (message is not null) { + using Activity? activity = DurableWorkflowInstrumentation.ActivitySource.StartActivity("message.send", ActivityKind.Producer); + Type messageType = message.GetType(); this.SentMessages.Add(new TypedPayload { @@ -269,7 +273,7 @@ internal sealed class DurableWorkflowContext : IWorkflowContext } /// - public IReadOnlyDictionary? TraceContext => null; + public IReadOnlyDictionary? TraceContext { get; } /// public bool ConcurrentRunsEnabled => false; diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowInstrumentation.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowInstrumentation.cs new file mode 100644 index 0000000000..9f76b18a4c --- /dev/null +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowInstrumentation.cs @@ -0,0 +1,16 @@ +// Copyright (c) Microsoft. All rights reserved. + +using System.Diagnostics; + +namespace Microsoft.Agents.AI.DurableTask.Workflows; + +/// +/// Provides centralized OpenTelemetry instrumentation for durable workflow execution. +/// +internal static class DurableWorkflowInstrumentation +{ + /// + /// The shared used by all durable workflow components. + /// + internal static readonly ActivitySource ActivitySource = new("Microsoft.Agents.AI.DurableTask.Workflows"); +} diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowRunner.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowRunner.cs index d133d16919..f809643fd2 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowRunner.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/DurableWorkflowRunner.cs @@ -44,6 +44,7 @@ // 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; @@ -115,7 +116,23 @@ internal sealed class DurableWorkflowRunner // Extract input - the start executor determines the expected input type from its own InputTypes object input = workflowInput.Input; - return await RunSuperstepLoopAsync(context, workflow, edgeMap, input, logger).ConfigureAwait(true); + string result = await RunSuperstepLoopAsync(context, workflow, edgeMap, input, logger).ConfigureAwait(true); + + // 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; } private Workflow GetWorkflowOrThrow(string orchestrationName) diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/EdgeRouters/DurableDirectEdgeRouter.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/EdgeRouters/DurableDirectEdgeRouter.cs index 3f78093183..cf8bdd540c 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/EdgeRouters/DurableDirectEdgeRouter.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/EdgeRouters/DurableDirectEdgeRouter.cs @@ -29,6 +29,7 @@ // Enqueue to // D's queue +using System.Diagnostics; using System.Diagnostics.CodeAnalysis; using System.Text.Json; using Microsoft.Extensions.Logging; @@ -84,6 +85,12 @@ internal sealed class DurableDirectEdgeRouter : IDurableEdgeRouter Dictionary> messageQueues, ILogger logger) { + using Activity? activity = DurableWorkflowInstrumentation.ActivitySource.StartActivity("edge_group.process"); + activity? + .SetTag("edge_group.type", nameof(DurableDirectEdgeRouter)) + .SetTag("message.source_id", this._sourceId) + .SetTag("message.target_id", this._sinkId); + if (this._condition is not null) { try @@ -92,17 +99,23 @@ internal sealed class DurableDirectEdgeRouter : IDurableEdgeRouter if (!this._condition(messageObj)) { logger.LogEdgeConditionFalse(this._sourceId, this._sinkId); + activity?.SetTag("edge_group.delivered", false) + .SetTag("edge_group.delivery_status", "dropped condition false"); return; } } catch (Exception ex) { logger.LogEdgeConditionEvaluationFailed(ex, this._sourceId, this._sinkId); + activity?.SetTag("edge_group.delivered", false) + .SetTag("edge_group.delivery_status", "exception"); return; } } logger.LogEdgeRoutingMessage(this._sourceId, this._sinkId); + activity?.SetTag("edge_group.delivered", true) + .SetTag("edge_group.delivery_status", "delivered"); EnqueueMessage(messageQueues, this._sinkId, envelope); } diff --git a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/EdgeRouters/DurableFanOutEdgeRouter.cs b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/EdgeRouters/DurableFanOutEdgeRouter.cs index f13a0def92..981ba4d146 100644 --- a/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/EdgeRouters/DurableFanOutEdgeRouter.cs +++ b/dotnet/src/Microsoft.Agents.AI.DurableTask/Workflows/EdgeRouters/DurableFanOutEdgeRouter.cs @@ -19,6 +19,7 @@ // Each DirectRouter independently evaluates its condition, // so resultB always reaches C, but only reaches D if NeedsReview is true. +using System.Diagnostics; using Microsoft.Extensions.Logging; namespace Microsoft.Agents.AI.DurableTask.Workflows.EdgeRouters; @@ -54,6 +55,11 @@ internal sealed class DurableFanOutEdgeRouter : IDurableEdgeRouter Dictionary> messageQueues, ILogger logger) { + using Activity? activity = DurableWorkflowInstrumentation.ActivitySource.StartActivity("edge_group.process"); + activity? + .SetTag("edge_group.type", nameof(DurableFanOutEdgeRouter)) + .SetTag("message.source_id", this._sourceId); + if (logger.IsEnabled(LogLevel.Debug)) { logger.LogDebug("Fan-Out from {Source}: routing to {Count} targets", this._sourceId, this._targetRouters.Count);