diff --git a/docs/mediator/adr/0005-opentelemetry-instrumentation.md b/docs/mediator/adr/0005-opentelemetry-instrumentation.md index eff6373..38f9e57 100644 --- a/docs/mediator/adr/0005-opentelemetry-instrumentation.md +++ b/docs/mediator/adr/0005-opentelemetry-instrumentation.md @@ -17,7 +17,7 @@ description: "Design of the OpenTelemetry companion package: automatic distribut ## Status -**Released in v1.0.0** +**Released in v1.0.0** — amended in v1.1.0-rc.2 (see [Amendment](#amendment--v110-rc2-database-enrichment--factory-based-metrics) below). ## Context @@ -537,9 +537,36 @@ added if there is demand, but the default is full-enumeration spans. --- +## Amendment — v1.1.0-rc.2 (database enrichment + factory-based metrics) + +Three capabilities were added after the original release. All are additive: `AddMediatorInstrumentation()` is unchanged at the call site, so existing consumers get them with no code change. + +### A1. Automatic database-span enrichment + +**Decision.** `AddMediatorInstrumentation()` on the `TracerProviderBuilder` now also registers a `BaseProcessor` (`DatabaseSpanEnrichmentProcessor`). For any database client span (one carrying `db.system`) flowing through the same provider, it derives a redaction-safe `db.operation.name`, `db.sql.table` and `db.stored_procedure.name` from `db.statement` / `db.query.text` — but only when the underlying driver emitted just the raw statement (e.g. an older Npgsql). + +**Why.** Without those structured attributes, every query on a connection collapses into one `"{system} → {host}"` dependency in the Pipeline Explorer flame. Parsing happens **in-process** — where the application owns its own SQL — via `SqlStatementParser`, and only the verb plus a single bare identifier are surfaced; the raw statement, parameters and row values are never copied onto the span. This lets the trace consumer (the Pipeline Explorer importer) keep its "never read `db.statement`" boundary while still attributing time per operation (`SELECT` vs `INSERT` vs a stored procedure by name). + +**Properties.** Strictly additive — an attribute already supplied by native instrumentation is never overwritten. Gated on a cheap `db.system` check, so non-DB spans are untouched. AOT/trim-safe (no reflection). New files: `DatabaseSpanEnrichmentProcessor.cs`, `SqlStatementParser.cs`. + +### A2. Metrics created from `IMeterFactory` (replaces §3.2's static `Meter`) + +**Decision.** The three instruments (§3.2) are now created from the DI `IMeterFactory` via a singleton `MediatorMetrics`, registered by `AddMediatorInstrumentation()` — which also calls `services.AddMetrics()` so the factory is present even outside a Generic Host. The meter name is unchanged (`"DSoftStudio.Mediator"`), so `AddMeter("DSoftStudio.Mediator")` still subscribes. The `ActivitySource` stays static + named (the .NET convention — there is no per-DI ActivitySource factory). + +**Why.** Microsoft prescribes `IMeterFactory` for DI-aware libraries; a static `Meter` is an anti-pattern because it cannot be isolated per service collection, which cross-contaminates parallel tests and multiple hosts in one process. Adds a `Microsoft.Extensions.Diagnostics` package reference for `AddMetrics()`. New file: `MediatorMetrics.cs`. + +### A3. Histogram bucket advice + +**Decision.** `mediator.request.duration` carries explicit sub-second `InstrumentAdvice` bucket boundaries (`0.0005 … 5 s`). + +**Why.** The OpenTelemetry default histogram buckets (`[0, 5, 10, 25, …] s`) collapse every sub-5-second request into the first bucket, making p95/p99 meaningless for millisecond-scale mediator requests. The advice is honoured by the OpenTelemetry .NET SDK (≥ 1.10) as the default boundaries. + +--- + ## Document History | Date | Version | Changes | |------------|---------|---------| | — | Draft | Initial ADR with instrumentation design | | 2026-03-15 | v1.0.0 | Released as DSoftStudio.Mediator.OpenTelemetry companion package | +| 2026-06-22 | v1.1.0-rc.2 | Amendment: automatic database-span enrichment, `IMeterFactory`-based metrics, sub-second histogram bucket advice | diff --git a/docs/mediator/integrations/opentelemetry.md b/docs/mediator/integrations/opentelemetry.md index 4a7fd46..8c77905 100644 --- a/docs/mediator/integrations/opentelemetry.md +++ b/docs/mediator/integrations/opentelemetry.md @@ -47,7 +47,8 @@ builder.Services.AddOpenTelemetry() | Signal | Details | |---|---| | **Tracing** | One span per `Send()`, `Publish()`, `CreateStream()` with CQRS-aware naming (`CreateUser command`, `GetUsers query`). Per-handler child spans for notifications. | -| **Metrics** | `mediator.request.duration` (histogram), `mediator.request.active` (up-down counter), `mediator.request.errors` (counter with `error.type` tag). | +| **Metrics** | `mediator.request.duration` (histogram, **seconds**, with sub-second bucket boundaries), `mediator.request.active` (up-down counter), `mediator.request.errors` (counter with `error.type` tag). Instruments are created from the DI `IMeterFactory`. | +| **Database enrichment** | Calling `AddMediatorInstrumentation()` on the `TracerProviderBuilder` also tags database spans with a redaction-safe `db.operation.name` / `db.sql.table` / `db.stored_procedure.name`, so each query (`SELECT`, `INSERT`, `CALL`…) is a distinct dependency instead of one aggregated row. No configuration. | | **Zero-cost when unused** | `HasListeners()` short-circuit adds ~1 ns when no exporter is configured. | ## Configuration diff --git a/src/DSoftStudio.Mediator.OpenTelemetry/DSoftStudio.Mediator.OpenTelemetry.csproj b/src/DSoftStudio.Mediator.OpenTelemetry/DSoftStudio.Mediator.OpenTelemetry.csproj index 9d1f008..99d0063 100644 --- a/src/DSoftStudio.Mediator.OpenTelemetry/DSoftStudio.Mediator.OpenTelemetry.csproj +++ b/src/DSoftStudio.Mediator.OpenTelemetry/DSoftStudio.Mediator.OpenTelemetry.csproj @@ -9,7 +9,7 @@ DSoftStudio.Mediator.OpenTelemetry - 1.1.0-rc.1 + 1.1.0-rc.2 DSoftStudio DSoftStudio @@ -66,6 +66,9 @@ + + diff --git a/src/DSoftStudio.Mediator.OpenTelemetry/DatabaseSpanEnrichmentProcessor.cs b/src/DSoftStudio.Mediator.OpenTelemetry/DatabaseSpanEnrichmentProcessor.cs new file mode 100644 index 0000000..db28dc5 --- /dev/null +++ b/src/DSoftStudio.Mediator.OpenTelemetry/DatabaseSpanEnrichmentProcessor.cs @@ -0,0 +1,62 @@ +// Copyright (c) DSoftStudio. All rights reserved. +// Licensed under the MIT License. See LICENSE in the project root for license information. + +using System.Diagnostics; + +namespace DSoftStudio.Mediator.OpenTelemetry; + +/// +/// Enriches database client spans (those carrying db.system) with a redaction-safe +/// db.operation.name and db.sql.table derived from the SQL statement, so that +/// downstream tooling (e.g. the Pipeline Explorer) can attribute time to the specific +/// operation — distinguishing a SELECT from an INSERT on the same connection instead +/// of collapsing every query into a single "{system} → {host}" dependency row. +/// +/// +/// +/// Registered automatically by AddMediatorInstrumentation() on the +/// ; no configuration is required. +/// It runs in-process, where the application owns its own SQL, so reading db.statement here +/// never crosses a trust boundary — and only the verb and a single bare table identifier are +/// copied onto the span. The raw statement, parameters and row values are never propagated, which +/// keeps the import-side contract intact (the trace consumer never has to read db.statement). +/// +/// +/// The enrichment is strictly additive: an attribute already supplied by native instrumentation +/// (a newer Npgsql / EF Core that emits db.operation.name) is never overwritten. +/// +/// +internal sealed class DatabaseSpanEnrichmentProcessor : global::OpenTelemetry.BaseProcessor +{ + public override void OnEnd(Activity activity) + { + // Cheap gate: only database client spans carry db.system / db.system.name. + if (activity.GetTagItem("db.system") is null && activity.GetTagItem("db.system.name") is null) + return; + + bool hasOperation = activity.GetTagItem("db.operation.name") is not null + || activity.GetTagItem("db.operation") is not null; + bool hasTable = activity.GetTagItem("db.sql.table") is not null + || activity.GetTagItem("db.collection.name") is not null; + bool hasProcedure = activity.GetTagItem("db.stored_procedure.name") is not null; + + if (hasOperation && hasTable && hasProcedure) + return; // Native instrumentation already described the operation — nothing to add. + + var statement = (activity.GetTagItem("db.query.text") as string) + ?? (activity.GetTagItem("db.statement") as string); + if (string.IsNullOrWhiteSpace(statement)) + return; + + if (!hasOperation && SqlStatementParser.Operation(statement) is { } operation) + activity.SetTag("db.operation.name", operation); + + // DML target table (SELECT/INSERT/UPDATE/DELETE) — distinguishes queries on different tables. + if (!hasTable && SqlStatementParser.Table(statement) is { } table) + activity.SetTag("db.sql.table", table); + + // Stored-procedure / function name (CALL/EXEC) — distinguishes one procedure from another. + if (!hasProcedure && SqlStatementParser.Procedure(statement) is { } procedure) + activity.SetTag("db.stored_procedure.name", procedure); + } +} diff --git a/src/DSoftStudio.Mediator.OpenTelemetry/InstrumentedNotificationPublisher.cs b/src/DSoftStudio.Mediator.OpenTelemetry/InstrumentedNotificationPublisher.cs index 06aa844..43021d7 100644 --- a/src/DSoftStudio.Mediator.OpenTelemetry/InstrumentedNotificationPublisher.cs +++ b/src/DSoftStudio.Mediator.OpenTelemetry/InstrumentedNotificationPublisher.cs @@ -11,7 +11,7 @@ namespace DSoftStudio.Mediator.OpenTelemetry; /// Decorator that wraps an with distributed tracing /// and metrics. Creates a parent span for the publish operation and per-handler child spans. /// -internal sealed class InstrumentedNotificationPublisher(INotificationPublisher inner, MediatorInstrumentationOptions options) : INotificationPublisher +internal sealed class InstrumentedNotificationPublisher(INotificationPublisher inner, MediatorInstrumentationOptions options, MediatorMetrics? metrics) : INotificationPublisher { private static readonly ActivitySource Source = MediatorInstrumentation.ActivitySource; private static readonly ConcurrentDictionary HandlerSpanNames = new(); @@ -23,7 +23,7 @@ public async Task Publish( where TNotification : INotification { bool tracingActive = options.EnableTracing && Source.HasListeners(); - bool metricsActive = options.EnableMetrics && MediatorInstrumentation.RequestDuration.Enabled; + bool metricsActive = options.EnableMetrics && metrics is not null && metrics.RequestDuration.Enabled; if (!tracingActive && !metricsActive) { @@ -65,7 +65,7 @@ public async Task Publish( { "mediator.request.kind", MediatorNotificationMetadata.RequestKind } }; - MediatorInstrumentation.RequestActive.Add(1, metricTags); + metrics!.RequestActive.Add(1, metricTags); startTimestamp = Stopwatch.GetTimestamp(); } @@ -96,7 +96,7 @@ public async Task Publish( { "error.type", ex.GetType().FullName! } }; - MediatorInstrumentation.RequestErrors.Add(1, errorTags); + metrics!.RequestErrors.Add(1, errorTags); } throw; @@ -108,8 +108,8 @@ public async Task Publish( if (metricsActive) { var elapsed = Stopwatch.GetElapsedTime(startTimestamp); - MediatorInstrumentation.RequestDuration.Record(elapsed.TotalSeconds, metricTags); - MediatorInstrumentation.RequestActive.Add(-1, metricTags); + metrics!.RequestDuration.Record(elapsed.TotalSeconds, metricTags); + metrics!.RequestActive.Add(-1, metricTags); } } } diff --git a/src/DSoftStudio.Mediator.OpenTelemetry/MediatorInstrumentation.cs b/src/DSoftStudio.Mediator.OpenTelemetry/MediatorInstrumentation.cs index 7659f43..57eadee 100644 --- a/src/DSoftStudio.Mediator.OpenTelemetry/MediatorInstrumentation.cs +++ b/src/DSoftStudio.Mediator.OpenTelemetry/MediatorInstrumentation.cs @@ -2,38 +2,25 @@ // Licensed under the MIT License. See LICENSE in the project root for license information. using System.Diagnostics; -using System.Diagnostics.Metrics; namespace DSoftStudio.Mediator.OpenTelemetry; /// -/// Provides the and used by the mediator instrumentation. +/// Provides the used by the mediator instrumentation, plus the shared name/version. +/// The metric instruments live on (created from the DI IMeterFactory). /// public static class MediatorInstrumentation { /// - /// The name used for both the and . + /// The name used for both the and the metrics Meter. /// Use this constant when manually calling AddSource() or AddMeter(). /// public const string SourceName = "DSoftStudio.Mediator"; - private static readonly string Version = typeof(MediatorInstrumentation) + /// The instrumentation version, stamped onto the and the metrics meter. + internal static readonly string Version = typeof(MediatorInstrumentation) .Assembly.GetName().Version?.ToString() ?? "0.0.0"; + // The ActivitySource stays static + named (the .NET convention — there is no per-DI ActivitySource factory). internal static readonly ActivitySource ActivitySource = new(SourceName, Version); - internal static readonly Meter Meter = new(SourceName, Version); - - // ── Metric instruments ───────────────────────────────────────────── - - internal static readonly Histogram RequestDuration = - Meter.CreateHistogram("mediator.request.duration", "s", - "Time from behavior entry to handler completion"); - - internal static readonly UpDownCounter RequestActive = - Meter.CreateUpDownCounter("mediator.request.active", "{request}", - "Number of in-flight requests"); - - internal static readonly Counter RequestErrors = - Meter.CreateCounter("mediator.request.errors", "{error}", - "Count of failed requests"); } diff --git a/src/DSoftStudio.Mediator.OpenTelemetry/MediatorMetrics.cs b/src/DSoftStudio.Mediator.OpenTelemetry/MediatorMetrics.cs new file mode 100644 index 0000000..1953d58 --- /dev/null +++ b/src/DSoftStudio.Mediator.OpenTelemetry/MediatorMetrics.cs @@ -0,0 +1,56 @@ +// Copyright (c) DSoftStudio. All rights reserved. +// Licensed under the MIT License. See LICENSE in the project root for license information. + +using System.Diagnostics.Metrics; + +namespace DSoftStudio.Mediator.OpenTelemetry; + +/// +/// Owns the mediator metric instruments. The is created from the DI +/// — the pattern Microsoft prescribes for a DI-aware library, because a +/// static cannot be isolated per service collection (it leaks measurements across +/// parallel tests and across hosts in the same process). Registered as a singleton by +/// AddMediatorInstrumentation(); the meter name is , so an +/// app still subscribes with the same AddMeter("DSoftStudio.Mediator") call. +/// +public sealed class MediatorMetrics +{ + // Explicit sub-second bucket boundaries (in SECONDS) so the duration histogram yields meaningful p50/p95/p99 + // for millisecond-scale mediator requests. Without this, the OpenTelemetry default buckets ([0, 5, 10, 25, …] + // seconds) collapse every sub-5-second request into the first bucket → useless percentiles. Supplied via + // InstrumentAdvice, which the OpenTelemetry .NET SDK (>= 1.10) honours as the default boundaries. + private static readonly double[] DurationSecondsBuckets = + [0.0005, 0.001, 0.0025, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1, 2.5, 5]; + + public MediatorMetrics(IMeterFactory meterFactory) + { + ArgumentNullException.ThrowIfNull(meterFactory); + + var meter = meterFactory.Create(MediatorInstrumentation.SourceName, MediatorInstrumentation.Version); + + RequestDuration = meter.CreateHistogram( + name: "mediator.request.duration", + unit: "s", + description: "Time from behavior entry to handler completion", + advice: new InstrumentAdvice { HistogramBucketBoundaries = DurationSecondsBuckets }); + + RequestActive = meter.CreateUpDownCounter( + name: "mediator.request.active", + unit: "{request}", + description: "Number of in-flight requests"); + + RequestErrors = meter.CreateCounter( + name: "mediator.request.errors", + unit: "{error}", + description: "Count of failed requests"); + } + + /// Histogram of request durations in SECONDS (record with elapsed.TotalSeconds). + public Histogram RequestDuration { get; } + + /// In-flight request count (+1 on entry, −1 on completion). + public UpDownCounter RequestActive { get; } + + /// Count of failed requests, tagged with error.type. + public Counter RequestErrors { get; } +} diff --git a/src/DSoftStudio.Mediator.OpenTelemetry/MediatorMetricsBehavior.cs b/src/DSoftStudio.Mediator.OpenTelemetry/MediatorMetricsBehavior.cs index a162924..fd0439e 100644 --- a/src/DSoftStudio.Mediator.OpenTelemetry/MediatorMetricsBehavior.cs +++ b/src/DSoftStudio.Mediator.OpenTelemetry/MediatorMetricsBehavior.cs @@ -9,7 +9,7 @@ namespace DSoftStudio.Mediator.OpenTelemetry; /// /// Pipeline behavior that records metrics (duration, active count, errors) for mediator requests. /// -public sealed class MediatorMetricsBehavior(MediatorInstrumentationOptions options) : IPipelineBehavior +public sealed class MediatorMetricsBehavior(MediatorInstrumentationOptions options, MediatorMetrics metrics) : IPipelineBehavior where TRequest : IRequest { @@ -18,7 +18,7 @@ public async ValueTask Handle( IRequestHandler next, CancellationToken cancellationToken) { - if (!options.EnableMetrics || !MediatorInstrumentation.RequestDuration.Enabled) + if (!options.EnableMetrics || !metrics.RequestDuration.Enabled) return await next.Handle(request, cancellationToken); if (options.Filter is not null && !options.Filter(typeof(TRequest))) @@ -30,7 +30,7 @@ public async ValueTask Handle( { "mediator.request.kind", MediatorTelemetryMetadata.RequestKind } }; - MediatorInstrumentation.RequestActive.Add(1, tags); + metrics.RequestActive.Add(1, tags); var startTimestamp = Stopwatch.GetTimestamp(); try @@ -46,14 +46,14 @@ public async ValueTask Handle( { "error.type", ex.GetType().FullName! } }; - MediatorInstrumentation.RequestErrors.Add(1, errorTags); + metrics.RequestErrors.Add(1, errorTags); throw; } finally { var elapsed = Stopwatch.GetElapsedTime(startTimestamp); - MediatorInstrumentation.RequestDuration.Record(elapsed.TotalSeconds, tags); - MediatorInstrumentation.RequestActive.Add(-1, tags); + metrics.RequestDuration.Record(elapsed.TotalSeconds, tags); + metrics.RequestActive.Add(-1, tags); } } } diff --git a/src/DSoftStudio.Mediator.OpenTelemetry/MediatorStreamMetricsBehavior.cs b/src/DSoftStudio.Mediator.OpenTelemetry/MediatorStreamMetricsBehavior.cs index 8b70bac..d697978 100644 --- a/src/DSoftStudio.Mediator.OpenTelemetry/MediatorStreamMetricsBehavior.cs +++ b/src/DSoftStudio.Mediator.OpenTelemetry/MediatorStreamMetricsBehavior.cs @@ -11,7 +11,7 @@ namespace DSoftStudio.Mediator.OpenTelemetry; /// Stream pipeline behavior that records metrics for streamed requests. /// Duration covers the entire enumeration lifetime. /// -public sealed class MediatorStreamMetricsBehavior(MediatorInstrumentationOptions options) : IStreamPipelineBehavior +public sealed class MediatorStreamMetricsBehavior(MediatorInstrumentationOptions options, MediatorMetrics metrics) : IStreamPipelineBehavior where TRequest : IStreamRequest { @@ -20,7 +20,7 @@ public IAsyncEnumerable Handle( IStreamRequestHandler next, CancellationToken cancellationToken) { - if (!options.EnableMetrics || !MediatorInstrumentation.RequestDuration.Enabled) + if (!options.EnableMetrics || !metrics.RequestDuration.Enabled) return next.Handle(request, cancellationToken); if (options.Filter is not null && !options.Filter(typeof(TRequest))) @@ -29,7 +29,7 @@ public IAsyncEnumerable Handle( return Instrumented(request, next, cancellationToken); } - private static async IAsyncEnumerable Instrumented( + private async IAsyncEnumerable Instrumented( TRequest request, IStreamRequestHandler next, [EnumeratorCancellation] CancellationToken cancellationToken) @@ -40,7 +40,7 @@ private static async IAsyncEnumerable Instrumented( { "mediator.request.kind", MediatorStreamMetadata.RequestKind } }; - MediatorInstrumentation.RequestActive.Add(1, tags); + metrics.RequestActive.Add(1, tags); var startTimestamp = Stopwatch.GetTimestamp(); try @@ -53,8 +53,8 @@ private static async IAsyncEnumerable Instrumented( finally { var elapsed = Stopwatch.GetElapsedTime(startTimestamp); - MediatorInstrumentation.RequestDuration.Record(elapsed.TotalSeconds, tags); - MediatorInstrumentation.RequestActive.Add(-1, tags); + metrics.RequestDuration.Record(elapsed.TotalSeconds, tags); + metrics.RequestActive.Add(-1, tags); } } } diff --git a/src/DSoftStudio.Mediator.OpenTelemetry/MediatorStreamTracingBehavior.cs b/src/DSoftStudio.Mediator.OpenTelemetry/MediatorStreamTracingBehavior.cs index fa35d7c..3d4285a 100644 --- a/src/DSoftStudio.Mediator.OpenTelemetry/MediatorStreamTracingBehavior.cs +++ b/src/DSoftStudio.Mediator.OpenTelemetry/MediatorStreamTracingBehavior.cs @@ -52,16 +52,35 @@ private async IAsyncEnumerable Instrumented( } bool success = false; + // Per-item production metrics — measured here (the span already wraps the full enumeration) so an imported + // trace can populate the profiler's STREAM TELEMETRY *production* block (items / TTFI / throughput), not + // just lifecycle + duration. Stopwatch.GetTimestamp() math keeps this allocation-free and TFM-agnostic. + long itemCount = 0; + long startTimestamp = Stopwatch.GetTimestamp(); + long firstItemTimestamp = 0; try { await foreach (var item in next.Handle(request, cancellationToken).WithCancellation(cancellationToken)) { + if (itemCount == 0) + firstItemTimestamp = Stopwatch.GetTimestamp(); + itemCount++; yield return item; } success = true; } finally { + if (activity is { IsAllDataRequested: true }) + { + double freq = Stopwatch.Frequency; + double elapsedMs = (Stopwatch.GetTimestamp() - startTimestamp) * 1000.0 / freq; + double firstItemMs = firstItemTimestamp > 0 ? (firstItemTimestamp - startTimestamp) * 1000.0 / freq : 0.0; + double throughputPerSec = elapsedMs > 0 ? itemCount * 1000.0 / elapsedMs : 0.0; + activity.SetTag("mediator.stream.item_count", itemCount); + activity.SetTag("mediator.stream.first_item_ms", firstItemMs); + activity.SetTag("mediator.stream.throughput_per_sec", throughputPerSec); + } activity?.SetStatus(success ? ActivityStatusCode.Ok : ActivityStatusCode.Error); } } diff --git a/src/DSoftStudio.Mediator.OpenTelemetry/README.md b/src/DSoftStudio.Mediator.OpenTelemetry/README.md index 2685573..b9fe208 100644 --- a/src/DSoftStudio.Mediator.OpenTelemetry/README.md +++ b/src/DSoftStudio.Mediator.OpenTelemetry/README.md @@ -10,7 +10,8 @@ OpenTelemetry instrumentation for [DSoftStudio.Mediator](https://www.nuget.org/p ## Features - **Distributed tracing** — Activity spans for `Send`, `Publish`, and `CreateStream` with semantic attributes -- **Metrics** — Histogram metrics for request duration and counters for operations +- **Metrics** — `mediator.request.duration` histogram (seconds, with sub-second buckets) plus `mediator.request.active` / `mediator.request.errors` counters, created from the DI `IMeterFactory` +- **Database dependency enrichment** — automatically tags database spans with a redaction-safe `db.operation.name` / `db.sql.table` / `db.stored_procedure.name`, so each query (`SELECT`, `INSERT`, `CALL`…) shows as its own dependency instead of one aggregated row — no configuration - **Configurable filtering** — Include or exclude specific request types via `MediatorInstrumentationOptions` - **Zero configuration** — Works out of the box with any OpenTelemetry exporter @@ -40,8 +41,15 @@ services ```csharp services.AddMediatorInstrumentation(options => { - options.RecordMetrics = true; + // Skip noisy request types (e.g. health checks). options.Filter = type => !type.Name.Contains("HealthCheck"); + + // Tracing and metrics are both ON by default — turn one off if you only want the other. + // options.EnableTracing = false; + // options.EnableMetrics = false; + + // Keep error.type on the span but drop the (verbose) exception stack trace. + options.RecordExceptionStackTraces = false; }); ``` diff --git a/src/DSoftStudio.Mediator.OpenTelemetry/ServiceCollectionExtensions.cs b/src/DSoftStudio.Mediator.OpenTelemetry/ServiceCollectionExtensions.cs index 6d037f6..35cb1fa 100644 --- a/src/DSoftStudio.Mediator.OpenTelemetry/ServiceCollectionExtensions.cs +++ b/src/DSoftStudio.Mediator.OpenTelemetry/ServiceCollectionExtensions.cs @@ -35,6 +35,12 @@ public static IServiceCollection AddMediatorInstrumentation( if (options.EnableMetrics) { + // The instruments are created from the DI IMeterFactory (Microsoft's prescribed pattern for a + // DI-aware library — a static Meter cannot be isolated per service collection). AddMetrics() is + // idempotent and registers the default IMeterFactory when the host has not already done so. + services.AddMetrics(); + services.TryAddSingleton(); + services.AddTransient(typeof(IPipelineBehavior<,>), typeof(MediatorMetricsBehavior<,>)); services.AddTransient(typeof(IStreamPipelineBehavior<,>), typeof(MediatorStreamMetricsBehavior<,>)); } @@ -50,7 +56,11 @@ public static IServiceCollection AddMediatorInstrumentation( services.AddSingleton(sp => { var inner = ResolveInnerPublisher(sp, existingDescriptor); - return new InstrumentedNotificationPublisher(inner, sp.GetRequiredService()); + // MediatorMetrics is only registered when metrics are enabled — null here means tracing-only. + return new InstrumentedNotificationPublisher( + inner, + sp.GetRequiredService(), + sp.GetService()); }); } diff --git a/src/DSoftStudio.Mediator.OpenTelemetry/SqlStatementParser.cs b/src/DSoftStudio.Mediator.OpenTelemetry/SqlStatementParser.cs new file mode 100644 index 0000000..30a36c2 --- /dev/null +++ b/src/DSoftStudio.Mediator.OpenTelemetry/SqlStatementParser.cs @@ -0,0 +1,238 @@ +// Copyright (c) DSoftStudio. All rights reserved. +// Licensed under the MIT License. See LICENSE in the project root for license information. + +using System.Text; + +namespace DSoftStudio.Mediator.OpenTelemetry; + +/// +/// Extracts the redaction-safe shape of a SQL statement — its leading operation +/// (SELECT/INSERT/…) and target table — without ever surfacing parameters, +/// predicates or row values. +/// +/// +/// +/// Used by to derive db.operation.name / +/// db.sql.table from db.statement when the underlying database instrumentation +/// (e.g. an older Npgsql) only emits the raw statement. This runs in-process, where the +/// application owns its own SQL; only the verb and a single bare identifier are read out — never +/// the statement text itself. +/// +/// +/// The scanner tokenises at parenthesis depth zero, skipping line/block comments, single-quoted +/// and dollar-quoted string literals, and quoted identifiers, so a keyword appearing inside a +/// sub-select, a string or a comment never masquerades as the top-level operation. +/// +/// +internal static class SqlStatementParser +{ + private static readonly HashSet Verbs = new(StringComparer.OrdinalIgnoreCase) + { + "SELECT", "INSERT", "UPDATE", "DELETE", "MERGE", "CALL", "EXEC", "EXECUTE", + }; + + /// The canonical upper-case operation verb, or null if none can be identified. + public static string? Operation(string? sql) + { + if (string.IsNullOrWhiteSpace(sql)) return null; + + foreach (var token in DepthZeroTokens(sql)) + { + if (Verbs.Contains(token)) + return Canonical(token); + } + return null; + } + + /// + /// The single target table/identifier the operation reads or writes, or null when it + /// cannot be determined unambiguously (multi-table joins return the first source table only). + /// + public static string? Table(string? sql) + { + if (string.IsNullOrWhiteSpace(sql)) return null; + + var tokens = DepthZeroTokens(sql); + + int opIndex = -1; + string? op = null; + for (int i = 0; i < tokens.Count; i++) + { + if (Verbs.Contains(tokens[i])) { opIndex = i; op = Canonical(tokens[i]); break; } + } + if (op is null) return null; + + // UPDATE: the table is the identifier immediately after the verb. + if (op == "UPDATE") + return QualifiedNameAfter(tokens, opIndex); + + var anchor = op switch + { + "SELECT" or "DELETE" => "FROM", + "INSERT" or "MERGE" => "INTO", + _ => null, // CALL/EXECUTE have no table. + }; + if (anchor is null) return null; + + for (int i = opIndex + 1; i < tokens.Count; i++) + { + if (string.Equals(tokens[i], anchor, StringComparison.OrdinalIgnoreCase)) + return QualifiedNameAfter(tokens, i); + } + return null; + } + + /// + /// The invoked stored-procedure / function name for a CALL / EXEC(UTE) statement, + /// or null when the statement is not a procedure call (or the name is dynamic). + /// + public static string? Procedure(string? sql) + { + if (string.IsNullOrWhiteSpace(sql)) return null; + + var tokens = DepthZeroTokens(sql); + for (int i = 0; i < tokens.Count; i++) + { + if (tokens[i].Equals("CALL", StringComparison.OrdinalIgnoreCase) + || tokens[i].Equals("EXEC", StringComparison.OrdinalIgnoreCase) + || tokens[i].Equals("EXECUTE", StringComparison.OrdinalIgnoreCase)) + { + return QualifiedNameAfter(tokens, i); + } + } + return null; + } + + private static string Canonical(string verb) + => verb.Equals("EXEC", StringComparison.OrdinalIgnoreCase) + ? "EXECUTE" + : verb.ToUpperInvariant(); + + /// + /// Reads the (possibly schema-qualified) identifier that follows + /// and returns its final segment — e.g. public . orders → orders. + /// + private static string? QualifiedNameAfter(List tokens, int keywordIndex) + { + int i = keywordIndex + 1; + if (i >= tokens.Count || tokens[i] == ".") return null; + + var last = tokens[i]; + i++; + while (i + 1 < tokens.Count && tokens[i] == ".") + { + last = tokens[i + 1]; + i += 2; + } + return last == "." ? null : last; + } + + /// + /// Yields word tokens and bare . separators that sit at parenthesis depth zero, + /// skipping comments, string literals and tracking quoted identifiers as single words. + /// + private static List DepthZeroTokens(string sql) + { + var tokens = new List(); + int depth = 0; + int n = sql.Length; + + for (int i = 0; i < n;) + { + char c = sql[i]; + + if (char.IsWhiteSpace(c)) { i++; continue; } + + // Line comment: -- … + if (c == '-' && i + 1 < n && sql[i + 1] == '-') + { + i += 2; + while (i < n && sql[i] != '\n') i++; + continue; + } + + // Block comment: /* … */ + if (c == '/' && i + 1 < n && sql[i + 1] == '*') + { + i += 2; + while (i + 1 < n && !(sql[i] == '*' && sql[i + 1] == '/')) i++; + i += 2; + continue; + } + + // Single-quoted string literal (with '' escape). + if (c == '\'') + { + i++; + while (i < n) + { + if (sql[i] == '\'') + { + if (i + 1 < n && sql[i + 1] == '\'') { i += 2; continue; } + i++; break; + } + i++; + } + continue; + } + + // Dollar-quoted string literal (PostgreSQL): $tag$ … $tag$ + if (c == '$') + { + int tagEnd = i + 1; + while (tagEnd < n && (char.IsLetterOrDigit(sql[tagEnd]) || sql[tagEnd] == '_')) tagEnd++; + if (tagEnd < n && sql[tagEnd] == '$') + { + var tag = sql.Substring(i, tagEnd - i + 1); + int close = sql.IndexOf(tag, tagEnd + 1, StringComparison.Ordinal); + i = close < 0 ? n : close + tag.Length; + continue; + } + i++; // Lone '$' — treat as punctuation. + continue; + } + + // Quoted identifier: "…" (with "" escape) — a single identifier token. + if (c == '"') + { + i++; + var sb = new StringBuilder(); + while (i < n) + { + if (sql[i] == '"') + { + if (i + 1 < n && sql[i + 1] == '"') { sb.Append('"'); i += 2; continue; } + i++; break; + } + sb.Append(sql[i]); i++; + } + if (depth == 0 && sb.Length > 0) tokens.Add(sb.ToString()); + continue; + } + + if (c == '(') { depth++; i++; continue; } + if (c == ')') { if (depth > 0) depth--; i++; continue; } + + if (c == '.') + { + if (depth == 0) tokens.Add("."); + i++; + continue; + } + + // Bare word (keyword / identifier). + if (char.IsLetter(c) || c == '_') + { + int start = i; + i++; + while (i < n && (char.IsLetterOrDigit(sql[i]) || sql[i] == '_' || sql[i] == '$')) i++; + if (depth == 0) tokens.Add(sql.Substring(start, i - start)); + continue; + } + + i++; // Any other punctuation (commas, operators, parameters …) is irrelevant. + } + + return tokens; + } +} diff --git a/src/DSoftStudio.Mediator.OpenTelemetry/TracerProviderBuilderExtensions.cs b/src/DSoftStudio.Mediator.OpenTelemetry/TracerProviderBuilderExtensions.cs index 6be3073..cfed280 100644 --- a/src/DSoftStudio.Mediator.OpenTelemetry/TracerProviderBuilderExtensions.cs +++ b/src/DSoftStudio.Mediator.OpenTelemetry/TracerProviderBuilderExtensions.cs @@ -13,13 +13,24 @@ namespace OpenTelemetry.Trace public static class MediatorTracerProviderBuilderExtensions { /// - /// Subscribes to the mediator . - /// Convenience method — equivalent to AddSource("DSoftStudio.Mediator"). + /// Subscribes to the mediator and installs the + /// database span enricher. /// + /// + /// In addition to AddSource("DSoftStudio.Mediator"), this registers + /// so that any database client span flowing through + /// the same provider is automatically tagged with a redaction-safe db.operation.name / + /// db.sql.table when the underlying instrumentation only emitted a raw db.statement. + /// That lets the Pipeline Explorer show each query (e.g. a SELECT vs an INSERT) as a + /// distinct dependency instead of one aggregated row — with zero configuration. Call this before + /// the exporter so the enrichment is applied prior to export. + /// public static TracerProviderBuilder AddMediatorInstrumentation(this TracerProviderBuilder builder) { ArgumentNullException.ThrowIfNull(builder); - return builder.AddSource(MediatorInstrumentation.SourceName); + return builder + .AddSource(MediatorInstrumentation.SourceName) + .AddProcessor(new DatabaseSpanEnrichmentProcessor()); } } } diff --git a/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/DSoftStudio.Mediator.OpenTelemetry.Tests.csproj b/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/DSoftStudio.Mediator.OpenTelemetry.Tests.csproj index 45542d9..5f88bbc 100644 --- a/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/DSoftStudio.Mediator.OpenTelemetry.Tests.csproj +++ b/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/DSoftStudio.Mediator.OpenTelemetry.Tests.csproj @@ -14,6 +14,7 @@ + diff --git a/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/DatabaseSpanEnrichmentTests.cs b/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/DatabaseSpanEnrichmentTests.cs new file mode 100644 index 0000000..a6c882f --- /dev/null +++ b/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/DatabaseSpanEnrichmentTests.cs @@ -0,0 +1,199 @@ +// Copyright (c) DSoftStudio. All rights reserved. +// Licensed under the MIT License. See LICENSE in the project root for license information. + +using System.Diagnostics; +using OpenTelemetry; +using OpenTelemetry.Trace; + +namespace DSoftStudio.Mediator.OpenTelemetry.Tests; + +/// +/// Covers the redaction-safe SQL shape parser, the +/// rules, and the automatic wiring through AddMediatorInstrumentation() — proving a +/// statement-only database span is split into a distinct operation in the live tracer pipeline. +/// +[Collection("OTel")] +public class DatabaseSpanEnrichmentTests +{ + // ── Parser: operation ───────────────────────────────────────────────── + + [Theory] + [InlineData("SELECT i.unit_price FROM inventory i WHERE i.sku = $1", "SELECT")] + [InlineData("INSERT INTO orders (id, total) VALUES ($1, $2)", "INSERT")] + [InlineData("update inventory set qty = qty - $1 where sku = $2", "UPDATE")] + [InlineData("DELETE FROM orders WHERE id = $1", "DELETE")] + [InlineData(" \n\t SELECT 1", "SELECT")] + [InlineData("/* hint */ -- comment\n SELECT 1", "SELECT")] + [InlineData("WITH recent AS (SELECT id FROM orders) INSERT INTO audit SELECT id FROM recent", "INSERT")] + [InlineData("EXEC sp_DoThing", "EXECUTE")] + public void Operation_extracts_the_leading_verb(string sql, string expected) + => SqlStatementParser.Operation(sql).ShouldBe(expected); + + [Fact] + public void Operation_ignores_keywords_inside_string_literals() + // The 'INSERT' here is a value, not the operation — the statement is a SELECT. + => SqlStatementParser.Operation("SELECT 'INSERT INTO x' AS note FROM t").ShouldBe("SELECT"); + + // ── ORM-generated SQL (the real shape an EF Core → Npgsql span carries) ─ + + [Theory] + // EF Core quotes every identifier and parameterizes values — the parser unwraps the quotes and the alias. + [InlineData("SELECT i.\"UnitPrice\" FROM \"Inventory\" AS i WHERE i.\"Sku\" = @__sku_0", "SELECT", "Inventory")] + [InlineData("INSERT INTO \"Orders\" (\"Id\", \"Total\") VALUES (@p0, @p1)", "INSERT", "Orders")] + [InlineData("UPDATE \"Inventory\" SET \"Qty\" = @p0 WHERE \"Id\" = @p1", "UPDATE", "Inventory")] + [InlineData("SELECT o.\"Id\" FROM \"public\".\"Orders\" AS o", "SELECT", "Orders")] // schema-qualified → table segment + public void Orm_generated_sql_is_parsed_for_operation_and_table(string sql, string op, string table) + { + SqlStatementParser.Operation(sql).ShouldBe(op); + SqlStatementParser.Table(sql).ShouldBe(table); + } + + [Theory] + [InlineData("")] + [InlineData(" ")] + [InlineData(null)] + [InlineData("BEGIN TRANSACTION")] + [InlineData("(SELECT 1)")] // wrapped sub-select has no depth-0 verb + public void Operation_returns_null_when_no_verb_is_found(string? sql) + => SqlStatementParser.Operation(sql).ShouldBeNull(); + + // ── Parser: table ───────────────────────────────────────────────────── + + [Theory] + [InlineData("SELECT * FROM inventory WHERE sku = $1", "inventory")] + [InlineData("SELECT i.unit_price FROM public.inventory i", "inventory")] + [InlineData("INSERT INTO orders (id) VALUES ($1)", "orders")] + [InlineData("UPDATE inventory SET qty = $1", "inventory")] + [InlineData("DELETE FROM orders WHERE id = $1", "orders")] + [InlineData("SELECT * FROM \"Order Items\" oi", "Order Items")] + [InlineData("SELECT a.x FROM orders a JOIN items b ON a.id = b.oid", "orders")] // first source table + public void Table_extracts_the_target_identifier(string sql, string expected) + => SqlStatementParser.Table(sql).ShouldBe(expected); + + [Theory] + [InlineData("EXEC sp_DoThing")] // no table anchor + [InlineData("SELECT 1")] // no FROM + [InlineData("")] + public void Table_returns_null_when_indeterminate(string sql) + => SqlStatementParser.Table(sql).ShouldBeNull(); + + // ── Parser: stored procedure ────────────────────────────────────────── + + [Theory] + [InlineData("CALL create_order($1, $2)", "create_order")] + [InlineData("CALL billing.charge_card($1)", "charge_card")] + [InlineData("EXEC sp_PlaceOrder @customerId = $1", "sp_PlaceOrder")] + [InlineData("EXECUTE dbo.RecalculateTotals", "RecalculateTotals")] + public void Procedure_extracts_the_invoked_routine(string sql, string expected) + => SqlStatementParser.Procedure(sql).ShouldBe(expected); + + [Theory] + [InlineData("SELECT * FROM orders")] // not a procedure call + [InlineData("EXEC ('dynamic sql here')")] // dynamic EXEC — no static name + [InlineData("")] + public void Procedure_returns_null_when_not_a_call(string sql) + => SqlStatementParser.Procedure(sql).ShouldBeNull(); + + // ── Processor rules ─────────────────────────────────────────────────── + + private static Activity NewSpan(Action configure) + { + var activity = new Activity("db.query"); + activity.Start(); + configure(activity); + return activity; + } + + [Fact] + public void Processor_enriches_a_statement_only_db_span() + { + var processor = new DatabaseSpanEnrichmentProcessor(); + using var span = NewSpan(a => + { + a.SetTag("db.system", "postgresql"); + a.SetTag("db.statement", "INSERT INTO orders (id) VALUES ($1)"); + }); + + processor.OnEnd(span); + + span.GetTagItem("db.operation.name").ShouldBe("INSERT"); + span.GetTagItem("db.sql.table").ShouldBe("orders"); + } + + [Fact] + public void Processor_enriches_a_stored_procedure_call() + { + var processor = new DatabaseSpanEnrichmentProcessor(); + using var span = NewSpan(a => + { + a.SetTag("db.system", "postgresql"); + a.SetTag("db.statement", "CALL create_order($1, $2)"); + }); + + processor.OnEnd(span); + + span.GetTagItem("db.operation.name").ShouldBe("CALL"); + span.GetTagItem("db.stored_procedure.name").ShouldBe("create_order"); + span.GetTagItem("db.sql.table").ShouldBeNull(); // a CALL has no DML table + } + + [Fact] + public void Processor_never_overwrites_native_operation() + { + var processor = new DatabaseSpanEnrichmentProcessor(); + using var span = NewSpan(a => + { + a.SetTag("db.system", "postgresql"); + a.SetTag("db.operation.name", "BATCH"); // supplied by native instrumentation + a.SetTag("db.statement", "INSERT INTO orders (id) VALUES ($1)"); + }); + + processor.OnEnd(span); + + span.GetTagItem("db.operation.name").ShouldBe("BATCH"); + } + + [Fact] + public void Processor_ignores_non_database_spans() + { + var processor = new DatabaseSpanEnrichmentProcessor(); + using var span = NewSpan(a => a.SetTag("http.request.method", "GET")); + + processor.OnEnd(span); + + span.GetTagItem("db.operation.name").ShouldBeNull(); + } + + [Fact] + public void Processor_is_a_no_op_when_statement_is_absent() + { + var processor = new DatabaseSpanEnrichmentProcessor(); + using var span = NewSpan(a => a.SetTag("db.system", "postgresql")); + + processor.OnEnd(span); + + span.GetTagItem("db.operation.name").ShouldBeNull(); + } + + // ── End-to-end wiring (proves the automatic registration runs) ───────── + + [Fact] + public void AddMediatorInstrumentation_auto_enriches_db_spans_in_the_pipeline() + { + using var dbSource = new ActivitySource("Test.Db.Enrichment"); + using var provider = global::OpenTelemetry.Sdk.CreateTracerProviderBuilder() + .AddMediatorInstrumentation() // registers the enricher — no extra config + .AddSource(dbSource.Name) + .SetSampler(new AlwaysOnSampler()) + .Build(); + + var span = dbSource.StartActivity("ordersdb", ActivityKind.Client); + span.ShouldNotBeNull(); + span.SetTag("db.system", "postgresql"); + span.SetTag("db.statement", "SELECT i.unit_price FROM inventory i WHERE i.sku = $1"); + span.Stop(); // triggers the processor's OnEnd in the provider pipeline + + span.GetTagItem("db.operation.name").ShouldBe("SELECT"); + span.GetTagItem("db.sql.table").ShouldBe("inventory"); + } +} diff --git a/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/FilteringTests.cs b/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/FilteringTests.cs index e514a95..956a88d 100644 --- a/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/FilteringTests.cs +++ b/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/FilteringTests.cs @@ -12,6 +12,7 @@ namespace DSoftStudio.Mediator.OpenTelemetry.Tests; public class FilteringTests : IDisposable { private readonly MeterListener _meterListener; + private readonly TestMetrics _metrics = new(); private readonly List<(string Name, double Value, KeyValuePair[] Tags)> _measurements = []; private readonly List<(string Name, long Value, KeyValuePair[] Tags)> _counterMeasurements = []; @@ -39,6 +40,7 @@ public FilteringTests() public void Dispose() { _meterListener.Dispose(); + _metrics.Dispose(); GC.SuppressFinalize(this); } @@ -82,7 +84,7 @@ public async Task Filter_suppresses_metrics_for_matched_request() { Filter = type => !type.Name.StartsWith("HealthCheck") }; - var behavior = new MediatorMetricsBehavior(options); + var behavior = new MediatorMetricsBehavior(options, _metrics.Metrics); var handler = new HealthCheckHandler(); await behavior.Handle(new HealthCheckQuery(), handler, TestContext.Current.CancellationToken); @@ -116,7 +118,7 @@ public async Task Filter_suppresses_stream_metrics() { Filter = type => !type.Name.StartsWith("HealthCheck") }; - var behavior = new MediatorStreamMetricsBehavior(options); + var behavior = new MediatorStreamMetricsBehavior(options, _metrics.Metrics); var handler = new HealthCheckStreamHandler(); await foreach (var _ in behavior.Handle(new HealthCheckStreamRequest(), handler, TestContext.Current.CancellationToken)) @@ -137,7 +139,7 @@ public async Task Filter_suppresses_notification_tracing_and_metrics() Filter = type => !type.Name.StartsWith("HealthCheck") }; var inner = new SequentialNotificationPublisher(); - var publisher = new InstrumentedNotificationPublisher(inner, options); + var publisher = new InstrumentedNotificationPublisher(inner, options, _metrics.Metrics); var handlers = new INotificationHandler[] { diff --git a/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/Fixtures/TestMetrics.cs b/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/Fixtures/TestMetrics.cs new file mode 100644 index 0000000..cd812d4 --- /dev/null +++ b/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/Fixtures/TestMetrics.cs @@ -0,0 +1,27 @@ +// Copyright (c) DSoftStudio. All rights reserved. +// Licensed under the MIT License. See LICENSE in the project root for license information. + +using System.Diagnostics.Metrics; +using Microsoft.Extensions.DependencyInjection; + +namespace DSoftStudio.Mediator.OpenTelemetry.Tests.Fixtures; + +/// +/// Builds a backed by a real DI (mirroring how the +/// library creates its instruments in production). Each instance owns an isolated service provider, so the +/// underlying Meter is released on and never leaks across tests. +/// +internal sealed class TestMetrics : IDisposable +{ + private readonly ServiceProvider _provider; + + public TestMetrics() + { + _provider = new ServiceCollection().AddMetrics().BuildServiceProvider(); + Metrics = new MediatorMetrics(_provider.GetRequiredService()); + } + + public MediatorMetrics Metrics { get; } + + public void Dispose() => _provider.Dispose(); +} diff --git a/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/MediatorMetricsTests.cs b/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/MediatorMetricsTests.cs new file mode 100644 index 0000000..8d8f0ac --- /dev/null +++ b/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/MediatorMetricsTests.cs @@ -0,0 +1,81 @@ +// Copyright (c) DSoftStudio. All rights reserved. +// Licensed under the MIT License. See LICENSE in the project root for license information. + +using System.Diagnostics.Metrics; +using Microsoft.Extensions.DependencyInjection; +using OpenTelemetry; +using OpenTelemetry.Metrics; + +namespace DSoftStudio.Mediator.OpenTelemetry.Tests; + +/// +/// Verifies the metric instruments follow the .NET / OpenTelemetry standards: created from an +/// (not a static Meter), in SECONDS, with explicit sub-second histogram bucket +/// boundaries so p50/p95/p99 are meaningful for millisecond-scale requests (the OTel default buckets would +/// collapse every sub-5-second request into one bucket). +/// +[Collection("OTel")] +public class MediatorMetricsTests +{ + [Fact] + public void Duration_histogram_advertises_sub_second_buckets_not_the_otel_defaults() + { + var exported = new List(); + using var provider = new ServiceCollection().AddMetrics().BuildServiceProvider(); + var metrics = new MediatorMetrics(provider.GetRequiredService()); + + using var meterProvider = global::OpenTelemetry.Sdk.CreateMeterProviderBuilder() + .AddMeter(MediatorInstrumentation.SourceName) + .AddInMemoryExporter(exported) + .Build(); + + // A typical 8 ms request, recorded in seconds. + metrics.RequestDuration.Record(0.008); + meterProvider!.ForceFlush(); + + var histogram = exported.Single(m => m.Name == "mediator.request.duration"); + histogram.MetricType.ShouldBe(MetricType.Histogram); + histogram.Unit.ShouldBe("s"); + + var bounds = new List(); + foreach (ref readonly var point in histogram.GetMetricPoints()) + { + foreach (var bucket in point.GetHistogramBuckets()) + if (!double.IsPositiveInfinity(bucket.ExplicitBound)) + bounds.Add(bucket.ExplicitBound); + break; + } + + // Our explicit sub-second boundaries — NOT the OTel default [0,5,10,25,…] s. + bounds.Count.ShouldBe(13); + bounds.ShouldContain(0.005); + bounds.ShouldContain(0.01); + bounds.ShouldContain(0.5); + bounds.ShouldNotContain(25); // a default-bucket boundary that must be absent + } + + [Fact] + public void Instruments_are_named_and_unit_per_convention() + { + var names = new HashSet(); + var units = new Dictionary(); + using var listener = new MeterListener + { + InstrumentPublished = (instrument, l) => + { + if (instrument.Meter.Name != MediatorInstrumentation.SourceName) return; + names.Add(instrument.Name); + units[instrument.Name] = instrument.Unit; + } + }; + listener.Start(); + + using var provider = new ServiceCollection().AddMetrics().BuildServiceProvider(); + _ = new MediatorMetrics(provider.GetRequiredService()); + + names.ShouldBe(new[] { "mediator.request.duration", "mediator.request.active", "mediator.request.errors" }, ignoreOrder: true); + units["mediator.request.duration"].ShouldBe("s"); // UCUM seconds + units["mediator.request.active"].ShouldBe("{request}"); // dimensionless annotation + units["mediator.request.errors"].ShouldBe("{error}"); + } +} diff --git a/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/MetricsBehaviorTests.cs b/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/MetricsBehaviorTests.cs index e91d186..6987994 100644 --- a/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/MetricsBehaviorTests.cs +++ b/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/MetricsBehaviorTests.cs @@ -11,6 +11,7 @@ namespace DSoftStudio.Mediator.OpenTelemetry.Tests; public class MetricsBehaviorTests : IDisposable { private readonly MeterListener _listener; + private readonly TestMetrics _metrics = new(); private readonly List<(string Name, double Value, KeyValuePair[] Tags)> _measurements = []; private readonly List<(string Name, long Value, KeyValuePair[] Tags)> _counterMeasurements = []; @@ -38,6 +39,7 @@ public MetricsBehaviorTests() public void Dispose() { _listener.Dispose(); + _metrics.Dispose(); GC.SuppressFinalize(this); } @@ -45,7 +47,7 @@ public void Dispose() public async Task Records_duration_on_success() { var options = new MediatorInstrumentationOptions(); - var behavior = new MediatorMetricsBehavior(options); + var behavior = new MediatorMetricsBehavior(options, _metrics.Metrics); var handler = new TestCommandHandler(); await behavior.Handle(new TestCommand("test"), handler, TestContext.Current.CancellationToken); @@ -64,7 +66,7 @@ public async Task Records_duration_on_success() public async Task Records_active_count_increment_and_decrement() { var options = new MediatorInstrumentationOptions(); - var behavior = new MediatorMetricsBehavior(options); + var behavior = new MediatorMetricsBehavior(options, _metrics.Metrics); var handler = new TestCommandHandler(); await behavior.Handle(new TestCommand("test"), handler, TestContext.Current.CancellationToken); @@ -80,7 +82,7 @@ public async Task Records_active_count_increment_and_decrement() public async Task Records_error_count_on_exception() { var options = new MediatorInstrumentationOptions(); - var behavior = new MediatorMetricsBehavior(options); + var behavior = new MediatorMetricsBehavior(options, _metrics.Metrics); var handler = new FailingCommandHandler(); await Should.ThrowAsync(async () => @@ -97,7 +99,7 @@ await Should.ThrowAsync(async () => public async Task No_metrics_when_disabled() { var options = new MediatorInstrumentationOptions { EnableMetrics = false }; - var behavior = new MediatorMetricsBehavior(options); + var behavior = new MediatorMetricsBehavior(options, _metrics.Metrics); var handler = new TestCommandHandler(); await behavior.Handle(new TestCommand("test"), handler, TestContext.Current.CancellationToken); diff --git a/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/NotificationPublisherTests.cs b/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/NotificationPublisherTests.cs index 26328eb..b550b83 100644 --- a/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/NotificationPublisherTests.cs +++ b/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/NotificationPublisherTests.cs @@ -12,6 +12,7 @@ namespace DSoftStudio.Mediator.OpenTelemetry.Tests; public class NotificationPublisherTests : IDisposable { private readonly MeterListener _meterListener; + private readonly TestMetrics _metrics = new(); private readonly List<(string Name, double Value, KeyValuePair[] Tags)> _measurements = []; private readonly List<(string Name, long Value, KeyValuePair[] Tags)> _counterMeasurements = []; @@ -39,6 +40,7 @@ public NotificationPublisherTests() public void Dispose() { _meterListener.Dispose(); + _metrics.Dispose(); GC.SuppressFinalize(this); } @@ -48,7 +50,7 @@ public async Task Creates_parent_span_with_per_handler_child_spans() using var collector = new ActivityCollector(); var options = new MediatorInstrumentationOptions(); var inner = new SequentialNotificationPublisher(); - var publisher = new InstrumentedNotificationPublisher(inner, options); + var publisher = new InstrumentedNotificationPublisher(inner, options, _metrics.Metrics); var handler1 = new TestNotificationHandler1(); var handler2 = new TestNotificationHandler2(); @@ -87,7 +89,7 @@ public async Task Handler_child_span_names_use_handler_type_name() using var collector = new ActivityCollector(); var options = new MediatorInstrumentationOptions(); var inner = new SequentialNotificationPublisher(); - var publisher = new InstrumentedNotificationPublisher(inner, options); + var publisher = new InstrumentedNotificationPublisher(inner, options, _metrics.Metrics); var handlers = new INotificationHandler[] { @@ -112,7 +114,7 @@ public async Task Error_in_handler_sets_error_status_on_parent_and_child() using var collector = new ActivityCollector(); var options = new MediatorInstrumentationOptions(); var inner = new SequentialNotificationPublisher(); - var publisher = new InstrumentedNotificationPublisher(inner, options); + var publisher = new InstrumentedNotificationPublisher(inner, options, _metrics.Metrics); var handlers = new INotificationHandler[] { @@ -137,7 +139,7 @@ public async Task Records_metrics_for_notification_publish() { var options = new MediatorInstrumentationOptions(); var inner = new SequentialNotificationPublisher(); - var publisher = new InstrumentedNotificationPublisher(inner, options); + var publisher = new InstrumentedNotificationPublisher(inner, options, _metrics.Metrics); var handlers = new INotificationHandler[] { @@ -162,7 +164,7 @@ public async Task Records_error_metrics_on_handler_failure() { var options = new MediatorInstrumentationOptions(); var inner = new SequentialNotificationPublisher(); - var publisher = new InstrumentedNotificationPublisher(inner, options); + var publisher = new InstrumentedNotificationPublisher(inner, options, _metrics.Metrics); var handlers = new INotificationHandler[] { @@ -186,7 +188,7 @@ public async Task Passthrough_when_no_listeners_and_no_meter() using var noMetrics = new MeterListener(); // empty, won't subscribe var options = new MediatorInstrumentationOptions(); var inner = new SequentialNotificationPublisher(); - var publisher = new InstrumentedNotificationPublisher(inner, options); + var publisher = new InstrumentedNotificationPublisher(inner, options, _metrics.Metrics); var handler = new TestNotificationHandler1(); var handlers = new INotificationHandler[] { handler }; @@ -209,7 +211,7 @@ public async Task EnrichActivity_callback_on_notification() } }; var inner = new SequentialNotificationPublisher(); - var publisher = new InstrumentedNotificationPublisher(inner, options); + var publisher = new InstrumentedNotificationPublisher(inner, options, _metrics.Metrics); var handlers = new INotificationHandler[] { diff --git a/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/StreamMetricsBehaviorTests.cs b/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/StreamMetricsBehaviorTests.cs index 418e2d6..6c34fb3 100644 --- a/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/StreamMetricsBehaviorTests.cs +++ b/tests/DSoftStudio.Mediator.OpenTelemetry.Tests/StreamMetricsBehaviorTests.cs @@ -10,6 +10,7 @@ namespace DSoftStudio.Mediator.OpenTelemetry.Tests; public class StreamMetricsBehaviorTests : IDisposable { private readonly MeterListener _listener; + private readonly TestMetrics _metrics = new(); private readonly List<(string Name, double Value, KeyValuePair[] Tags)> _measurements = []; private readonly List<(string Name, long Value, KeyValuePair[] Tags)> _counterMeasurements = []; @@ -37,6 +38,7 @@ public StreamMetricsBehaviorTests() public void Dispose() { _listener.Dispose(); + _metrics.Dispose(); GC.SuppressFinalize(this); } @@ -44,7 +46,7 @@ public void Dispose() public async Task Records_duration_covering_full_enumeration() { var options = new MediatorInstrumentationOptions(); - var behavior = new MediatorStreamMetricsBehavior(options); + var behavior = new MediatorStreamMetricsBehavior(options, _metrics.Metrics); var handler = new TestStreamHandler(); await foreach (var _ in behavior.Handle(new TestStreamRequest(3), handler, TestContext.Current.CancellationToken)) @@ -65,7 +67,7 @@ public async Task Records_duration_covering_full_enumeration() public async Task Records_active_count_for_stream() { var options = new MediatorInstrumentationOptions(); - var behavior = new MediatorStreamMetricsBehavior(options); + var behavior = new MediatorStreamMetricsBehavior(options, _metrics.Metrics); var handler = new TestStreamHandler(); await foreach (var _ in behavior.Handle(new TestStreamRequest(1), handler, TestContext.Current.CancellationToken)) @@ -83,7 +85,7 @@ public async Task Records_active_count_for_stream() public async Task Records_duration_even_on_stream_error() { var options = new MediatorInstrumentationOptions(); - var behavior = new MediatorStreamMetricsBehavior(options); + var behavior = new MediatorStreamMetricsBehavior(options, _metrics.Metrics); var handler = new FailingStreamHandler(); await Should.ThrowAsync(async () => @@ -104,7 +106,7 @@ await Should.ThrowAsync(async () => public async Task No_metrics_when_disabled() { var options = new MediatorInstrumentationOptions { EnableMetrics = false }; - var behavior = new MediatorStreamMetricsBehavior(options); + var behavior = new MediatorStreamMetricsBehavior(options, _metrics.Metrics); var handler = new TestStreamHandler(); await foreach (var _ in behavior.Handle(new TestStreamRequest(1), handler, TestContext.Current.CancellationToken))