Prometheus

The metrics feature counts and times deliveries and counts publishes in a Prometheus registry that the service serves on /metrics.

A Metrics value owns the collectors and the registry they are registered in. Metrics::new registers them in the process's default registry, so a process builds one. Metrics::with_registry registers them in a registry of your own, and registry hands the registry out for collectors of your own. consume_layer is a layer for the app: it times every handler and counts its outcome. publish_layer is a publish layer: it counts every publish a handler makes, replies included, and the failed ones too. Metrics is cheap to clone, and every clone writes into the same collectors.

Metric	Type	Labels
`ruststream_messages_consumed_total`	counter	`name`, `status`: `ack` or `nack`
`ruststream_consume_duration_seconds`	histogram	`name`
`ruststream_messages_published_total`	counter	`name`, `status`: `ok` or `error`

name is the subscription on the consume side and the destination on the publish side. With the poll-diagnostics feature, observe_poll_diagnostics adds the time handlers spend in one poll, as described in Diagnostics. export renders the registry in the Prometheus text format. Serving it is the service's part: a route of its HTTP server, or a push to a gateway.

Example

The example serves /metrics on port 8080:

cargo run --example metrics --features testing,macros,memory,json,metrics

Macros
Manual

examples/metrics.rs
#[subscriber("orders", reply("confirmations"))]
async fn confirm(order: &Order) -> Confirmation {
    Confirmation {
        id: order.id,
        accepted: order.quantity > 0,
    }
}

fn app(metrics: &Metrics, broker: MemoryBroker) -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        // Counts and times every delivery, by subscription and outcome.
        .layer(metrics.consume_layer())
        // Counts every publish, the replies included, by destination and outcome.
        .publish_layer(metrics.publish_layer())
        .with_broker(broker, |b| {
            b.include(confirm);
        })
}

examples/manual/metrics.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,
            accepted: order.quantity > 0,
        }))
    }
}

fn app(metrics: &Metrics, broker: MemoryBroker) -> impl App {
    RustStream::new(AppInfo::new("orders", "0.1.0"))
        // Counts and times every delivery, by subscription and outcome.
        .layer(metrics.consume_layer())
        // Counts every publish, the replies included, by destination and outcome.
        .publish_layer(metrics.publish_layer())
        .with_broker(broker, |b| {
            b.include(
                subscriber("orders", Confirm)
                    .reply()
                    .to("confirmations")
                    .build(),
            );
        })
}

The route that serves the registry is an ordinary axum handler.

examples/metrics.rs
/// The `/metrics` route: the registry in the Prometheus text format.
async fn scrape(State(metrics): State<Metrics>) -> Result<String, StatusCode> {
    metrics
        .export()
        .map_err(|_| StatusCode::INTERNAL_SERVER_ERROR)
}

Example​

Example