Propagation

The otel feature carries the W3C trace context from a delivery to its replies and opens a span for every delivery.

OpenTelemetry hands out two pieces. consume_layer is a layer for the app. For each delivery it reads the traceparent and tracestate headers and continues that trace with a new span id: the consumer's span. A delivery without a valid traceparent starts a new sampled trace. The handler runs inside an INFO span named ruststream.consume, which records the subscription and both ids. The layer writes the consumer's context into the delivery's working headers, where a handler reads it: ctx.headers().get_str("traceparent"). propagation is a publish transform for a reply position. It copies traceparent and tracestate from the working headers onto every reply, so the next service continues the same trace with the consumer span as the parent. for_batch(otel.propagation()) puts it on a batch reply. The headers are parsed and written by the OpenTelemetry SDK's W3C propagator, which follows the specification strictly. Propagation works with any tracing subscriber, or none; Export sends the spans to a collector.

Example

Macros

examples/otel_propagation.rs

#[subscriber("orders", reply("confirmations"))]
async fn confirm(order: &Order) -> Confirmation {
    Confirmation { id: order.id }
}

#[ruststream::app]
fn app() -> impl App {
    let otel = OpenTelemetry::new();
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        // A span per delivery, a child of the trace the delivery carries.
        .layer(otel.consume_layer())
        .with_broker(MemoryBroker::new(), |b| {
            // The reply carries the consumer span's trace context.
            b.include(confirm)
                .out_reply(Publish)
                .transform(otel.propagation());
        })
}

Manual

examples/manual/otel_propagation.rs

struct Confirm;

impl Handle<Order, Confirmation> for Confirm {
    fn handle(
        &self,
        order: &Order,
        _outs: &(),
        _ctx: &mut Context<'_>,
    ) -> impl Future<Output = Result<Confirmation, HandlerOutcome>> {
        ready(Ok(Confirmation { id: order.id }))
    }
}

fn app() -> impl App {
    let otel = OpenTelemetry::new();
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        // A span per delivery, a child of the trace the delivery carries.
        .layer(otel.consume_layer())
        .with_broker(MemoryBroker::new(), |b| {
            // The reply carries the consumer span's trace context.
            b.include(
                subscriber("orders", Confirm)
                    .reply()
                    .to("confirmations")
                    .build(),
            )
            .out_reply(Publish)
            .transform(otel.propagation());
        })
}