diff --git a/README.md b/README.md index 1eff57c..78ef404 100644 --- a/README.md +++ b/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 diff --git a/docs/.vitepress/config.mts b/docs/.vitepress/config.mts index 011d755..3289637 100644 --- a/docs/.vitepress/config.mts +++ b/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' }, ] diff --git a/docs/guide/tracing.md b/docs/guide/tracing.md new file mode 100644 index 0000000..fee3931 --- /dev/null +++ b/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 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. diff --git a/docs/public/llms.txt b/docs/public/llms.txt index 9a008fd..e632ac7 100644 --- a/docs/public/llms.txt +++ b/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 diff --git a/src/AnyMediator/AnyMediatorPublisher.cs b/src/AnyMediator/AnyMediatorPublisher.cs index ffabf9f..c7ec1e9 100644 --- a/src/AnyMediator/AnyMediatorPublisher.cs +++ b/src/AnyMediator/AnyMediatorPublisher.cs @@ -8,10 +8,24 @@ internal sealed class AnyMediatorPublisher(IServiceProvider services) : IPublish { private static readonly ConcurrentDictionary> ObjectPublishCache = new(); - public async Task Publish(TNotification notification, CancellationToken cancellationToken = default) where TNotification : INotification + public Task Publish(TNotification notification, CancellationToken cancellationToken = default) where TNotification : INotification { ArgumentNullException.ThrowIfNull(notification); + return Telemetry.IsEnabled + ? PublishTraced(notification, cancellationToken) + : PublishCore(notification, cancellationToken); + } + + private static readonly ConcurrentDictionary PublishSpans = new(); + + private Task PublishTraced(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 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? failures = null; diff --git a/src/AnyMediator/Diagnostics/AnyMediatorDiagnostics.cs b/src/AnyMediator/Diagnostics/AnyMediatorDiagnostics.cs new file mode 100644 index 0000000..e196ed7 --- /dev/null +++ b/src/AnyMediator/Diagnostics/AnyMediatorDiagnostics.cs @@ -0,0 +1,21 @@ +namespace AnyMediator; + +/// +/// Names to subscribe to AnyMediator's traces and metrics, for example with OpenTelemetry: +/// .WithTracing(t => t.AddSource(AnyMediatorDiagnostics.ActivitySourceName)) and +/// .WithMetrics(m => m.AddMeter(AnyMediatorDiagnostics.MeterName)). Nothing is recorded until a listener subscribes. +/// +public static class AnyMediatorDiagnostics +{ + /// + /// The name. It emits one span per Send, CreateStream + /// and Publish, and one child span per notification handler. + /// + public const string ActivitySourceName = "AnyMediator"; + + /// + /// The name. It records the anymediator.operation.duration + /// histogram, in seconds. + /// + public const string MeterName = "AnyMediator"; +} diff --git a/src/AnyMediator/Diagnostics/Telemetry.cs b/src/AnyMediator/Diagnostics/Telemetry.cs new file mode 100644 index 0000000..0fa1b4a --- /dev/null +++ b/src/AnyMediator/Diagnostics/Telemetry.cs @@ -0,0 +1,144 @@ +using System.Diagnostics; +using System.Diagnostics.Metrics; +using System.Runtime.CompilerServices; + +namespace AnyMediator; + +/// +/// Spans and duration measurements for mediator operations. Callers check 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. +/// +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 Duration = Meter.CreateHistogram( + "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 Run(OperationSpan span, TState state, Func> 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(OperationSpan span, TState state, Func 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); + } + } + + /// A span that lasts until the stream is fully enumerated, fails or is abandoned. + internal static async IAsyncEnumerable RunStream( + OperationSpan span, TState state, Func> 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(), + })); + } +} + +/// +/// The name and attributes of one kind of span, such as "send GetUser", built once and reused by every operation of +/// that kind. +/// +internal sealed class OperationSpan(string operation, Type messageType, Type? handlerType = null) +{ + private readonly string _name = $"{operation} {messageType.Name}"; + + private readonly KeyValuePair[] _metricTags = + [ + new("anymediator.operation", operation), + new("anymediator.message.type", messageType.FullName), + ]; + + private readonly KeyValuePair[] _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)]); + } +} diff --git a/src/AnyMediator/HandlerWrappers.cs b/src/AnyMediator/HandlerWrappers.cs index 37c6f4c..6a5c285 100644 --- a/src/AnyMediator/HandlerWrappers.cs +++ b/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 : RequestHandle where TRequest : IRequest { public override Task Handle(IRequest 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 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 HandleCore(TRequest request, IServiceProvider services, HandlerRegistry registry, CancellationToken cancellationToken) { var behaviors = services.GetAll>(); return behaviors.Length == 0 - ? registry.GetRequired>(services).Handle((TRequest)request, cancellationToken) - : RunPipeline((TRequest)request, behaviors, services, registry, cancellationToken); + ? registry.GetRequired>(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 : StreamHandlerW where TRequest : IStreamRequest { public override IAsyncEnumerable Handle(IStreamRequest 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 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 HandleCore(TRequest request, IServiceProvider services, HandlerRegistry registry, CancellationToken cancellationToken) { var behaviors = services.GetAll>(); return behaviors.Length == 0 - ? registry.GetRequired>(services).Handle((TRequest)request, cancellationToken) - : RunPipeline((TRequest)request, behaviors, services, registry, cancellationToken); + ? registry.GetRequired>(services).Handle(request, cancellationToken) + : RunPipeline(request, behaviors, services, registry, cancellationToken); } private static IAsyncEnumerable RunPipeline( @@ -127,5 +150,18 @@ public override object[] ResolveHandlers(IServiceProvider services) } public override Func CreateCall(object handler, object notification, CancellationToken cancellationToken) + => Telemetry.IsEnabled ? CreateTracedCall(handler, notification, cancellationToken) : CreatePlainCall(handler, notification, cancellationToken); + + private static Func CreatePlainCall(object handler, object notification, CancellationToken cancellationToken) => () => ((INotificationHandler)handler).Handle((TNotification)notification, cancellationToken); + + private static readonly ConcurrentDictionary HandlerSpans = new(); + + // One child span per handler, under the publish span. + private static Func 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)s.handler).Handle((TNotification)s.notification, s.cancellationToken)); + } } diff --git a/src/Directory.Build.props b/src/Directory.Build.props index 2e79397..8043709 100644 --- a/src/Directory.Build.props +++ b/src/Directory.Build.props @@ -3,7 +3,7 @@ net8.0;net10.0 $(MSBuildProjectName) - 2.1.1 + 2.2.0 Andrea Ficarra Copyright © 2026 Andrea Ficarra AnyMediator (formerly OpenMediatR) is a free, open-source (Apache-2.0) in-process mediator for .NET and a MediatR alternative with a MediatR-compatible API: requests, streaming requests, notifications, pipeline behaviors, pre/post-processors, exception handlers/actions and pluggable notification sinks. Independent project, not affiliated with or endorsed by MediatR or Lucky Penny Software. @@ -12,10 +12,12 @@ git Apache-2.0 -2.1.1: performance. No API or behavior change. -- Requests, streams and notifications are dispatched through generic wrappers built once per message type, without reflection. Send with no behaviors: 457 ns / 360 B in 2.1.0, 97 ns / 96 B now (MediatR 12.5: 92 ns / 200 B). Publish with one handler: 379 ns / 656 B in 2.1.0, 196 ns / 344 B now. -- Simpler runtime library: about 20% less code. -- Benchmarks against MediatR and Mediator: https://adiakys.github.io/AnyMediator/guide/performance +2.2.0: tracing and metrics. No behavior change. +- OpenTelemetry-compatible spans for every Send ("send {Request}"), CreateStream ("create_stream {Request}") and Publish ("publish {Notification}"), with one "handle {Notification}" child span per notification handler. Failures set Error status, error.type and an exception event. +- Histogram anymediator.operation.duration (seconds) by operation, message type and error type. +- New AnyMediatorDiagnostics.ActivitySourceName and MeterName to subscribe: AddSource(...) and AddMeter(...) in OpenTelemetry. Uses .NET's ActivitySource and Meter, no new dependency. +- No cost when nobody listens; about 0.5 µs per span when recording. +- Guide: https://adiakys.github.io/AnyMediator/guide/tracing mediator;anymediator;openmediatr;mediatr;mediatr-alternative;cqrs;pipeline;notifications;request-response;streaming;messaging;dotnet;open-source;apache-2.0 README.md diff --git a/tests/AnyMediator.PerformanceTests/TracingBenchmarks.cs b/tests/AnyMediator.PerformanceTests/TracingBenchmarks.cs new file mode 100644 index 0000000..05758ef --- /dev/null +++ b/tests/AnyMediator.PerformanceTests/TracingBenchmarks.cs @@ -0,0 +1,69 @@ +using System.Diagnostics; +using System.Diagnostics.Metrics; +using BenchmarkDotNet.Attributes; +using Microsoft.Extensions.DependencyInjection; + +namespace AnyMediator.PerformanceTests; + +// The cost of tracing: the same Send and Publish with nobody listening, and with a listener that records every span +// and every measurement, as an OpenTelemetry SDK sampling 100% does (minus the exporter's own work). +[MemoryDiagnoser] +public class TracingBenchmarks +{ + private ISender _sender = null!; + private IPublisher _publisher = null!; + private ActivityListener? _activityListener; + private MeterListener? _meterListener; + private readonly BenchmarkRequest _request = new(); + private readonly BenchmarkNotification _notification = new(); + + [Params(false, true)] + public bool Listening { get; set; } + + [GlobalSetup] + public void Setup() + { + if (Listening) + { + _activityListener = new ActivityListener + { + ShouldListenTo = source => source.Name == AnyMediatorDiagnostics.ActivitySourceName, + Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllDataAndRecorded, + }; + ActivitySource.AddActivityListener(_activityListener); + + _meterListener = new MeterListener + { + InstrumentPublished = (instrument, listener) => + { + if (instrument.Meter.Name == AnyMediatorDiagnostics.MeterName) + listener.EnableMeasurementEvents(instrument); + }, + }; + _meterListener.SetMeasurementEventCallback((_, _, _, _) => { }); + _meterListener.Start(); + } + + var provider = new ServiceCollection() + .AddAnyMediator(cfg => cfg.RegisterServicesFromAssemblyContaining()) + .BuildServiceProvider(); + _sender = provider.GetRequiredService(); + _publisher = provider.GetRequiredService(); + + Send().GetAwaiter().GetResult(); + Publish().GetAwaiter().GetResult(); + } + + [GlobalCleanup] + public void Cleanup() + { + _activityListener?.Dispose(); + _meterListener?.Dispose(); + } + + [Benchmark] + public Task Send() => _sender.Send(_request); + + [Benchmark] + public Task Publish() => _publisher.Publish(_notification); +} diff --git a/tests/AnyMediator.Tests/TelemetryTests.cs b/tests/AnyMediator.Tests/TelemetryTests.cs new file mode 100644 index 0000000..f7a14b9 --- /dev/null +++ b/tests/AnyMediator.Tests/TelemetryTests.cs @@ -0,0 +1,152 @@ +using System.Collections.Concurrent; +using System.Diagnostics; +using System.Diagnostics.Metrics; +using System.Runtime.CompilerServices; +using FluentAssertions; +using Microsoft.Extensions.DependencyInjection; + +namespace AnyMediator.Tests; + +public sealed class TelemetryTests : IDisposable +{ + // Listeners are process-wide and tests run in parallel: each test filters on message types of its own. + private readonly ConcurrentBag _activities = []; + private readonly ConcurrentBag<(double Value, Dictionary Tags)> _measurements = []; + private readonly ActivityListener _activityListener; + private readonly MeterListener _meterListener = new(); + + public TelemetryTests() + { + _activityListener = new ActivityListener + { + ShouldListenTo = source => source.Name == AnyMediatorDiagnostics.ActivitySourceName, + Sample = (ref ActivityCreationOptions _) => ActivitySamplingResult.AllDataAndRecorded, + ActivityStopped = _activities.Add, + }; + ActivitySource.AddActivityListener(_activityListener); + + _meterListener.InstrumentPublished = (instrument, listener) => + { + if (instrument.Meter.Name == AnyMediatorDiagnostics.MeterName) + listener.EnableMeasurementEvents(instrument); + }; + _meterListener.SetMeasurementEventCallback((_, value, tags, _) => + _measurements.Add((value, tags.ToArray().ToDictionary(t => t.Key, t => t.Value)))); + _meterListener.Start(); + } + + public void Dispose() + { + _activityListener.Dispose(); + _meterListener.Dispose(); + } + + private static IServiceProvider BuildProvider() + { + var services = new ServiceCollection(); + services.AddAnyMediator(cfg => cfg.RegisterServicesFromAssemblyContaining()); + return services.BuildServiceProvider(); + } + + private Activity[] ActivitiesFor() + => _activities.Where(a => (string?)a.GetTagItem("anymediator.message.type") == typeof(TMessage).FullName).ToArray(); + + [Fact] + public async Task Send_RecordsSpanAndDuration() + { + var response = await BuildProvider().GetRequiredService().Send(new TracedRequest()); + + response.Should().Be("traced"); + var span = ActivitiesFor().Should().ContainSingle().Subject; + span.DisplayName.Should().Be("send TracedRequest"); + span.GetTagItem("anymediator.operation").Should().Be("send"); + span.Status.Should().Be(ActivityStatusCode.Unset); + + _measurements.Should().Contain(m => + Equals(m.Tags["anymediator.message.type"], typeof(TracedRequest).FullName) + && Equals(m.Tags["anymediator.operation"], "send") + && !m.Tags.ContainsKey("error.type")); + } + + [Fact] + public async Task Send_WhenHandlerThrows_RecordsErrorOnSpanAndDuration() + { + var act = () => BuildProvider().GetRequiredService().Send(new FailingTracedRequest()); + + await act.Should().ThrowAsync(); + var span = ActivitiesFor().Should().ContainSingle().Subject; + span.Status.Should().Be(ActivityStatusCode.Error); + span.GetTagItem("error.type").Should().Be(typeof(TimeoutException).FullName); + span.Events.Should().ContainSingle(e => e.Name == "exception"); + + _measurements.Should().Contain(m => + Equals(m.Tags["anymediator.message.type"], typeof(FailingTracedRequest).FullName) + && Equals(m.Tags["error.type"], typeof(TimeoutException).FullName)); + } + + [Fact] + public async Task Publish_RecordsPublishSpanWithOneChildPerHandler() + { + await BuildProvider().GetRequiredService().Publish(new TracedEvent()); + + var spans = ActivitiesFor(); + var publish = spans.Should().ContainSingle(a => a.DisplayName == "publish TracedEvent").Subject; + var handlers = spans.Where(a => a.DisplayName == "handle TracedEvent").ToArray(); + + handlers.Should().HaveCount(2).And.OnlyContain(h => h.ParentSpanId == publish.SpanId); + handlers.Select(h => h.GetTagItem("anymediator.handler.type")) + .Should().BeEquivalentTo(new[] { typeof(FirstTracedEventHandler).FullName, typeof(SecondTracedEventHandler).FullName }); + } + + [Fact] + public async Task CreateStream_SpanLastsUntilEnumerationEnds() + { + var items = new List(); + await foreach (var item in BuildProvider().GetRequiredService().CreateStream(new TracedStreamRequest())) + { + items.Add(item); + ActivitiesFor().Should().BeEmpty("the span is still open while items are read"); + } + + items.Should().Equal(1, 2); + ActivitiesFor().Should().ContainSingle().Which.DisplayName.Should().Be("create_stream TracedStreamRequest"); + } + + private sealed record TracedRequest : IRequest; + + private sealed class TracedRequestHandler : IRequestHandler + { + public Task Handle(TracedRequest request, CancellationToken cancellationToken) => Task.FromResult("traced"); + } + + private sealed record FailingTracedRequest : IRequest; + + private sealed class FailingTracedRequestHandler : IRequestHandler + { + public Task Handle(FailingTracedRequest request, CancellationToken cancellationToken) => throw new TimeoutException("too slow"); + } + + private sealed record TracedEvent : INotification; + + private sealed class FirstTracedEventHandler : INotificationHandler + { + public Task Handle(TracedEvent notification, CancellationToken cancellationToken) => Task.CompletedTask; + } + + private sealed class SecondTracedEventHandler : INotificationHandler + { + public Task Handle(TracedEvent notification, CancellationToken cancellationToken) => Task.CompletedTask; + } + + private sealed record TracedStreamRequest : IStreamRequest; + + private sealed class TracedStreamHandler : IStreamRequestHandler + { + public async IAsyncEnumerable Handle(TracedStreamRequest request, [EnumeratorCancellation] CancellationToken cancellationToken) + { + yield return 1; + await Task.Yield(); + yield return 2; + } + } +}