Export

Otel sends the service's spans and metrics to an OpenTelemetry collector over OTLP, the dispatch metrics included.

Otel::builder starts an OtelBuilder. It takes the service.name, the OTLP endpoint, the messaging.system the dispatch metrics carry, and extra resource attributes such as the environment. An endpoint left unset comes from the standard OTEL_EXPORTER_OTLP_* environment, http://localhost:4317 by default. init builds the exporters and installs the tracer and meter providers as the process's globals. It also installs a tracing subscriber that bridges spans into the tracer, so the spans of Propagation are exported with no further wiring. A service with a subscriber of its own turns the bridge off with tracing_bridge(false) and adds it to its own stack. consume_layer and publish_layer record the dispatch metrics, named by the messaging semantic conventions and labelled per subscription or destination:

Instrument	What it measures
`messaging.client.consumed.messages`	deliveries received
`messaging.process.duration`	handler time
`ruststream.messages.processed`	settlements, by `outcome`: `ack`, `nack_requeue`, `nack_drop`, `retry_after`
`ruststream.messages.in_flight`	deliveries inside handlers
`ruststream.message.queue_time`	time from publish to handler, from the publish-time header the publish layer stamps
`ruststream.messages.decode_failures`, `ruststream.messages.panics`	payloads the codec rejected, handlers that panicked
`messaging.client.sent.messages`, `messaging.client.operation.duration`	publishes and their time
`ruststream.message.payload.size`	published payload sizes
`ruststream.batch.size`	decoded batch sizes, recorded by batch dispatch once `init` has run

The duration histograms use the buckets the conventions advise, 5 ms to 10 s; view changes them per instrument. observe_health adds the ruststream.app.state gauge over the app's health probe. The service's own instruments come from opentelemetry::global::meter and go out through the same pipeline; the example builds them once at startup and shares them as state. Spans and metric points are batched in the background. shutdown flushes them, after the app has stopped; dropping Otel does not. A hand-written main makes that call, which is why the example keeps one with and without the macros. attach takes providers built by the caller and installs no globals: an exporter of your own, or an in-memory one in a test. A Grafana dashboard over these instruments is in ruststream-grafana.

Example

The example exports to a collector on localhost:4317; without one, the exporter retries in the background and the service keeps working:

cargo run --example otel_export --features testing,macros,memory,json,otel -- run

main installs the exporters, runs the app, and flushes at the end.

examples/otel_export.rs
// Written by hand rather than by `#[ruststream::app]`: the exporters batch in the background,
// so `main` flushes them once the app has drained.
fn main() -> Result<ExitCode, Box<dyn Error>> {
    let otel = Otel::builder()
        .service_name("orders")
        .otlp_endpoint("http://localhost:4317")
        .messaging_system("memory")
        .attribute("deployment.environment", "dev")
        .init()?;

    let code = run_main(|| app(&otel));
    // Ships the last spans and metric points; dropping `Otel` does not.
    otel.shutdown()?;
    Ok(code)
}

The app takes both layers and builds the service's own instrument at startup.

Macros
Manual

examples/otel_export.rs
// `use<>`: the app keeps its own handles to the instruments, not a borrow of `otel`.
fn app(otel: &Otel) -> impl App + use<> {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        // Per delivery: consumed, processing time, outcomes, in flight, queue time.
        .layer(otel.consume_layer())
        // Per publish: sent, operation time, payload size, the publish-time stamp.
        .publish_layer(otel.publish_layer())
        .on_startup(async move |()| {
            Ok::<_, Infallible>(AppState {
                metrics: OrderMetrics {
                    accepted: global::meter("orders")
                        .u64_counter("orders_accepted")
                        .build(),
                },
            })
        })
        .with_broker(MemoryBroker::new(), |b| {
            b.include(accept);
        })
}

examples/manual/otel_export.rs
// `use<>`: the app keeps its own handles to the instruments, not a borrow of `otel`.
fn app(otel: &Otel) -> impl App + use<> {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        // Per delivery: consumed, processing time, outcomes, in flight, queue time.
        .layer(otel.consume_layer())
        // Per publish: sent, operation time, payload size, the publish-time stamp.
        .publish_layer(otel.publish_layer())
        .on_startup(async move |()| {
            Ok::<_, Infallible>(AppState {
                metrics: OrderMetrics {
                    accepted: global::meter("orders")
                        .u64_counter("orders_accepted")
                        .build(),
                },
            })
        })
        .with_broker(MemoryBroker::new(), |b| {
            b.include(subscriber("orders", Accept).build());
        })
}

The handler records the instrument it reads from the state.

Macros
Manual

examples/otel_export.rs
/// The service's own instruments, built once at startup against the global meter, so they
/// export through the same pipeline as the 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>) -> HandlerOutcome {
    metrics.accepted.add(1, &[KeyValue::new("region", "eu")]);
    HandlerOutcome::ack()
}

examples/manual/otel_export.rs
/// The service's own instruments, built once at startup against the global meter, so they
/// export through the same pipeline as the dispatch metrics.
#[derive(Clone)]
struct OrderMetrics {
    accepted: Counter<u64>,
}

// A hand-written handler reads the state whole off the context, so it needs no `FromRef`.
#[derive(Clone)]
struct AppState {
    metrics: OrderMetrics,
}

struct Accept;

// The state is the last axis of `Handle`; the mount checks it against the app's.
impl Handle<Order, (), (), (), AppState> for Accept {
    fn handle(
        &self,
        _order: &Order,
        _outs: &(),
        ctx: &mut Context<'_, (), AppState>,
    ) -> impl Future<Output = Result<(), HandlerOutcome>> {
        ctx.state()
            .metrics
            .accepted
            .add(1, &[KeyValue::new("region", "eu")]);
        ready(Ok(()))
    }
}

Example​

Example