OpenTelemetry

The otel feature gives a service distributed tracing: a trace flows from an incoming message onto the replies it produces, so one trace spans the whole consume-transform-produce chain. It is built on the typed publish-path context - the same seam that lets a publish transform read the delivery that produced a reply.

ruststream = { version = "0.6", features = ["macros", "memory", "json", "otel"] }

The feature has two halves. Propagation carries the W3C Trace Context and emits tracing spans; it is broker-agnostic and works with no exporter at all. Export ships with the feature: the OpenTelemetry SDK and OTLP exporters behind Otel::builder().init(), which installs the global providers and bridges the spans into them - or assemble your own subscriber (for example tracing-opentelemetry), exactly as the logging guide leaves the subscriber to you.

Wiring it up

Create an OpenTelemetry, add its consume layer app-wide, and bake its propagation onto the reply publisher:

tests/opentelemetry.rs

let otel = OpenTelemetry::new();
let broker = MemoryBroker::new();
let ingress = broker.publisher();
// The reply wiring propagates the delivery's trace context onto each reply.
let reply_pub = TypedPublisher::new(MemoryPublish).transform(otel.propagation());

let app = RustStream::new(AppInfo::new("svc", "0.1.0"))
    // The consume layer opens a span per delivery and records the consumer's trace context.
    .layer(otel.consume_layer())
    .with_broker(broker, |b| {
        b.include(echo).publisher(reply_pub);
        b.include(capture);
    });

• consume_layer() is a consume-side layer: per delivery it reads the incoming traceparent, opens a tracing span for the handler, and records the consumer's span on the working headers. It applies to handlers mounted directly and through a router.
• propagation() is a static publish layer: it copies the working traceparent (and tracestate) onto every reply, so a downstream service sees the consumer span as the reply's parent. Reuse it on a batch publisher with for_batch(otel.propagation()).

What gets propagated

A delivery carrying 00-<trace-id>-<span-id>-01 continues that trace: the reply keeps the same trace-id and carries a fresh span-id (the consumer's span), so the trace is linked end to end. A delivery with no traceparent starts a fresh, sampled root trace. The spans are emitted under the ruststream.consume target with trace_id / span_id / subscription fields.

Reading the trace in a handler

The consumer's trace context is on the working headers, so a handler reads it the same way any header is read - through the context:

let traceparent = ctx.headers().get_str("traceparent");

Parse it with the OpenTelemetry SDK's TraceContextPropagator (the same parser the consume layer uses) into an opentelemetry::trace::SpanContext to read the trace_id() / span_id() or to check is_sampled().

Exporting to a collector

The propagation module stops at the W3C context and tracing spans; there are two ways to ship them to a collector. Assemble tracing-opentelemetry and an exporter yourself in the binary (the same split as logging) - or let Otel::builder().init() below do it for you.

The otel feature: SDK, OTLP, and the metrics inventory

Otel::builder().init() builds the OTLP exporters, installs the OpenTelemetry tracer and meter providers as the process globals, and bridges tracing spans into them - so the spans the propagation layer already opens are exported with no further wiring:

examples/otel_export.rs

// `use<>` opts out of capturing the `otel` borrow: the layers `Arc`-share the instruments, so
// the built app owns its half and `main` keeps the other for the final flush.
fn app(otel: &Otel) -> impl App + use<> {
    RustStream::new(AppInfo::new("orders-svc", "0.1.0"))
        // per-delivery metrics: consumed, process duration, outcomes, in-flight, queue time
        .layer(otel.consume_layer())
        // per-publish metrics: sent, operation duration, payload size, queue-time stamp
        .publish_layer(otel.publish_layer())
        // business instruments: built once against the global meter, shared as typed state
        .on_startup(async move |()| {
            Ok::<_, Infallible>(AppState {
                metrics: OrderMetrics {
                    accepted: global::meter("orders-svc")
                        .u64_counter("orders_accepted")
                        .build(),
                },
            })
        })
        .with_broker(MemoryBroker::new(), |b| {
            b.include(accept);
        })
}

// A hand-written entry point instead of #[ruststream::app]: the macro's generated main ends at
// run_main, and the exporters batch in the background, so flushing them needs one call after the
// app has drained.
fn main() -> ExitCode {
    // Installs the global tracer + meter providers, the OTLP exporters, and the tracing bridge;
    // every span the propagation module opens and every metric the service records now exports.
    let otel = Otel::builder()
        .service_name("orders-svc")
        .otlp_endpoint("http://localhost:4317")
        .messaging_system("memory")
        .attribute("deployment.environment", "dev")
        .init()
        .expect("otel init failed");

    let code = run_main(|| app(&otel));
    // Ships the last buffered spans and metric points to the collector; dropping `Otel` does not.
    otel.shutdown().expect("otel shutdown failed");
    code
}

The two middleware carry the dispatch metrics, labeled per handler (messaging.destination.name), following the messaging semantic conventions plus a ruststream.* namespace:

Instrument	Kind	What it measures
messaging.client.consumed.messages	counter	deliveries received
messaging.process.duration	histogram (semconv buckets)	handler processing time
ruststream.messages.processed	counter, outcome attribute	settlements: ack, nack_requeue, nack_drop, retry_after
ruststream.messages.in_flight	up-down counter	deliveries inside handlers (pool saturation vs workers(n))
ruststream.message.queue_time	histogram	publish-to-handler-start lag, from the stamped publish-time header
ruststream.messages.decode_failures	counter	deliveries whose payload the codec rejected
ruststream.messages.panics	counter	handler invocations that panicked
messaging.client.sent.messages	counter, error.type on failure	publishes
messaging.client.operation.duration	histogram	the publish operation
ruststream.message.payload.size	histogram (By)	published payload sizes
ruststream.batch.size	histogram	decoded batch sizes handed to batch handlers
ruststream.app.state	observable gauge	the lifecycle state, from RunningApp::health via otel.observe_health(running.health())

Batch handlers bypass the per-message consume layer (the documented middleware exception), so ruststream.batch.size is recorded by the batch dispatch itself through the global meter: it goes live once init() installs the global providers, and stays silent under a bare attach() unless you install your provider globally yourself.

Because init() installs the global providers, business metrics need no exporter plumbing: build the instruments once at startup into one storage object, share it through the typed state (injectable with State<..> via FromRef), and everything in it rides the same OTLP pipeline:

examples/otel_export.rs

/// The service's business instruments: one storage object, built once at startup against the
/// global meter `Otel::init` installed, so everything in it rides the same OTLP pipeline as the
/// framework's dispatch metrics.
#[derive(Clone)]
struct OrderMetrics {
    accepted: Counter<u64>,
}

#[derive(Clone, FromRef)]
struct AppState {
    metrics: OrderMetrics,
}

#[subscriber("orders")]
async fn accept(order: &Order, State(metrics): State<OrderMetrics>) -> HandlerResult {
    metrics.accepted.add(1, &[KeyValue::new("region", "eu")]);
    let _ = order;
    HandlerResult::Ack
}

A ready-made Grafana dashboard over exactly this inventory lives in ruststream-grafana: import dashboards/ruststream.json, point it at any Prometheus-compatible backend receiving the OTLP metrics, and the panels light up per handler; its README doubles as the metrics contract.

Call otel.shutdown() at the end of main, after the app's graceful shutdown, to flush the last spans and points. To compose the span bridge into your own subscriber stack (for example with the logging feature's fmt layer), build with .tracing_bridge(false) and install the bridge yourself; .messaging_system("kafka") stamps the semconv system attribute the core cannot derive broker-agnostically.