systemprompt_api/services/server/
metrics.rs

1//! Prometheus recorder installation and HTTP metrics rendering.
2//!
3//! Copyright (c) systemprompt.io — Business Source License 1.1.
4//! See <https://systemprompt.io> for licensing details.
5
6use std::sync::{Mutex, OnceLock};
7use std::time::Instant;
8
9use axum::extract::{MatchedPath, Request};
10use axum::http::header::CONTENT_TYPE;
11use axum::middleware::Next;
12use axum::response::{IntoResponse, Response};
13use metrics_exporter_prometheus::{Matcher, PrometheusBuilder, PrometheusHandle};
14use systemprompt_events::{
15    A2A_BROADCASTER, AGUI_BROADCASTER, ANALYTICS_BROADCASTER, Broadcaster, CONTEXT_BROADCASTER,
16};
17use systemprompt_identifiers::InstanceId;
18use systemprompt_traits::OwnedTask;
19
20const METRICS_CONTENT_TYPE: &str = "text/plain; version=0.0.4; charset=utf-8";
21
22const HTTP_REQUESTS_TOTAL: &str = "http_requests_total";
23pub const HTTP_REQUEST_DURATION_SECONDS: &str = "http_request_duration_seconds";
24pub const HTTP_BUCKETS: [f64; 11] = [
25    0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0,
26];
27const HTTP_REQUESTS_IN_FLIGHT: &str = "http_requests_in_flight";
28const SSE_CONNECTIONS: &str = "sse_active_connections";
29
30// Why: metrics::set_global_recorder rejects a second installation in the same
31// process.
32static RECORDER: OnceLock<PrometheusHandle> = OnceLock::new();
33static RECORDER_INIT: Mutex<()> = Mutex::new(());
34
35pub fn install_recorder(instance_id: &InstanceId) -> anyhow::Result<PrometheusHandle> {
36    if let Some(handle) = RECORDER.get() {
37        return Ok(handle.clone());
38    }
39    let _guard = RECORDER_INIT
40        .lock()
41        .unwrap_or_else(std::sync::PoisonError::into_inner);
42    if let Some(handle) = RECORDER.get() {
43        return Ok(handle.clone());
44    }
45    let handle = PrometheusBuilder::new()
46        .add_global_label("instance", instance_id.as_str())
47        .set_buckets_for_metric(
48            Matcher::Full(systemprompt_gateway::GATEWAY_OVERHEAD_SECONDS.to_owned()),
49            &systemprompt_gateway::OVERHEAD_BUCKETS,
50        )?
51        .set_buckets_for_metric(
52            Matcher::Full(systemprompt_gateway::GATEWAY_UPSTREAM_DURATION_SECONDS.to_owned()),
53            &systemprompt_gateway::UPSTREAM_BUCKETS,
54        )?
55        .set_buckets_for_metric(
56            Matcher::Full(HTTP_REQUEST_DURATION_SECONDS.to_owned()),
57            &HTTP_BUCKETS,
58        )?
59        .install_recorder()
60        .map_err(|e| anyhow::anyhow!("failed to install Prometheus recorder: {e}"))?;
61    describe_metrics();
62    Ok(RECORDER.get_or_init(|| handle).clone())
63}
64
65fn describe_metrics() {
66    use crate::services::middleware::load_shed::{
67        HTTP_IN_FLIGHT_LIMIT, HTTP_IN_FLIGHT_SATURATION, HTTP_LOAD_SHED_TOTAL,
68    };
69    super::pool_metrics::describe();
70    metrics::describe_counter!(
71        HTTP_LOAD_SHED_TOTAL,
72        "Requests refused with 503 because the in-flight ceiling was reached"
73    );
74    metrics::describe_gauge!(
75        HTTP_IN_FLIGHT_LIMIT,
76        "Configured server.max_in_flight ceiling"
77    );
78    metrics::describe_gauge!(
79        HTTP_IN_FLIGHT_SATURATION,
80        "Fraction of the in-flight ceiling in use (1.0 = shedding)"
81    );
82    metrics::describe_histogram!(
83        HTTP_REQUEST_DURATION_SECONDS,
84        metrics::Unit::Seconds,
85        "HTTP request duration by method, matched path and status"
86    );
87    metrics::describe_histogram!(
88        systemprompt_gateway::GATEWAY_OVERHEAD_SECONDS,
89        metrics::Unit::Seconds,
90        "Time the gateway added to a completed inference request, excluding the upstream call"
91    );
92    metrics::describe_histogram!(
93        systemprompt_gateway::GATEWAY_UPSTREAM_DURATION_SECONDS,
94        metrics::Unit::Seconds,
95        "Upstream provider call duration for a completed inference request"
96    );
97}
98
99pub fn metrics_router(handle: PrometheusHandle) -> axum::Router {
100    axum::Router::new()
101        .route("/metrics", axum::routing::get(handle_metrics))
102        .with_state(handle)
103}
104
105pub async fn serve_metrics_listener(
106    addr: std::net::SocketAddr,
107    handle: PrometheusHandle,
108) -> anyhow::Result<OwnedTask<()>> {
109    let listener = tokio::net::TcpListener::bind(addr).await?;
110    let mut readiness = super::readiness::get_readiness_receiver();
111    let shutdown = async move {
112        loop {
113            match readiness.recv().await {
114                Ok(super::readiness::ReadinessEvent::ApiShuttingDown) | Err(_) => break,
115                Ok(_) => {},
116            }
117        }
118    };
119    tracing::info!(%addr, "metrics listener bound");
120    Ok(OwnedTask::spawn("metrics_listener", async move {
121        if let Err(error) = axum::serve(listener, metrics_router(handle))
122            .with_graceful_shutdown(shutdown)
123            .await
124        {
125            tracing::warn!(error = %error, "metrics listener exited with error");
126        }
127    }))
128}
129
130pub async fn handle_metrics(
131    axum::extract::State(handle): axum::extract::State<PrometheusHandle>,
132) -> Response {
133    refresh_connection_gauges().await;
134    let body = handle.render();
135    ([(CONTENT_TYPE, METRICS_CONTENT_TYPE)], body).into_response()
136}
137
138async fn refresh_connection_gauges() {
139    let context = CONTEXT_BROADCASTER.total_connections().await;
140    let agui = AGUI_BROADCASTER.total_connections().await;
141    let a2a = A2A_BROADCASTER.total_connections().await;
142    let analytics = ANALYTICS_BROADCASTER.total_connections().await;
143
144    metrics::gauge!(SSE_CONNECTIONS, "channel" => "context").set(context as f64);
145    metrics::gauge!(SSE_CONNECTIONS, "channel" => "agui").set(agui as f64);
146    metrics::gauge!(SSE_CONNECTIONS, "channel" => "a2a").set(a2a as f64);
147    metrics::gauge!(SSE_CONNECTIONS, "channel" => "analytics").set(analytics as f64);
148}
149
150pub async fn track_metrics(req: Request, next: Next) -> Response {
151    let method = req.method().clone();
152    let path = req
153        .extensions()
154        .get::<MatchedPath>()
155        .map_or_else(|| req.uri().path().to_owned(), |m| m.as_str().to_owned());
156
157    let in_flight = metrics::gauge!(HTTP_REQUESTS_IN_FLIGHT);
158    in_flight.increment(1.0);
159
160    let start = Instant::now();
161    let response = next.run(req).await;
162    let elapsed = start.elapsed().as_secs_f64();
163
164    in_flight.decrement(1.0);
165
166    let status = response.status().as_u16().to_string();
167    let method = method.to_string();
168
169    metrics::counter!(
170        HTTP_REQUESTS_TOTAL,
171        "method" => method.clone(),
172        "path" => path.clone(),
173        "status" => status.clone(),
174    )
175    .increment(1);
176    metrics::histogram!(
177        HTTP_REQUEST_DURATION_SECONDS,
178        "method" => method,
179        "path" => path,
180        "status" => status,
181    )
182    .record(elapsed);
183
184    response
185}
systemprompt_api/services/server/metrics.rs

systemprompt_api/services/server/
metrics.rs
