fix(telemetry): align Pulse activities and metrics with OpenTelemetry conventions and record abandoned streams

decisions/2026-09-28-opentelemetry-telemetry-conventions.md

@@ -0,0 +1,79 @@
+---
+authors:
+  - Martin Stühmer
+
+applyTo:
+  - "src/NetEvolve.Pulse/Interceptors/ActivityAndMetrics*.cs"
+  - "src/NetEvolve.Pulse/Outbox/OutboxProcessorHostedService.cs"
+  - "src/NetEvolve.Pulse/Internals/Defaults.cs"
+  - "src/NetEvolve.Pulse/Internals/TelemetryUnits.cs"
+
+created: 2026-09-28
+
+lastModified: 2026-09-28
+
+state: proposed
+
+instructions: |
+  MUST leave the activity status Unset when a Pulse operation succeeds or a stream consumer stops early; MUST set Error with the exception message only on failure.
+  MUST set the error.type tag to the exception's full type name on the span, the error counter and the duration histogram of a failed operation, and MUST NOT set it otherwise.
+  MUST record the stream query duration once for every outcome (completed, faulted, abandoned) and tag the abandoned outcome with pulse.stream.completed=false.
+  MUST keep the legacy units (ms, requests, events, queries, errors, messages) as default until the opt-in ActivityAndMetricsOptions.UseSemanticConventionUnits becomes the default in a later 0.x release; with the opt-in, durations use seconds (unit s) with explicit bucket advice and counters use UCUM annotations ({request}, {event}, {query}, {error}, {message}).
+  MUST keep the existing metric and tag names.
+---
+
+# Decision: OpenTelemetry Telemetry Conventions
+
+Pulse activities and metrics follow the OpenTelemetry Trace API and semantic conventions for status handling, `error.type` and units. Status and `error.type` changes apply immediately, the unit changes are opt-in during a transition period.
+
+## Context
+
+Issue #832 lists the deviations of the built-in telemetry in `NetEvolve.Pulse` from the OpenTelemetry guidance:
+
+* All three `ActivityAndMetrics*Interceptor` classes set `ActivityStatusCode.Ok` on success. The [Trace API](https://opentelemetry.io/docs/specs/otel/trace/api/#set-status) states: "Generally, Instrumentation Libraries SHOULD NOT set the status code to `Ok`, unless explicitly configured to do so."
+* Failed operations carry no `error.type`. [Recording errors](https://opentelemetry.io/docs/specs/semconv/general/recording-errors/) states that spans and metrics SHOULD carry `error.type` on failure and SHOULD NOT include it on success.
+* Duration histograms use `ms`, counters use plain words (`requests`, `errors`, `messages`). The [metrics guidelines](https://opentelemetry.io/docs/specs/semconv/general/metrics/#instrument-units) state: "When instruments are measuring durations, seconds (i.e. `s`) SHOULD be used." and ask for UCUM annotations such as `{request}`.
+
+Issue #831 shows that `ActivityAndMetricsStreamQueryInterceptor` records no duration and no outcome when the consumer stops enumerating early, because the recording code runs after the `try`/`finally` block that holds the `yield return`.
+
+No earlier decision defines the Pulse telemetry contract. Units are part of the exported metric identity: the OpenTelemetry Prometheus exporter appends the unit to the metric name (for example `_milliseconds` or `_seconds`). Changing a unit therefore breaks existing dashboards and alerts.
+
+## Decision
+
+* MUST leave the activity status `Unset` when an operation succeeds and when a stream consumer stops early. MUST set `Error` with the exception message on failure.
+* MUST set `error.type` to `Exception.GetType().FullName` on the span, the error counter and the duration histogram of a failed operation. The value MUST be identical on all three. The existing `pulse.exception.*` tags stay.
+* MUST record `pulse.stream_query.duration` exactly once per stream query, from the `finally` block of the iterator. A stream the consumer abandons carries `pulse.stream.completed=false` on the histogram and the activity, keeps the status `Unset` and carries no `pulse.success` tag. Completed and faulted streams keep their `pulse.success` semantics. A stream is faulted when its handler, `MoveNextAsync`, `Current` or the inner `DisposeAsync` throws, including an `OperationCanceledException` from a cancelled token. When the inner `DisposeAsync` throws after an earlier fault, the earlier exception is thrown and recorded, so the thrown exception and `error.type` always match.
+* MUST treat an exception thrown synchronously by the stream handler delegate like an exception thrown during enumeration.
+* MUST offer the new units through `ActivityAndMetricsOptions.UseSemanticConventionUnits` (default `false`). The option applies to the interceptors and to `OutboxProcessorHostedService`:
+
+  | Instrument | Legacy unit | Opt-in unit |
+  | --- | --- | --- |
+  | `pulse.requests.total` | `requests` | `{request}` |
+  | `pulse.events.total` | `events` | `{event}` |
+  | `pulse.stream_query.total` | `queries` | `{query}` |
+  | `pulse.request.errors`, `pulse.event.errors`, `pulse.stream_query.errors` | `errors` | `{error}` |
+  | `pulse.outbox.processed.total`, `pulse.outbox.failed.total`, `pulse.outbox.deadletter.total`, `pulse.outbox.pending` | `messages` | `{message}` |
+  | `pulse.request.duration`, `pulse.event.duration`, `pulse.stream_query.duration`, `pulse.outbox.processing.duration` | `ms` | `s` |
+
+* MUST provide explicit histogram bucket boundaries for the `s` unit on all target frameworks (`InstrumentAdvice<T>`), using the boundaries of the HTTP semantic conventions: 0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1, 2.5, 5, 7.5, 10.
+* MUST keep the metric names and tag names unchanged.
+* Transition plan: a later `0.x` release makes `UseSemanticConventionUnits` the default. The legacy units and the option are removed before `1.0.0`.
+
+## Consequences
+
+* Backends treat successful Pulse spans as `Unset` and group failures by `error.type`.
+* Abandoned streams appear in the duration histogram, so `pulse.stream_query.total` and `pulse.stream_query.duration` stay consistent.
+* Dashboards that filter on `Ok` status must filter on "not `Error`" instead.
+* Operators opt into the new units once their dashboards and alerts are migrated. Until then the exported metric names stay as they are.
+* A process that mixes both unit modes creates two instruments with the same name on the `NetEvolve.Pulse` meter. The option is meant to be set once per application.
+
+## Alternatives Considered
+
+* **Switch the units immediately.** Rejected: issue #832 requires an opt-in during the transition, because the unit is part of the exported metric name.
+* **AppContext switch instead of options.** Rejected: a process-wide switch cannot be tested per test, and the repository already configures interceptors through options.
+* **Rename the `.total` counters.** Not part of this decision. Renaming breaks every dashboard without an opt-in path and can be decided separately.
+
+## Related Decisions
+
+* [Extensibility Interface Evolution Before 1.0](./2026-09-24-extensibility-interface-evolution-pre-1-0.md) - Governs how the changed telemetry contract is announced while the major version is `0`.
+* [DateTimeOffset and TimeProvider Usage](./2026-01-21-datetimeoffset-and-timeprovider-usage.md) - Durations are measured with `TimeProvider`.

src/NetEvolve.Pulse/ActivityMetricsExtensions.cs

@@ -19,11 +19,17 @@ public static class ActivityMetricsExtensions
     /// </summary>
     /// <param name="builder">The mediator builder.</param>
     /// <returns>The builder for method chaining.</returns>
+    /// <remarks>
+    /// Metrics keep their legacy units by default. Use
+    /// <see cref="AddActivityAndMetrics(IMediatorBuilder, Action{ActivityAndMetricsOptions})"/> to opt into the
+    /// units of the OpenTelemetry semantic conventions.
+    /// </remarks>
     /// <exception cref="ArgumentNullException">Thrown when <paramref name="builder"/> is <see langword="null"/>.</exception>
     public static IMediatorBuilder AddActivityAndMetrics(this IMediatorBuilder builder)
     {
         ArgumentNullException.ThrowIfNull(builder);
 
+        _ = builder.Services.AddOptions<ActivityAndMetricsOptions>();
         builder.Services.TryAddEnumerable(
             ServiceDescriptor.Singleton(typeof(IEventInterceptor<>), typeof(ActivityAndMetricsEventInterceptor<>))
         );
@@ -39,4 +45,28 @@ public static IMediatorBuilder AddActivityAndMetrics(this IMediatorBuilder build
 
         return builder;
     }
+
+    /// <summary>
+    /// Adds activity tracing and metrics collection for all requests processed by the mediator and configures
+    /// the telemetry options, for example to opt into the units of the OpenTelemetry semantic conventions.
+    /// The options also apply to the metrics of the outbox processor.
+    /// </summary>
+    /// <param name="builder">The mediator builder.</param>
+    /// <param name="configure">The action that configures the <see cref="ActivityAndMetricsOptions"/>.</param>
+    /// <returns>The builder for method chaining.</returns>
+    /// <exception cref="ArgumentNullException">
+    /// Thrown when <paramref name="builder"/> or <paramref name="configure"/> is <see langword="null"/>.
+    /// </exception>
+    public static IMediatorBuilder AddActivityAndMetrics(
+        this IMediatorBuilder builder,
+        Action<ActivityAndMetricsOptions> configure
+    )
+    {
+        ArgumentNullException.ThrowIfNull(builder);
+        ArgumentNullException.ThrowIfNull(configure);
+
+        _ = builder.Services.Configure(configure);
+
+        return builder.AddActivityAndMetrics();
+    }
 }

src/NetEvolve.Pulse/Interceptors/ActivityAndMetricsEventInterceptor{TEvent}.cs

@@ -2,6 +2,7 @@
 
 using System.Diagnostics;
 using System.Diagnostics.Metrics;
+using Microsoft.Extensions.Options;
 using NetEvolve.Pulse.Extensibility;
 using NetEvolve.Pulse.Internals;
 using static Internals.Defaults.Tags;
@@ -16,31 +17,24 @@ internal sealed class ActivityAndMetricsEventInterceptor<TEvent> : IEventInterce
     where TEvent : IEvent
 {
     /// <summary>
-    /// Counter tracking the total number of events processed, tagged by event type.
+    /// Counter <c>pulse.events.total</c>, tagged by type.
     /// </summary>
-    private static readonly Counter<long> EventCounter = Defaults.Meter.CreateCounter<long>(
-        "pulse.events.total",
-        "events",
-        "Total number of events processed."
-    );
+    private readonly Counter<long> _eventCounter;
 
     /// <summary>
-    /// Counter tracking the total number of event errors, tagged by event type.
+    /// Counter <c>pulse.event.errors</c>, tagged by type.
     /// </summary>
-    private static readonly Counter<long> ErrorsCounter = Defaults.Meter.CreateCounter<long>(
-        "pulse.event.errors",
-        "errors",
-        "Total number of event errors."
-    );
+    private readonly Counter<long> _errorsCounter;
 
     /// <summary>
-    /// Histogram measuring event processing duration in milliseconds, with percentile distributions.
+    /// Histogram <c>pulse.event.duration</c> in milliseconds, or in seconds when semantic convention units are enabled.
     /// </summary>
-    private static readonly Histogram<double> EventDurationHistogram = Defaults.Meter.CreateHistogram<double>(
-        "pulse.event.duration",
-        "ms",
-        "Duration of event processing in milliseconds."
-    );
+    private readonly Histogram<double> _eventDurationHistogram;
+
+    /// <summary>
+    /// Whether the metrics use the units of the OpenTelemetry semantic conventions.
+    /// </summary>
+    private readonly bool _useSemanticConventionUnits;
 
     /// <summary>
     /// Time provider for consistent timestamp generation, supporting testability.
@@ -51,7 +45,34 @@ internal sealed class ActivityAndMetricsEventInterceptor<TEvent> : IEventInterce
     /// Initializes a new instance of the <see cref="ActivityAndMetricsEventInterceptor{TEvent}"/> class.
     /// </summary>
     /// <param name="timeProvider">The time provider for timestamp generation.</param>
-    public ActivityAndMetricsEventInterceptor(TimeProvider timeProvider) => _timeProvider = timeProvider;
+    /// <param name="options">The telemetry options; <see langword="null"/> keeps the legacy units.</param>
+    public ActivityAndMetricsEventInterceptor(
+        TimeProvider timeProvider,
+        IOptions<ActivityAndMetricsOptions>? options = null
+    )
+    {
+        _timeProvider = timeProvider;
+        _useSemanticConventionUnits = options?.Value.UseSemanticConventionUnits ?? false;
+        _eventCounter = TelemetryUnits.GetSharedCounter(
+            "pulse.events.total",
+            "events",
+            "{event}",
+            "Total number of events processed.",
+            _useSemanticConventionUnits
+        );
+        _errorsCounter = TelemetryUnits.GetSharedCounter(
+            "pulse.event.errors",
+            "errors",
+            "{error}",
+            "Total number of event errors.",
+            _useSemanticConventionUnits
+        );
+        _eventDurationHistogram = TelemetryUnits.GetSharedDurationHistogram(
+            "pulse.event.duration",
+            "event processing",
+            _useSemanticConventionUnits
+        );
+    }
 
     /// <inheritdoc />
     /// <remarks>
@@ -62,7 +83,9 @@ internal sealed class ActivityAndMetricsEventInterceptor<TEvent> : IEventInterce
     /// <item>Increments event counter metrics</item>
     /// <item>Measures and records execution duration</item>
     /// <item>Captures exception details on failure</item>
-    /// <item>Marks success/failure status in both activity and metrics</item>
+    /// <item>Tags <c>pulse.success</c> on the activity and the duration histogram</item>
+    /// <item>Leaves the activity status <see cref="ActivityStatusCode.Unset"/> unless the handler fails</item>
+    /// <item>Sets <c>error.type</c> on the activity, the error counter and the duration histogram on failure</item>
     /// </list>
     /// </remarks>
     public async Task HandleAsync(
@@ -93,7 +116,7 @@ public async Task HandleAsync(
             .SetTag(EventCorrelationId, message.CorrelationId)
             .SetTag(EventCausationId, message.CausationId)
             .SetTag(EventTimestamp, startTime);
-        EventCounter.Add(1, tags);
+        _eventCounter.Add(1, tags);
 
         try
         {
@@ -102,33 +125,40 @@ public async Task HandleAsync(
 
             var endTime = _timeProvider.GetUtcNow();
 
-            // Mark activity as successful
+            // Status stays Unset on success, as required by the OpenTelemetry Trace API.
             _ = activity
-                ?.SetStatus(ActivityStatusCode.Ok)
-                .SetEndTime(endTime.UtcDateTime)
+                ?.SetEndTime(endTime.UtcDateTime)
                 .SetTag(EventCompletionTimestamp, endTime)
                 .SetTag(Success, value: true);
 
             // Record successful execution duration
-            EventDurationHistogram.Record((endTime - startTime).TotalMilliseconds, [.. tags, new(Success, true)]);
+            _eventDurationHistogram.Record(
+                TelemetryUnits.ToDuration(endTime - startTime, _useSemanticConventionUnits),
+                [.. tags, new(Success, true)]
+            );
         }
         catch (Exception ex)
         {
             var errorTime = _timeProvider.GetUtcNow();
+            var errorType = ex.GetType().FullName;
 
             // Capture comprehensive exception details in the activity
             _ = activity
                 ?.SetStatus(ActivityStatusCode.Error, ex.Message)
                 .SetEndTime(errorTime.UtcDateTime)
-                .SetTag(ExceptionType, ex.GetType().FullName)
+                .SetTag(ErrorType, errorType)
+                .SetTag(ExceptionType, errorType)
                 .SetTag(ExceptionMessage, ex.Message)
                 .SetTag(ExceptionStackTrace, ex.StackTrace)
                 .SetTag(ExceptionTimestamp, errorTime)
                 .SetTag(Success, value: false);
 
             // Increment error counters and record failed execution duration
-            ErrorsCounter.Add(1, tags);
-            EventDurationHistogram.Record((errorTime - startTime).TotalMilliseconds, [.. tags, new(Success, false)]);
+            _errorsCounter.Add(1, [.. tags, new(ErrorType, errorType)]);
+            _eventDurationHistogram.Record(
+                TelemetryUnits.ToDuration(errorTime - startTime, _useSemanticConventionUnits),
+                [.. tags, new(Success, false), new(ErrorType, errorType)]
+            );
 
             throw;
         }

src/NetEvolve.Pulse/Interceptors/ActivityAndMetricsOptions.cs

@@ -0,0 +1,27 @@
+namespace NetEvolve.Pulse.Interceptors;
+
+/// <summary>
+/// Options for the built-in activity and metrics telemetry registered via <c>AddActivityAndMetrics()</c>.
+/// The options also apply to the metrics of the outbox processor.
+/// </summary>
+/// <example>
+/// <code>
+/// services.AddPulse(c =&gt; c.AddActivityAndMetrics(o =&gt; o.UseSemanticConventionUnits = true));
+/// </code>
+/// </example>
+public sealed class ActivityAndMetricsOptions
+{
+    /// <summary>
+    /// Gets or sets a value indicating whether Pulse metrics use the units of the OpenTelemetry semantic conventions.
+    /// </summary>
+    /// <remarks>
+    /// When <see langword="true"/>, duration histograms record seconds with unit <c>s</c> and explicit bucket
+    /// boundaries, and counters use UCUM annotations (<c>{request}</c>, <c>{event}</c>, <c>{query}</c>,
+    /// <c>{error}</c>, <c>{message}</c>). When <see langword="false"/> (default), durations are recorded in
+    /// milliseconds with unit <c>ms</c> and counters keep their legacy units (<c>requests</c>, <c>events</c>,
+    /// <c>queries</c>, <c>errors</c>, <c>messages</c>). Exporters such as Prometheus derive the exported metric
+    /// name from the unit, so enabling this option changes the exported names. A later <c>0.x</c> release
+    /// makes <see langword="true"/> the default.
+    /// </remarks>
+    public bool UseSemanticConventionUnits { get; set; }
+}