OpenTelemetry

Adds OpenTelemetry-compatible metrics to Shuttle.Pipelines by observing the events already exposed on PipelineOptions — no changes are required to your pipelines or observers.

Installation

dotnet add package Shuttle.Pipelines.OpenTelemetry

Registration

services.AddPipelines()
    .AddOpenTelemetry();

This subscribes to every event on PipelineOptions and records it against a Meter named Shuttle.Pipelines. The instrumentation is built on System.Diagnostics.Metrics.Meter directly, so this package has no dependency on the OpenTelemetry SDK — it only becomes "live" once something subscribes to that meter, for example:

services.AddOpenTelemetry()
    .WithMetrics(builder => builder.AddMeter("Shuttle.Pipelines"));

Since every Shuttle.Hopper pipeline is a Shuttle.Pipelines pipeline underneath, installing this package alongside Shuttle.Hopper.OpenTelemetry gives you these metrics for Hopper's inbox, outbox, send, dispatch, deferred, startup and shutdown pipelines too — with no overlap, since the two packages use different meter names and don't share any state.

Metrics

Name	Instrument	Unit	Description
pipelines.started	Counter	{execution}	Number of pipeline executions started.
pipelines.completed	Counter	{execution}	Number of pipeline executions that completed successfully.
pipelines.failed	Counter	{execution}	Number of pipeline executions that failed with an unhandled exception.
pipelines.aborted	Counter	{execution}	Number of pipeline executions that were aborted.
pipelines.recursive_exceptions	Counter	{exception}	Number of exceptions raised while handling a pipeline failure.
pipelines.duration	Histogram	ms	Duration of a pipeline execution, from PipelineStarting to its outcome.
pipelines.stage.duration	Histogram	ms	Duration of a single pipeline stage.
pipelines.event.duration	Histogram	ms	Duration of a single pipeline event.

• pipelines.started / .completed / .failed / .aborted / .recursive_exceptions and pipelines.duration are tagged with pipeline.type (the pipeline's runtime type name); pipelines.duration also carries pipeline.outcome (completed / failed / aborted / recursive-exception).
• pipelines.stage.duration additionally carries pipeline.stage.
• pipelines.event.duration additionally carries pipeline.stage and pipeline.event (the pipeline event marker type, e.g. AssembleMessage, HandleMessage). This one fires for every event checkpoint in every stage, so expect it to be the highest-cardinality, highest-frequency metric of the set — enable the meter accordingly.

A pipeline execution that fails while an observer subscribed to the PipelineFailed event itself throws does not reach PipelineCompleted / Failed / Aborted — it raises PipelineRecursiveException instead. That case is still counted and still closes out pipelines.duration (tagged pipeline.outcome = recursive-exception), so in-flight pipeline timings never leak.

Extension points

• PipelineTelemetry.Meter – the shared Meter instance used throughout, exposed so you can add your own measurements under the same name.