Skip to main content

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:

InstrumentWhat it measures
messaging.client.consumed.messagesdeliveries received
messaging.process.durationhandler time
ruststream.messages.processedsettlements, by outcome: ack, nack_requeue, nack_drop, retry_after
ruststream.messages.in_flightdeliveries inside handlers
ruststream.message.queue_timetime from publish to handler, from the publish-time header the publish layer stamps
ruststream.messages.decode_failures, ruststream.messages.panicspayloads the codec rejected, handlers that panicked
messaging.client.sent.messages, messaging.client.operation.durationpublishes and their time
ruststream.message.payload.sizepublished payload sizes
ruststream.batch.sizedecoded 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.

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);
        })
}

The handler records the instrument it reads from the state.

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()
}