ullun/
observability.rs

1//! Request ids, access logs and Prometheus metrics
2//!
3//! ```no_run
4//! use ullun::observability::{AccessLog, Metrics, RequestIdLayer};
5//! use ullun::prelude::*;
6//!
7//! # async fn run() {
8//! let metrics = Metrics::new();
9//! App::new()
10//!     .middleware(RequestIdLayer::new())      // first: the others read the id
11//!     .middleware(AccessLog::new())           // one line per request
12//!     .middleware(metrics.clone())            // counters and latency histograms
13//!     .get("/metrics", metrics.handler())     // Prometheus text format
14//!     .get("/users/{id}", |req: Request| async move {
15//!         // requests of this route are logged and counted as `/users/{id}`,
16//!         // not once per id
17//!         Ok(Response::text(req.request_id().unwrap_or("-").to_string()))
18//!     })
19//!     .run("127.0.0.1:3000")
20//!     .await
21//!     .unwrap();
22//! # }
23//! ```
24//!
25//! Metric labels use [`Request::matched_path`], the route *pattern*, so the
26//! number of series is bounded by your routes, not by the ids or by whatever
27//! paths scanners probe (all unmatched requests share `route="unmatched"`).
28//! Do not expose `/metrics` publicly without authentication or a network
29//! boundary.
30
31use crate::error::Result;
32use crate::files::iso8601;
33use crate::handler::Handler;
34use crate::middleware::{Middleware, Next};
35use crate::request::Request;
36use crate::response::Response;
37use crate::security::random_token;
38use async_trait::async_trait;
39use std::collections::HashMap;
40use std::sync::atomic::{AtomicI64, Ordering};
41use std::sync::{Arc, Mutex, MutexGuard};
42use std::time::{Instant, SystemTime, UNIX_EPOCH};
43
44// ---- request ids ----
45
46/// The id of the current request, stored in `Request::extensions`
47#[derive(Debug, Clone, PartialEq, Eq)]
48pub struct RequestId(pub String);
49
50impl Request {
51    /// Id assigned by [`RequestIdLayer`], if installed
52    pub fn request_id(&self) -> Option<&str> {
53        self.extensions.get::<RequestId>().map(|r| r.0.as_str())
54    }
55}
56
57fn valid_incoming_id(id: &str) -> bool {
58    !id.is_empty()
59        && id.len() <= 64
60        && id
61            .bytes()
62            .all(|b| b.is_ascii_alphanumeric() || b"-_.".contains(&b))
63}
64
65/// Assigns every request an id (or keeps a sane one sent by the caller),
66/// exposes it as [`Request::request_id`] and echoes it in a response header
67pub struct RequestIdLayer {
68    header: String,
69    trust_incoming: bool,
70}
71
72impl RequestIdLayer {
73    /// Header `x-request-id`, incoming ids accepted
74    pub fn new() -> Self {
75        Self {
76            header: "x-request-id".into(),
77            trust_incoming: true,
78        }
79    }
80
81    pub fn header_name(mut self, name: impl Into<String>) -> Self {
82        self.header = name.into().to_ascii_lowercase();
83        self
84    }
85
86    /// Always generate a fresh id, even when the caller sent one (edge servers
87    /// facing the internet)
88    pub fn ignore_incoming(mut self) -> Self {
89        self.trust_incoming = false;
90        self
91    }
92}
93
94impl Default for RequestIdLayer {
95    fn default() -> Self {
96        Self::new()
97    }
98}
99
100#[async_trait]
101impl Middleware for RequestIdLayer {
102    async fn handle(&self, mut req: Request, next: Next) -> Result<Response> {
103        let incoming = self
104            .trust_incoming
105            .then(|| {
106                req.headers
107                    .get(self.header.as_str())
108                    .and_then(|v| v.to_str().ok())
109                    .filter(|v| valid_incoming_id(v))
110                    .map(str::to_owned)
111            })
112            .flatten();
113        let id = match incoming {
114            Some(id) => id,
115            None => random_token(8)?,
116        };
117        req.extensions.insert(RequestId(id.clone()));
118        let res = match next.run(req).await {
119            Ok(res) => res,
120            Err(err) => Response::from_error(err),
121        };
122        Ok(res.with_header(self.header.clone(), id))
123    }
124}
125
126// ---- access log ----
127
128#[derive(Clone, Copy, PartialEq, Eq)]
129enum Format {
130    Text,
131    Json,
132}
133
134/// One finished request, as handed to the log sink
135#[derive(Debug, Clone)]
136pub struct LogEntry {
137    pub time: u64,
138    pub remote: Option<String>,
139    pub method: String,
140    pub path: String,
141    pub route: Option<String>,
142    pub status: u16,
143    pub bytes: Option<usize>,
144    pub duration_ms: u128,
145    pub request_id: Option<String>,
146    pub user_agent: Option<String>,
147}
148
149/// Control characters would let a request forge log lines
150fn clean(s: &str, max: usize) -> String {
151    s.chars()
152        .take(max)
153        .map(|c| if c.is_control() { '?' } else { c })
154        .collect()
155}
156
157impl LogEntry {
158    fn text(&self) -> String {
159        format!(
160            "{} {} {} {} {} {} {}ms route={} id={} ua=\"{}\"",
161            iso8601(self.time),
162            self.remote.as_deref().unwrap_or("-"),
163            self.method,
164            self.path,
165            self.status,
166            self.bytes.map_or("-".to_string(), |b| format!("{b}B")),
167            self.duration_ms,
168            self.route.as_deref().unwrap_or("-"),
169            self.request_id.as_deref().unwrap_or("-"),
170            self.user_agent.as_deref().unwrap_or("-").replace('"', "'"),
171        )
172    }
173
174    fn json(&self) -> String {
175        serde_json::json!({
176            "time": iso8601(self.time),
177            "remote": self.remote,
178            "method": self.method,
179            "path": self.path,
180            "route": self.route,
181            "status": self.status,
182            "bytes": self.bytes,
183            "duration_ms": self.duration_ms as u64,
184            "request_id": self.request_id,
185            "user_agent": self.user_agent,
186        })
187        .to_string()
188    }
189}
190
191/// One log line per request, written when the response is ready
192///
193/// Lines go to stdout unless a [`sink`](AccessLog::sink) is set. Query strings
194/// are left out by default because they often carry tokens.
195pub struct AccessLog {
196    format: Format,
197    include_query: bool,
198    sink: Arc<dyn Fn(&str) + Send + Sync>,
199    clock: Arc<dyn Fn() -> u64 + Send + Sync>,
200}
201
202impl AccessLog {
203    /// Plain text lines to stdout
204    pub fn new() -> Self {
205        Self {
206            format: Format::Text,
207            include_query: false,
208            sink: Arc::new(|line| println!("{line}")),
209            clock: Arc::new(|| {
210                SystemTime::now()
211                    .duration_since(UNIX_EPOCH)
212                    .map(|d| d.as_secs())
213                    .unwrap_or(0)
214            }),
215        }
216    }
217
218    /// One JSON object per line (for log shippers)
219    pub fn json(mut self) -> Self {
220        self.format = Format::Json;
221        self
222    }
223
224    /// Log the query string too (careful with secrets)
225    pub fn include_query(mut self) -> Self {
226        self.include_query = true;
227        self
228    }
229
230    /// Send lines somewhere else than stdout
231    pub fn sink(mut self, sink: impl Fn(&str) + Send + Sync + 'static) -> Self {
232        self.sink = Arc::new(sink);
233        self
234    }
235
236    /// Replace the time source (tests)
237    pub fn clock(mut self, clock: impl Fn() -> u64 + Send + Sync + 'static) -> Self {
238        self.clock = Arc::new(clock);
239        self
240    }
241}
242
243impl Default for AccessLog {
244    fn default() -> Self {
245        Self::new()
246    }
247}
248
249#[async_trait]
250impl Middleware for AccessLog {
251    async fn handle(&self, req: Request, next: Next) -> Result<Response> {
252        let started = Instant::now();
253        let method = req.method.to_string();
254        let path = if self.include_query {
255            req.uri
256                .path_and_query()
257                .map(|p| p.as_str())
258                .unwrap_or(req.uri.path())
259        } else {
260            req.uri.path()
261        };
262        let path = clean(path, 300);
263        let remote = req.remote_addr.map(|a| a.ip().to_string());
264        let route = req.matched_path.as_deref().map(str::to_owned);
265        let user_agent = req
266            .headers
267            .get(http::header::USER_AGENT)
268            .and_then(|v| v.to_str().ok())
269            .map(|v| clean(v, 200));
270        let mut request_id = req.request_id().map(str::to_owned);
271
272        let result = next.run(req).await;
273
274        let (status, bytes) = match &result {
275            Ok(res) => {
276                if request_id.is_none() {
277                    request_id = res
278                        .headers
279                        .get("x-request-id")
280                        .and_then(|v| v.to_str().ok())
281                        .map(str::to_owned);
282                }
283                (
284                    res.status.as_u16(),
285                    (!res.is_streaming()).then_some(res.body.len()),
286                )
287            }
288            Err(err) => (err.status_code(), None),
289        };
290        let entry = LogEntry {
291            time: (self.clock)(),
292            remote,
293            method,
294            path,
295            route,
296            status,
297            bytes,
298            duration_ms: started.elapsed().as_millis(),
299            request_id,
300            user_agent,
301        };
302        let line = match self.format {
303            Format::Text => entry.text(),
304            Format::Json => entry.json(),
305        };
306        (self.sink)(&line);
307        result
308    }
309}
310
311// ---- metrics ----
312
313const BUCKETS: [f64; 11] = [
314    0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0,
315];
316/// Distinct (method, route, status) series; more are folded into `route="other"`
317const MAX_SERIES: usize = 1000;
318
319#[derive(Default)]
320struct Histogram {
321    buckets: [u64; BUCKETS.len()],
322    count: u64,
323    sum: f64,
324}
325
326#[derive(Default)]
327struct Inner {
328    requests: Mutex<HashMap<(String, String, u16), u64>>,
329    durations: Mutex<HashMap<(String, String), Histogram>>,
330    in_flight: AtomicI64,
331    started: Option<Instant>,
332}
333
334/// Request counters, latency histograms and an in-flight gauge, rendered in the
335/// Prometheus text format. Clone it: all clones share the numbers.
336#[derive(Clone)]
337pub struct Metrics {
338    inner: Arc<Inner>,
339}
340
341struct InFlight<'a>(&'a AtomicI64);
342
343impl<'a> InFlight<'a> {
344    fn new(counter: &'a AtomicI64) -> Self {
345        counter.fetch_add(1, Ordering::Relaxed);
346        Self(counter)
347    }
348}
349
350impl Drop for InFlight<'_> {
351    fn drop(&mut self) {
352        // also runs when the request future is dropped (client went away)
353        self.0.fetch_sub(1, Ordering::Relaxed);
354    }
355}
356
357fn lock<T>(m: &Mutex<T>) -> MutexGuard<'_, T> {
358    m.lock().unwrap_or_else(|e| e.into_inner())
359}
360
361fn known_method(m: &str) -> &str {
362    match m {
363        "GET" | "HEAD" | "POST" | "PUT" | "PATCH" | "DELETE" | "OPTIONS" => m,
364        _ => "OTHER",
365    }
366}
367
368fn escape_label(v: &str) -> String {
369    v.replace('\\', "\\\\")
370        .replace('"', "\\\"")
371        .replace('\n', "\\n")
372}
373
374impl Metrics {
375    pub fn new() -> Self {
376        Self {
377            inner: Arc::new(Inner {
378                started: Some(Instant::now()),
379                ..Inner::default()
380            }),
381        }
382    }
383
384    /// Record one finished request
385    pub fn observe(&self, method: &str, route: Option<&str>, status: u16, seconds: f64) {
386        let method = known_method(method).to_string();
387        let route = route.unwrap_or("unmatched").to_string();
388        {
389            let mut requests = lock(&self.inner.requests);
390            let key = (method.clone(), route.clone(), status);
391            let key = if requests.contains_key(&key) || requests.len() < MAX_SERIES {
392                key
393            } else {
394                (method.clone(), "other".to_string(), status)
395            };
396            *requests.entry(key).or_insert(0) += 1;
397        }
398        let mut durations = lock(&self.inner.durations);
399        let key = (method.clone(), route);
400        let key = if durations.contains_key(&key) || durations.len() < MAX_SERIES {
401            key
402        } else {
403            (method, "other".to_string())
404        };
405        let hist = durations.entry(key).or_default();
406        hist.count += 1;
407        hist.sum += seconds;
408        for (i, le) in BUCKETS.iter().enumerate() {
409            if seconds <= *le {
410                hist.buckets[i] += 1;
411            }
412        }
413    }
414
415    /// Prometheus text exposition format
416    pub fn render(&self) -> String {
417        let mut out = String::new();
418
419        out.push_str(
420            "# HELP http_requests_total Total HTTP requests by method, route pattern and status.\n",
421        );
422        out.push_str("# TYPE http_requests_total counter\n");
423        let requests = lock(&self.inner.requests);
424        let mut rows: Vec<_> = requests.iter().collect();
425        rows.sort();
426        for ((method, route, status), n) in rows {
427            out.push_str(&format!(
428                "http_requests_total{{method=\"{}\",route=\"{}\",status=\"{}\"}} {}\n",
429                method,
430                escape_label(route),
431                status,
432                n
433            ));
434        }
435        drop(requests);
436
437        out.push_str(
438            "# HELP http_request_duration_seconds Request latency by method and route pattern.\n",
439        );
440        out.push_str("# TYPE http_request_duration_seconds histogram\n");
441        let durations = lock(&self.inner.durations);
442        let mut rows: Vec<_> = durations.iter().collect();
443        rows.sort_by(|a, b| a.0.cmp(b.0));
444        for ((method, route), h) in rows {
445            let labels = format!("method=\"{}\",route=\"{}\"", method, escape_label(route));
446            for (i, le) in BUCKETS.iter().enumerate() {
447                out.push_str(&format!(
448                    "http_request_duration_seconds_bucket{{{labels},le=\"{le}\"}} {}\n",
449                    h.buckets[i]
450                ));
451            }
452            out.push_str(&format!(
453                "http_request_duration_seconds_bucket{{{labels},le=\"+Inf\"}} {}\n",
454                h.count
455            ));
456            out.push_str(&format!(
457                "http_request_duration_seconds_sum{{{labels}}} {}\n",
458                h.sum
459            ));
460            out.push_str(&format!(
461                "http_request_duration_seconds_count{{{labels}}} {}\n",
462                h.count
463            ));
464        }
465        drop(durations);
466
467        out.push_str("# HELP http_requests_in_flight Requests being handled right now.\n");
468        out.push_str("# TYPE http_requests_in_flight gauge\n");
469        out.push_str(&format!(
470            "http_requests_in_flight {}\n",
471            self.inner.in_flight.load(Ordering::Relaxed)
472        ));
473        out.push_str("# HELP process_uptime_seconds Seconds since the metrics were created.\n");
474        out.push_str("# TYPE process_uptime_seconds gauge\n");
475        out.push_str(&format!(
476            "process_uptime_seconds {}\n",
477            self.inner.started.map_or(0, |s| s.elapsed().as_secs())
478        ));
479        out
480    }
481
482    /// A handler serving [`render`](Metrics::render) with the Prometheus content type
483    pub fn handler(&self) -> impl Handler {
484        let metrics = self.clone();
485        move |_req: Request| {
486            let body = metrics.render();
487            async move {
488                Ok(Response::ok()
489                    .with_header("content-type", "text/plain; version=0.0.4; charset=utf-8")
490                    .with_header("cache-control", "no-store")
491                    .with_body(body))
492            }
493        }
494    }
495}
496
497impl Default for Metrics {
498    fn default() -> Self {
499        Self::new()
500    }
501}
502
503#[async_trait]
504impl Middleware for Metrics {
505    async fn handle(&self, req: Request, next: Next) -> Result<Response> {
506        let started = Instant::now();
507        let method = req.method.to_string();
508        let route = req.matched_path.as_deref().map(str::to_owned);
509        let guard = InFlight::new(&self.inner.in_flight);
510
511        let result = next.run(req).await;
512
513        let status = match &result {
514            Ok(res) => res.status.as_u16(),
515            Err(err) => err.status_code(),
516        };
517        drop(guard);
518        self.observe(
519            &method,
520            route.as_deref(),
521            status,
522            started.elapsed().as_secs_f64(),
523        );
524        result
525    }
526}
527
528#[cfg(test)]
529mod tests {
530    use super::*;
531
532    #[test]
533    fn incoming_request_ids_are_validated() {
534        assert!(valid_incoming_id("abc-123_X.y"));
535        assert!(valid_incoming_id(&"a".repeat(64)));
536        for bad in [
537            "",
538            "has space",
539            "new\nline",
540            "semi;colon",
541            "<b>",
542            &"a".repeat(65),
543            "ümlaut",
544        ] {
545            assert!(!valid_incoming_id(bad), "{bad:?}");
546        }
547    }
548
549    #[test]
550    fn log_lines_cannot_be_forged() {
551        assert_eq!(clean("a\nb\r\tc", 100), "a?b??c");
552        assert_eq!(clean(&"x".repeat(500), 10).len(), 10);
553        let e = LogEntry {
554            time: 784_111_777,
555            remote: Some("10.0.0.1".into()),
556            method: "GET".into(),
557            path: "/a".into(),
558            route: Some("/a".into()),
559            status: 200,
560            bytes: Some(12),
561            duration_ms: 3,
562            request_id: Some("rid".into()),
563            user_agent: Some("curl \"x\"".into()),
564        };
565        assert_eq!(
566            e.text(),
567            "1994-11-06T08:49:37Z 10.0.0.1 GET /a 200 12B 3ms route=/a id=rid ua=\"curl 'x'\""
568        );
569        let j: serde_json::Value = serde_json::from_str(&e.json()).unwrap();
570        assert_eq!(
571            (j["status"].as_u64(), j["bytes"].as_u64()),
572            (Some(200), Some(12))
573        );
574        assert_eq!(j["user_agent"], "curl \"x\"");
575    }
576
577    #[test]
578    fn metrics_render_prometheus_text() {
579        let m = Metrics::new();
580        m.observe("GET", Some("/users/{id}"), 200, 0.004);
581        m.observe("GET", Some("/users/{id}"), 200, 0.3);
582        m.observe("GET", Some("/users/{id}"), 404, 0.02);
583        m.observe("PROPFIND", None, 404, 20.0);
584        let text = m.render();
585
586        assert!(text.contains(
587            "http_requests_total{method=\"GET\",route=\"/users/{id}\",status=\"200\"} 2\n"
588        ));
589        assert!(text.contains(
590            "http_requests_total{method=\"GET\",route=\"/users/{id}\",status=\"404\"} 1\n"
591        ));
592        // unknown methods and unmatched routes get fixed labels
593        assert!(text.contains(
594            "http_requests_total{method=\"OTHER\",route=\"unmatched\",status=\"404\"} 1\n"
595        ));
596        // histogram: cumulative buckets, sum and count
597        let b = |le: &str| {
598            format!("http_request_duration_seconds_bucket{{method=\"GET\",route=\"/users/{{id}}\",le=\"{le}\"}} ")
599        };
600        assert!(text.contains(&format!("{}1\n", b("0.005"))));
601        assert!(text.contains(&format!("{}2\n", b("0.025"))));
602        assert!(text.contains(&format!("{}3\n", b("0.5"))));
603        assert!(text.contains(&format!("{}3\n", b("+Inf"))));
604        assert!(text.contains(
605            "http_request_duration_seconds_count{method=\"GET\",route=\"/users/{id}\"} 3\n"
606        ));
607        // slower than every bucket: only +Inf counts it
608        assert!(text.contains("route=\"unmatched\",le=\"10\"} 0\n"));
609        assert!(text.contains("route=\"unmatched\",le=\"+Inf\"} 1\n"));
610        assert!(text.contains("# TYPE http_requests_total counter\n"));
611        assert!(text.contains("http_requests_in_flight 0\n"));
612    }
613
614    #[test]
615    fn label_values_are_escaped_and_series_are_capped() {
616        assert_eq!(escape_label("a\"b\\c\nd"), "a\\\"b\\\\c\\nd");
617
618        let m = Metrics::new();
619        for i in 0..(MAX_SERIES + 50) {
620            m.observe("GET", Some(&format!("/r{i}")), 200, 0.001);
621        }
622        let text = m.render();
623        let series = text
624            .lines()
625            .filter(|l| l.starts_with("http_requests_total{"))
626            .count();
627        assert!(series <= MAX_SERIES + 1, "{series}");
628        assert!(text.contains("route=\"other\""));
629        // existing series keep counting after the cap is reached
630        m.observe("GET", Some("/r0"), 200, 0.001);
631        assert!(m.render().contains("route=\"/r0\",status=\"200\"} 2\n"));
632    }
633}
ullun/observability.rs

ullun/
observability.rs
