OpenTelemetry tracing and metrics for Send, CreateStream, Publish and handlers

README.md

@@ -324,6 +324,18 @@ public class OrderService(IMediator mediator)
 
 You can inject `ISender`, `IPublisher`, or `IMediator` depending on what the consumer needs.
 
+## Tracing and Metrics
+
+AnyMediator emits OpenTelemetry spans for every `Send`, `CreateStream`, `Publish` and notification handler, and an `anymediator.operation.duration` histogram, through .NET's own `ActivitySource` and `Meter`. Nothing is recorded until you subscribe:
+
+```csharp
+builder.Services.AddOpenTelemetry()
+    .WithTracing(t => t.AddSource(AnyMediatorDiagnostics.ActivitySourceName))
+    .WithMetrics(m => m.AddMeter(AnyMediatorDiagnostics.MeterName));
+```
+
+Span names, attributes and the metric are described in the [tracing guide](https://adiakys.github.io/AnyMediator/guide/tracing).
+
 ## Configuration
 
 ```csharp

docs/.vitepress/config.mts

@@ -18,6 +18,7 @@ const guide = [
   { text: 'Exception handling', link: '/guide/exception-handling' },
   { text: 'Notification sinks', link: '/guide/notification-sinks' },
   { text: 'Configuration', link: '/guide/configuration' },
+  { text: 'Tracing and metrics', link: '/guide/tracing' },
   { text: 'Performance', link: '/guide/performance' },
 ]
 

docs/guide/tracing.md

@@ -0,0 +1,83 @@
+---
+description: "Trace every Send, CreateStream, Publish and notification handler with OpenTelemetry, and measure their duration, using AnyMediator's built-in ActivitySource and Meter."
+---
+
+# Tracing and metrics
+
+AnyMediator emits OpenTelemetry-compatible spans and a duration metric through .NET's own `ActivitySource` and `Meter`. There is no extra package to install and no option to turn on: subscribe to the names in `AnyMediatorDiagnostics`, and nothing is recorded until something does.
+
+## Set up OpenTelemetry
+
+```csharp
+using AnyMediator;
+
+builder.Services.AddOpenTelemetry()
+    .WithTracing(tracing => tracing
+        .AddSource(AnyMediatorDiagnostics.ActivitySourceName)
+        .AddAspNetCoreInstrumentation()
+        .AddOtlpExporter())
+    .WithMetrics(metrics => metrics
+        .AddMeter(AnyMediatorDiagnostics.MeterName)
+        .AddOtlpExporter());
+```
+
+A request to `GET /users/42` then shows up as one trace: the ASP.NET Core span, the `send GetUser` span under it, and the database spans under that.
+
+## Spans
+
+| Operation | Span name | Ends |
+|-----------|-----------|------|
+| `Send` | `send {RequestType}` | when the handler and every behavior have finished |
+| `CreateStream` | `create_stream {RequestType}` | when the stream is fully read, fails, or is abandoned |
+| `Publish` | `publish {NotificationType}` | when every sink has finished |
+| each notification handler | `handle {NotificationType}`, child of the publish span | when the handler has finished |
+
+Attributes:
+
+| Attribute | Value |
+|-----------|-------|
+| `anymediator.operation` | `send`, `create_stream`, `publish` or `handle` |
+| `anymediator.message.type` | full name of the request or notification type |
+| `anymediator.handler.type` | full name of the notification handler, on `handle` spans |
+| `error.type` | full name of the exception, when the operation failed |
+
+A failed operation sets the span status to `Error` and adds an `exception` event with `exception.type`, `exception.message` and `exception.stacktrace`, as the OpenTelemetry conventions describe. Spans started by your handlers, such as database or HTTP calls, become children of the AnyMediator span.
+
+## Metric
+
+| Name | Type | Unit | Attributes |
+|------|------|------|------------|
+| `anymediator.operation.duration` | histogram | seconds | `anymediator.operation`, `anymediator.message.type`, `error.type` when failed |
+
+The histogram also gives the count of operations and of failures, per message type: for example the rate of `send` operations whose `error.type` is set.
+
+## Cost
+
+With no listener, each operation checks two flags and continues: the [benchmarks](./performance) are unchanged. With a listener that records everything, as an OpenTelemetry SDK sampling every request does, each span costs about half a microsecond, mostly the `Activity` object that any instrumented library creates:
+
+| Operation | Not listening | Listening |
+|-----------|--------------:|----------:|
+| `Send` (1 span) | 87 ns, 96 B | 564 ns, 704 B |
+| `Publish` with 3 handlers (4 spans) | 307 ns, 632 B | 2,310 ns, 2,872 B |
+
+Next to a database query or an HTTP call this is not measurable. With a sampling rate below 100%, requests that are not sampled create no span and only record their duration.
+
+## Across services
+
+A [notification sink](./notification-sinks) that forwards notifications to a message broker can carry the trace along, so the consumer's spans join the same trace. Add the current trace context to the message:
+
+```csharp
+public Task Dispatch<TNotification>(TNotification notification, CancellationToken cancellationToken = default)
+    where TNotification : INotification
+{
+    var message = new ServiceBusMessage(BinaryData.FromObjectAsJson(notification)) { Subject = typeof(TNotification).Name };
+
+    // W3C trace context of the publish span, read back by the consumer to continue the trace.
+    if (Activity.Current is { } activity)
+        message.ApplicationProperties["traceparent"] = activity.Id;
+
+    return sender.SendMessageAsync(message, cancellationToken);
+}
+```
+
+Many broker clients, the Azure Service Bus and RabbitMQ ones among them, have OpenTelemetry instrumentation that does this for you; then the sink needs no code at all.

docs/public/llms.txt

@@ -19,6 +19,7 @@ Differences from MediatR: two handlers for the same request make `AddAnyMediator
 - [Pre- and post-processors](https://adiakys.github.io/AnyMediator/guide/processors): IRequestPreProcessor, IRequestPostProcessor
 - [Exception handling](https://adiakys.github.io/AnyMediator/guide/exception-handling): IRequestExceptionHandler, IRequestExceptionAction
 - [Notification sinks](https://adiakys.github.io/AnyMediator/guide/notification-sinks): INotificationSink for brokers and webhooks
+- [Tracing and metrics](https://adiakys.github.io/AnyMediator/guide/tracing): OpenTelemetry spans and duration histogram via AnyMediatorDiagnostics.ActivitySourceName and MeterName
 - [Configuration](https://adiakys.github.io/AnyMediator/guide/configuration): every AddAnyMediator option
 
 ## Optional

src/AnyMediator/AnyMediatorPublisher.cs

@@ -8,10 +8,24 @@ internal sealed class AnyMediatorPublisher(IServiceProvider services) : IPublish
 {
     private static readonly ConcurrentDictionary<Type, Func<AnyMediatorPublisher, object, CancellationToken, Task>> ObjectPublishCache = new();
 
-    public async Task Publish<TNotification>(TNotification notification, CancellationToken cancellationToken = default) where TNotification : INotification
+    public Task Publish<TNotification>(TNotification notification, CancellationToken cancellationToken = default) where TNotification : INotification
     {
         ArgumentNullException.ThrowIfNull(notification);
 
+        return Telemetry.IsEnabled
+            ? PublishTraced(notification, cancellationToken)
+            : PublishCore(notification, cancellationToken);
+    }
+
+    private static readonly ConcurrentDictionary<Type, OperationSpan> PublishSpans = new();
+
+    private Task PublishTraced<TNotification>(TNotification notification, CancellationToken cancellationToken) where TNotification : INotification
+        => Telemetry.Run(PublishSpans.GetOrAdd(notification.GetType(), static type => new OperationSpan("publish", type)),
+            (publisher: this, notification, cancellationToken),
+            static s => s.publisher.PublishCore(s.notification, s.cancellationToken));
+
+    private async Task PublishCore<TNotification>(TNotification notification, CancellationToken cancellationToken) where TNotification : INotification
+    {
         // Every sink runs even if an earlier one fails; the failures reach the caller afterwards, as in MediatR.
         List<Exception>? failures = null;
 

src/AnyMediator/Diagnostics/AnyMediatorDiagnostics.cs

@@ -0,0 +1,21 @@
+namespace AnyMediator;
+
+/// <summary>
+/// Names to subscribe to AnyMediator's traces and metrics, for example with OpenTelemetry:
+/// <c>.WithTracing(t => t.AddSource(AnyMediatorDiagnostics.ActivitySourceName))</c> and
+/// <c>.WithMetrics(m => m.AddMeter(AnyMediatorDiagnostics.MeterName))</c>. Nothing is recorded until a listener subscribes.
+/// </summary>
+public static class AnyMediatorDiagnostics
+{
+    /// <summary>
+    /// The <see cref="System.Diagnostics.ActivitySource"/> name. It emits one span per <c>Send</c>, <c>CreateStream</c>
+    /// and <c>Publish</c>, and one child span per notification handler.
+    /// </summary>
+    public const string ActivitySourceName = "AnyMediator";
+
+    /// <summary>
+    /// The <see cref="System.Diagnostics.Metrics.Meter"/> name. It records the <c>anymediator.operation.duration</c>
+    /// histogram, in seconds.
+    /// </summary>
+    public const string MeterName = "AnyMediator";
+}

src/AnyMediator/Diagnostics/Telemetry.cs

@@ -0,0 +1,144 @@
+using System.Diagnostics;
+using System.Diagnostics.Metrics;
+using System.Runtime.CompilerServices;
+
+namespace AnyMediator;
+
+/// <summary>
+/// Spans and duration measurements for mediator operations. Callers check <see cref="IsEnabled"/> first, so with no
+/// listener the dispatch path pays one check and nothing else. Operations take their inputs as a state value and a
+/// static lambda, so tracing allocates no closure.
+/// </summary>
+internal static class Telemetry
+{
+    private static readonly string? Version = typeof(Telemetry).Assembly.GetName().Version?.ToString(3);
+    internal static readonly ActivitySource Source = new(AnyMediatorDiagnostics.ActivitySourceName, Version);
+    private static readonly Meter Meter = new(AnyMediatorDiagnostics.MeterName, Version);
+
+    internal static readonly Histogram<double> Duration = Meter.CreateHistogram<double>(
+        "anymediator.operation.duration", "s", "Duration of a mediator operation, behaviors and handlers included.");
+
+    internal static bool IsEnabled => Source.HasListeners() || Duration.Enabled;
+
+    internal static async Task<T> Run<TState, T>(OperationSpan span, TState state, Func<TState, Task<T>> operation)
+    {
+        using var activity = span.Start();
+        var start = Stopwatch.GetTimestamp();
+        Exception? failure = null;
+        try
+        {
+            return await operation(state);
+        }
+        catch (Exception ex)
+        {
+            RecordFailure(activity, failure = ex);
+            throw;
+        }
+        finally
+        {
+            span.RecordDuration(start, failure);
+        }
+    }
+
+    internal static async Task Run<TState>(OperationSpan span, TState state, Func<TState, Task> operation)
+    {
+        using var activity = span.Start();
+        var start = Stopwatch.GetTimestamp();
+        Exception? failure = null;
+        try
+        {
+            await operation(state);
+        }
+        catch (Exception ex)
+        {
+            RecordFailure(activity, failure = ex);
+            throw;
+        }
+        finally
+        {
+            span.RecordDuration(start, failure);
+        }
+    }
+
+    /// <summary>A span that lasts until the stream is fully enumerated, fails or is abandoned.</summary>
+    internal static async IAsyncEnumerable<T> RunStream<TState, T>(
+        OperationSpan span, TState state, Func<TState, IAsyncEnumerable<T>> operation, [EnumeratorCancellation] CancellationToken cancellationToken = default)
+    {
+        using var activity = span.Start();
+        var start = Stopwatch.GetTimestamp();
+        Exception? failure = null;
+
+        await using var items = operation(state).GetAsyncEnumerator(cancellationToken);
+        try
+        {
+            while (true)
+            {
+                bool hasItem;
+                try
+                {
+                    hasItem = await items.MoveNextAsync();
+                }
+                catch (Exception ex)
+                {
+                    RecordFailure(activity, failure = ex);
+                    throw;
+                }
+
+                if (!hasItem) yield break;
+                yield return items.Current;
+            }
+        }
+        finally
+        {
+            span.RecordDuration(start, failure);
+        }
+    }
+
+    // Follows the OpenTelemetry conventions for exceptions: error status, error.type, and an "exception" event.
+    private static void RecordFailure(Activity? activity, Exception exception)
+    {
+        if (activity is null) return;
+
+        activity.SetStatus(ActivityStatusCode.Error, exception.Message);
+        activity.SetTag("error.type", exception.GetType().FullName);
+        activity.AddEvent(new ActivityEvent("exception", tags: new ActivityTagsCollection
+        {
+            ["exception.type"] = exception.GetType().FullName,
+            ["exception.message"] = exception.Message,
+            ["exception.stacktrace"] = exception.ToString(),
+        }));
+    }
+}
+
+/// <summary>
+/// The name and attributes of one kind of span, such as "send GetUser", built once and reused by every operation of
+/// that kind.
+/// </summary>
+internal sealed class OperationSpan(string operation, Type messageType, Type? handlerType = null)
+{
+    private readonly string _name = $"{operation} {messageType.Name}";
+
+    private readonly KeyValuePair<string, object?>[] _metricTags =
+    [
+        new("anymediator.operation", operation),
+        new("anymediator.message.type", messageType.FullName),
+    ];
+
+    private readonly KeyValuePair<string, object?>[] _spanTags = handlerType is null
+        ? [new("anymediator.operation", operation), new("anymediator.message.type", messageType.FullName)]
+        : [new("anymediator.operation", operation), new("anymediator.message.type", messageType.FullName), new("anymediator.handler.type", handlerType.FullName)];
+
+    // A default parent context makes the span a child of Activity.Current, as StartActivity(name) does.
+    public Activity? Start() => Telemetry.Source.StartActivity(_name, ActivityKind.Internal, default(ActivityContext), _spanTags);
+
+    public void RecordDuration(long start, Exception? failure)
+    {
+        if (!Telemetry.Duration.Enabled) return;
+
+        var seconds = Stopwatch.GetElapsedTime(start).TotalSeconds;
+        if (failure is null)
+            Telemetry.Duration.Record(seconds, _metricTags);
+        else
+            Telemetry.Duration.Record(seconds, [.. _metricTags, new("error.type", failure.GetType().FullName)]);
+    }
+}

src/AnyMediator/HandlerWrappers.cs

@@ -1,3 +1,4 @@
+using System.Collections.Concurrent;
 using System.Runtime.CompilerServices;
 
 namespace AnyMediator;
@@ -23,12 +24,23 @@ internal sealed class RequestHandlerWrapper<TRequest, TResponse> : RequestHandle
     where TRequest : IRequest<TResponse>
 {
     public override Task<TResponse> Handle(IRequest<TResponse> request, IServiceProvider services, HandlerRegistry registry, CancellationToken cancellationToken)
+        => Telemetry.IsEnabled
+            ? HandleTraced((TRequest)request, services, registry, cancellationToken)
+            : HandleCore((TRequest)request, services, registry, cancellationToken);
+
+    private static readonly OperationSpan Span = new("send", typeof(TRequest));
+
+    private static Task<TResponse> HandleTraced(TRequest request, IServiceProvider services, HandlerRegistry registry, CancellationToken cancellationToken)
+        => Telemetry.Run(Span, (request, services, registry, cancellationToken),
+            static s => HandleCore(s.request, s.services, s.registry, s.cancellationToken));
+
+    private static Task<TResponse> HandleCore(TRequest request, IServiceProvider services, HandlerRegistry registry, CancellationToken cancellationToken)
     {
         var behaviors = services.GetAll<IPipelineBehavior<TRequest, TResponse>>();
 
         return behaviors.Length == 0
-            ? registry.GetRequired<IRequestHandler<TRequest, TResponse>>(services).Handle((TRequest)request, cancellationToken)
-            : RunPipeline((TRequest)request, behaviors, services, registry, cancellationToken);
+            ? registry.GetRequired<IRequestHandler<TRequest, TResponse>>(services).Handle(request, cancellationToken)
+            : RunPipeline(request, behaviors, services, registry, cancellationToken);
     }
 
     // Apart from Handle so that its closures are only allocated when there are behaviors.
@@ -75,12 +87,23 @@ internal sealed class StreamHandlerWrapper<TRequest, TResponse> : StreamHandlerW
     where TRequest : IStreamRequest<TResponse>
 {
     public override IAsyncEnumerable<TResponse> Handle(IStreamRequest<TResponse> request, IServiceProvider services, HandlerRegistry registry, CancellationToken cancellationToken)
+        => Telemetry.IsEnabled
+            ? HandleTraced((TRequest)request, services, registry, cancellationToken)
+            : HandleCore((TRequest)request, services, registry, cancellationToken);
+
+    private static readonly OperationSpan Span = new("create_stream", typeof(TRequest));
+
+    private static IAsyncEnumerable<TResponse> HandleTraced(TRequest request, IServiceProvider services, HandlerRegistry registry, CancellationToken cancellationToken)
+        => Telemetry.RunStream(Span, (request, services, registry, cancellationToken),
+            static s => HandleCore(s.request, s.services, s.registry, s.cancellationToken), cancellationToken);
+
+    private static IAsyncEnumerable<TResponse> HandleCore(TRequest request, IServiceProvider services, HandlerRegistry registry, CancellationToken cancellationToken)
     {
         var behaviors = services.GetAll<IStreamPipelineBehavior<TRequest, TResponse>>();
 
         return behaviors.Length == 0
-            ? registry.GetRequired<IStreamRequestHandler<TRequest, TResponse>>(services).Handle((TRequest)request, cancellationToken)
-            : RunPipeline((TRequest)request, behaviors, services, registry, cancellationToken);
+            ? registry.GetRequired<IStreamRequestHandler<TRequest, TResponse>>(services).Handle(request, cancellationToken)
+            : RunPipeline(request, behaviors, services, registry, cancellationToken);
     }
 
     private static IAsyncEnumerable<TResponse> RunPipeline(
@@ -127,5 +150,18 @@ public override object[] ResolveHandlers(IServiceProvider services)
     }
 
     public override Func<Task> CreateCall(object handler, object notification, CancellationToken cancellationToken)
+        => Telemetry.IsEnabled ? CreateTracedCall(handler, notification, cancellationToken) : CreatePlainCall(handler, notification, cancellationToken);
+
+    private static Func<Task> CreatePlainCall(object handler, object notification, CancellationToken cancellationToken)
         => () => ((INotificationHandler<TNotification>)handler).Handle((TNotification)notification, cancellationToken);
+
+    private static readonly ConcurrentDictionary<Type, OperationSpan> HandlerSpans = new();
+
+    // One child span per handler, under the publish span.
+    private static Func<Task> CreateTracedCall(object handler, object notification, CancellationToken cancellationToken)
+    {
+        var span = HandlerSpans.GetOrAdd(handler.GetType(), static type => new OperationSpan("handle", typeof(TNotification), type));
+        return () => Telemetry.Run(span, (handler, notification, cancellationToken),
+            static s => ((INotificationHandler<TNotification>)s.handler).Handle((TNotification)s.notification, s.cancellationToken));
+    }
 }