edda_server/
measure.rs

1//! What a service measures, for Prometheus: its requests by route and
2//! status and how long they take, what the limits refuse, and its
3//! database connections; services add their own (the combat service's
4//! WebSockets, say). Served at `/metrics` on `METRICS_ADDR`, an address of
5//! its own that the proxy never serves; without it nothing is measured.
6
7use std::{sync::OnceLock, time::Instant};
8
9use axum::{
10    Router,
11    extract::{MatchedPath, Request},
12    middleware::{self, Next},
13    response::Response,
14    routing::get,
15};
16use metrics_exporter_prometheus::{BuildError, Matcher, PrometheusBuilder, PrometheusHandle};
17use sqlx::PgPool;
18
19/// How long requests take, in seconds: the histogram's buckets.
20const DURATION_BUCKETS: [f64; 10] = [0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 10.0];
21
22/// The database whose connections are measured, from [`watch_pool`].
23static POOL: OnceLock<PgPool> = OnceLock::new();
24
25/// Measures `service` from now on, everything labelled with its name; the
26/// handle renders what was measured. Once per process.
27pub fn install(service: &str) -> Result<PrometheusHandle, BuildError> {
28    builder(service)?.install_recorder()
29}
30
31fn builder(service: &str) -> Result<PrometheusBuilder, BuildError> {
32    PrometheusBuilder::new()
33        .add_global_label("service", service)
34        .set_buckets_for_metric(
35            Matcher::Full("edda_http_request_duration_seconds".into()),
36            &DURATION_BUCKETS,
37        )
38}
39
40/// `GET /metrics`: what was measured, in Prometheus' text format.
41pub fn endpoint(handle: PrometheusHandle) -> Router {
42    Router::new().route(
43        "/metrics",
44        get(move || {
45            let handle = handle.clone();
46            async move {
47                measure_pool();
48                handle.run_upkeep();
49                handle.render()
50            }
51        }),
52    )
53}
54
55/// Counts and times `router`'s requests, by the route they matched (its
56/// template, such as `/characters/{id}`, not the ids themselves).
57pub fn requests(router: Router) -> Router {
58    router.route_layer(middleware::from_fn(time))
59}
60
61async fn time(request: Request, next: Next) -> Response {
62    let started = Instant::now();
63    let method = request.method().to_string();
64    let route = request
65        .extensions()
66        .get::<MatchedPath>()
67        .map_or_else(String::new, |path| path.as_str().to_owned());
68    let response = next.run(request).await;
69    let status = response.status().as_u16().to_string();
70    metrics::counter!(
71        "edda_http_requests_total",
72        "method" => method.clone(),
73        "route" => route.clone(),
74        "status" => status,
75    )
76    .increment(1);
77    metrics::histogram!(
78        "edda_http_request_duration_seconds",
79        "method" => method,
80        "route" => route,
81    )
82    .record(started.elapsed().as_secs_f64());
83    response
84}
85
86/// Counts a request a limit refused: `limit` is which.
87pub fn refused(limit: &str) {
88    metrics::counter!("edda_limits_refused_total", "limit" => limit.to_owned()).increment(1);
89}
90
91/// Measures `pool`'s connections whenever the measurements are read. The
92/// first pool a service watches is the one measured.
93pub fn watch_pool(pool: &PgPool) {
94    let _ = POOL.set(pool.clone());
95}
96
97fn measure_pool() {
98    let Some(pool) = POOL.get() else { return };
99    let idle = pool.num_idle();
100    let open = usize::try_from(pool.size()).unwrap_or(usize::MAX);
101    gauge_of("edda_db_connections", "idle", idle);
102    gauge_of("edda_db_connections", "busy", open.saturating_sub(idle));
103}
104
105fn gauge_of(name: &'static str, state: &'static str, count: usize) {
106    metrics::gauge!(name, "state" => state).set(as_f64(count));
107}
108
109/// `count` as a gauge's value, at most `u32::MAX`: far beyond any count here.
110pub fn as_f64(count: usize) -> f64 {
111    u32::try_from(count).map_or(f64::from(u32::MAX), f64::from)
112}
113
114/// The recorder for this crate's tests, installed once for the process.
115#[cfg(test)]
116fn test_handle() -> PrometheusHandle {
117    static HANDLE: std::sync::Mutex<Option<PrometheusHandle>> = std::sync::Mutex::new(None);
118    HANDLE
119        .lock()
120        .unwrap()
121        .get_or_insert_with(|| builder("test").unwrap().install_recorder().unwrap())
122        .clone()
123}
124
125#[cfg(test)]
126mod tests {
127    use axum::{body::Body, http::StatusCode};
128    use http_body_util::BodyExt;
129    use sqlx::postgres::PgPoolOptions;
130    use tower::ServiceExt;
131
132    use super::*;
133
134    async fn read(endpoint: &Router) -> String {
135        let response = endpoint
136            .clone()
137            .oneshot(Request::get("/metrics").body(Body::empty()).unwrap())
138            .await
139            .unwrap();
140        assert_eq!(response.status(), StatusCode::OK);
141        let bytes = response.into_body().collect().await.unwrap().to_bytes();
142        String::from_utf8(bytes.to_vec()).unwrap()
143    }
144
145    #[tokio::test]
146    async fn requests_refusals_and_connections_are_measured() {
147        let endpoint = endpoint(test_handle());
148        let app = requests(Router::new().route("/characters/{id}", get(|| async { "ok" })));
149        for id in ["grog", "vex"] {
150            let request = Request::get(format!("/characters/{id}")).body(Body::empty());
151            app.clone().oneshot(request.unwrap()).await.unwrap();
152        }
153        refused("sign-ins");
154        watch_pool(
155            &PgPoolOptions::new()
156                .connect_lazy("postgres://postgres@127.0.0.1:1/nothing")
157                .unwrap(),
158        );
159
160        let measured = read(&endpoint).await;
161        let line = |start: &str| {
162            measured
163                .lines()
164                .find(|l| l.starts_with(start))
165                .map(str::to_owned)
166        };
167        let requests = line("edda_http_requests_total{").unwrap();
168        assert!(
169            requests.contains(r#"route="/characters/{id}""#),
170            "{requests}"
171        );
172        assert!(
173            requests.contains(r#"status="200""#) && requests.ends_with(" 2"),
174            "{requests}"
175        );
176        assert!(requests.contains(r#"service="test""#), "{requests}");
177        assert!(!measured.contains("grog"), "routes, not ids");
178        assert!(line("edda_http_request_duration_seconds_bucket{").is_some());
179        assert!(
180            line("edda_limits_refused_total{")
181                .unwrap()
182                .contains(r#"limit="sign-ins""#)
183        );
184        assert!(line("edda_db_connections{").is_some(), "{measured}");
185    }
186
187    #[test]
188    fn counts_become_gauge_values() {
189        assert!((as_f64(3) - 3.0).abs() < f64::EPSILON);
190        assert!((as_f64(usize::MAX) - f64::from(u32::MAX)).abs() < f64::EPSILON);
191    }
192}
edda_server/measure.rs

edda_server/
measure.rs
