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    /// Trace id of [`TraceLayer`](crate::trace::TraceLayer), when installed
147    pub trace_id: Option<String>,
148    pub user_agent: Option<String>,
149}
150
151/// Control characters would let a request forge log lines
152fn clean(s: &str, max: usize) -> String {
153    s.chars()
154        .take(max)
155        .map(|c| if c.is_control() { '?' } else { c })
156        .collect()
157}
158
159impl LogEntry {
160    fn text(&self) -> String {
161        let mut line = format!(
162            "{} {} {} {} {} {} {}ms route={} id={} ua=\"{}\"",
163            iso8601(self.time),
164            self.remote.as_deref().unwrap_or("-"),
165            self.method,
166            self.path,
167            self.status,
168            self.bytes.map_or("-".to_string(), |b| format!("{b}B")),
169            self.duration_ms,
170            self.route.as_deref().unwrap_or("-"),
171            self.request_id.as_deref().unwrap_or("-"),
172            self.user_agent.as_deref().unwrap_or("-").replace('"', "'"),
173        );
174        if let Some(trace) = &self.trace_id {
175            line.push_str(" trace=");
176            line.push_str(trace);
177        }
178        line
179    }
180
181    fn json(&self) -> String {
182        let mut entry = serde_json::json!({
183            "time": iso8601(self.time),
184            "remote": self.remote,
185            "method": self.method,
186            "path": self.path,
187            "route": self.route,
188            "status": self.status,
189            "bytes": self.bytes,
190            "duration_ms": self.duration_ms as u64,
191            "request_id": self.request_id,
192            "user_agent": self.user_agent,
193        });
194        if let Some(trace) = &self.trace_id {
195            entry["trace_id"] = trace.clone().into();
196        }
197        entry.to_string()
198    }
199}
200
201/// One log line per request, written when the response is ready
202///
203/// Lines go to stdout unless a [`sink`](AccessLog::sink) is set. Query strings
204/// are left out by default because they often carry tokens.
205pub struct AccessLog {
206    format: Format,
207    include_query: bool,
208    sink: Arc<dyn Fn(&str) + Send + Sync>,
209    clock: Arc<dyn Fn() -> u64 + Send + Sync>,
210}
211
212impl AccessLog {
213    /// Plain text lines to stdout
214    pub fn new() -> Self {
215        Self {
216            format: Format::Text,
217            include_query: false,
218            sink: Arc::new(|line| println!("{line}")),
219            clock: Arc::new(|| {
220                SystemTime::now()
221                    .duration_since(UNIX_EPOCH)
222                    .map(|d| d.as_secs())
223                    .unwrap_or(0)
224            }),
225        }
226    }
227
228    /// One JSON object per line (for log shippers)
229    pub fn json(mut self) -> Self {
230        self.format = Format::Json;
231        self
232    }
233
234    /// Log the query string too (careful with secrets)
235    pub fn include_query(mut self) -> Self {
236        self.include_query = true;
237        self
238    }
239
240    /// Send lines somewhere else than stdout
241    pub fn sink(mut self, sink: impl Fn(&str) + Send + Sync + 'static) -> Self {
242        self.sink = Arc::new(sink);
243        self
244    }
245
246    /// Replace the time source (tests)
247    pub fn clock(mut self, clock: impl Fn() -> u64 + Send + Sync + 'static) -> Self {
248        self.clock = Arc::new(clock);
249        self
250    }
251}
252
253impl Default for AccessLog {
254    fn default() -> Self {
255        Self::new()
256    }
257}
258
259#[async_trait]
260impl Middleware for AccessLog {
261    async fn handle(&self, req: Request, next: Next) -> Result<Response> {
262        let started = Instant::now();
263        let method = req.method.to_string();
264        let path = if self.include_query {
265            req.uri
266                .path_and_query()
267                .map(|p| p.as_str())
268                .unwrap_or(req.uri.path())
269        } else {
270            req.uri.path()
271        };
272        let path = clean(path, 300);
273        let remote = req.remote_addr.map(|a| a.ip().to_string());
274        let route = req.matched_path.as_deref().map(str::to_owned);
275        let user_agent = req
276            .headers
277            .get(http::header::USER_AGENT)
278            .and_then(|v| v.to_str().ok())
279            .map(|v| clean(v, 200));
280        let mut request_id = req.request_id().map(str::to_owned);
281        let trace_id = req.trace().map(|t| t.trace_id.clone());
282
283        let result = next.run(req).await;
284
285        let (status, bytes) = match &result {
286            Ok(res) => {
287                if request_id.is_none() {
288                    request_id = res
289                        .headers
290                        .get("x-request-id")
291                        .and_then(|v| v.to_str().ok())
292                        .map(str::to_owned);
293                }
294                (
295                    res.status.as_u16(),
296                    (!res.is_streaming()).then_some(res.body.len()),
297                )
298            }
299            Err(err) => (err.status_code(), None),
300        };
301        let entry = LogEntry {
302            time: (self.clock)(),
303            remote,
304            method,
305            path,
306            route,
307            status,
308            bytes,
309            duration_ms: started.elapsed().as_millis(),
310            request_id,
311            trace_id,
312            user_agent,
313        };
314        let line = match self.format {
315            Format::Text => entry.text(),
316            Format::Json => entry.json(),
317        };
318        (self.sink)(&line);
319        result
320    }
321}
322
323// ---- metrics ----
324
325const BUCKETS: [f64; 11] = [
326    0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0,
327];
328/// Distinct (method, route, status) series; more are folded into `route="other"`
329const MAX_SERIES: usize = 1000;
330
331#[derive(Default)]
332struct Histogram {
333    buckets: [u64; BUCKETS.len()],
334    count: u64,
335    sum: f64,
336}
337
338#[derive(Default)]
339struct Inner {
340    requests: Mutex<HashMap<(String, String, u16), u64>>,
341    durations: Mutex<HashMap<(String, String), Histogram>>,
342    in_flight: AtomicI64,
343    started: Option<Instant>,
344}
345
346/// Request counters, latency histograms and an in-flight gauge, rendered in the
347/// Prometheus text format. Clone it: all clones share the numbers.
348#[derive(Clone)]
349pub struct Metrics {
350    inner: Arc<Inner>,
351}
352
353struct InFlight<'a>(&'a AtomicI64);
354
355impl<'a> InFlight<'a> {
356    fn new(counter: &'a AtomicI64) -> Self {
357        counter.fetch_add(1, Ordering::Relaxed);
358        Self(counter)
359    }
360}
361
362impl Drop for InFlight<'_> {
363    fn drop(&mut self) {
364        // also runs when the request future is dropped (client went away)
365        self.0.fetch_sub(1, Ordering::Relaxed);
366    }
367}
368
369fn lock<T>(m: &Mutex<T>) -> MutexGuard<'_, T> {
370    m.lock().unwrap_or_else(|e| e.into_inner())
371}
372
373fn known_method(m: &str) -> &str {
374    match m {
375        "GET" | "HEAD" | "POST" | "PUT" | "PATCH" | "DELETE" | "OPTIONS" => m,
376        _ => "OTHER",
377    }
378}
379
380fn escape_label(v: &str) -> String {
381    v.replace('\\', "\\\\")
382        .replace('"', "\\\"")
383        .replace('\n', "\\n")
384}
385
386impl Metrics {
387    pub fn new() -> Self {
388        Self {
389            inner: Arc::new(Inner {
390                started: Some(Instant::now()),
391                ..Inner::default()
392            }),
393        }
394    }
395
396    /// Record one finished request
397    pub fn observe(&self, method: &str, route: Option<&str>, status: u16, seconds: f64) {
398        let method = known_method(method).to_string();
399        let route = route.unwrap_or("unmatched").to_string();
400        {
401            let mut requests = lock(&self.inner.requests);
402            let key = (method.clone(), route.clone(), status);
403            let key = if requests.contains_key(&key) || requests.len() < MAX_SERIES {
404                key
405            } else {
406                (method.clone(), "other".to_string(), status)
407            };
408            *requests.entry(key).or_insert(0) += 1;
409        }
410        let mut durations = lock(&self.inner.durations);
411        let key = (method.clone(), route);
412        let key = if durations.contains_key(&key) || durations.len() < MAX_SERIES {
413            key
414        } else {
415            (method, "other".to_string())
416        };
417        let hist = durations.entry(key).or_default();
418        hist.count += 1;
419        hist.sum += seconds;
420        for (i, le) in BUCKETS.iter().enumerate() {
421            if seconds <= *le {
422                hist.buckets[i] += 1;
423            }
424        }
425    }
426
427    /// Prometheus text exposition format
428    pub fn render(&self) -> String {
429        let mut out = String::new();
430
431        out.push_str(
432            "# HELP http_requests_total Total HTTP requests by method, route pattern and status.\n",
433        );
434        out.push_str("# TYPE http_requests_total counter\n");
435        let requests = lock(&self.inner.requests);
436        let mut rows: Vec<_> = requests.iter().collect();
437        rows.sort();
438        for ((method, route, status), n) in rows {
439            out.push_str(&format!(
440                "http_requests_total{{method=\"{}\",route=\"{}\",status=\"{}\"}} {}\n",
441                method,
442                escape_label(route),
443                status,
444                n
445            ));
446        }
447        drop(requests);
448
449        out.push_str(
450            "# HELP http_request_duration_seconds Request latency by method and route pattern.\n",
451        );
452        out.push_str("# TYPE http_request_duration_seconds histogram\n");
453        let durations = lock(&self.inner.durations);
454        let mut rows: Vec<_> = durations.iter().collect();
455        rows.sort_by(|a, b| a.0.cmp(b.0));
456        for ((method, route), h) in rows {
457            let labels = format!("method=\"{}\",route=\"{}\"", method, escape_label(route));
458            for (i, le) in BUCKETS.iter().enumerate() {
459                out.push_str(&format!(
460                    "http_request_duration_seconds_bucket{{{labels},le=\"{le}\"}} {}\n",
461                    h.buckets[i]
462                ));
463            }
464            out.push_str(&format!(
465                "http_request_duration_seconds_bucket{{{labels},le=\"+Inf\"}} {}\n",
466                h.count
467            ));
468            out.push_str(&format!(
469                "http_request_duration_seconds_sum{{{labels}}} {}\n",
470                h.sum
471            ));
472            out.push_str(&format!(
473                "http_request_duration_seconds_count{{{labels}}} {}\n",
474                h.count
475            ));
476        }
477        drop(durations);
478
479        out.push_str("# HELP http_requests_in_flight Requests being handled right now.\n");
480        out.push_str("# TYPE http_requests_in_flight gauge\n");
481        out.push_str(&format!(
482            "http_requests_in_flight {}\n",
483            self.inner.in_flight.load(Ordering::Relaxed)
484        ));
485        out.push_str("# HELP process_uptime_seconds Seconds since the metrics were created.\n");
486        out.push_str("# TYPE process_uptime_seconds gauge\n");
487        out.push_str(&format!(
488            "process_uptime_seconds {}\n",
489            self.inner.started.map_or(0, |s| s.elapsed().as_secs())
490        ));
491        out
492    }
493
494    /// A handler serving [`render`](Metrics::render) with the Prometheus content type
495    pub fn handler(&self) -> impl Handler {
496        let metrics = self.clone();
497        move |_req: Request| {
498            let body = metrics.render();
499            async move {
500                Ok(Response::ok()
501                    .with_header("content-type", "text/plain; version=0.0.4; charset=utf-8")
502                    .with_header("cache-control", "no-store")
503                    .with_body(body))
504            }
505        }
506    }
507}
508
509impl Default for Metrics {
510    fn default() -> Self {
511        Self::new()
512    }
513}
514
515#[async_trait]
516impl Middleware for Metrics {
517    async fn handle(&self, req: Request, next: Next) -> Result<Response> {
518        let started = Instant::now();
519        let method = req.method.to_string();
520        let route = req.matched_path.as_deref().map(str::to_owned);
521        let guard = InFlight::new(&self.inner.in_flight);
522
523        let result = next.run(req).await;
524
525        let status = match &result {
526            Ok(res) => res.status.as_u16(),
527            Err(err) => err.status_code(),
528        };
529        drop(guard);
530        self.observe(
531            &method,
532            route.as_deref(),
533            status,
534            started.elapsed().as_secs_f64(),
535        );
536        result
537    }
538}
539
540#[cfg(test)]
541mod tests {
542    use super::*;
543
544    #[test]
545    fn incoming_request_ids_are_validated() {
546        assert!(valid_incoming_id("abc-123_X.y"));
547        assert!(valid_incoming_id(&"a".repeat(64)));
548        for bad in [
549            "",
550            "has space",
551            "new\nline",
552            "semi;colon",
553            "<b>",
554            &"a".repeat(65),
555            "ümlaut",
556        ] {
557            assert!(!valid_incoming_id(bad), "{bad:?}");
558        }
559    }
560
561    #[test]
562    fn log_lines_cannot_be_forged() {
563        assert_eq!(clean("a\nb\r\tc", 100), "a?b??c");
564        assert_eq!(clean(&"x".repeat(500), 10).len(), 10);
565        let e = LogEntry {
566            time: 784_111_777,
567            remote: Some("10.0.0.1".into()),
568            method: "GET".into(),
569            path: "/a".into(),
570            route: Some("/a".into()),
571            status: 200,
572            bytes: Some(12),
573            duration_ms: 3,
574            request_id: Some("rid".into()),
575            trace_id: None,
576            user_agent: Some("curl \"x\"".into()),
577        };
578        assert_eq!(
579            e.text(),
580            "1994-11-06T08:49:37Z 10.0.0.1 GET /a 200 12B 3ms route=/a id=rid ua=\"curl 'x'\""
581        );
582        let j: serde_json::Value = serde_json::from_str(&e.json()).unwrap();
583        assert_eq!(
584            (j["status"].as_u64(), j["bytes"].as_u64()),
585            (Some(200), Some(12))
586        );
587        assert_eq!(j["user_agent"], "curl \"x\"");
588    }
589
590    #[test]
591    fn metrics_render_prometheus_text() {
592        let m = Metrics::new();
593        m.observe("GET", Some("/users/{id}"), 200, 0.004);
594        m.observe("GET", Some("/users/{id}"), 200, 0.3);
595        m.observe("GET", Some("/users/{id}"), 404, 0.02);
596        m.observe("PROPFIND", None, 404, 20.0);
597        let text = m.render();
598
599        assert!(text.contains(
600            "http_requests_total{method=\"GET\",route=\"/users/{id}\",status=\"200\"} 2\n"
601        ));
602        assert!(text.contains(
603            "http_requests_total{method=\"GET\",route=\"/users/{id}\",status=\"404\"} 1\n"
604        ));
605        // unknown methods and unmatched routes get fixed labels
606        assert!(text.contains(
607            "http_requests_total{method=\"OTHER\",route=\"unmatched\",status=\"404\"} 1\n"
608        ));
609        // histogram: cumulative buckets, sum and count
610        let b = |le: &str| {
611            format!("http_request_duration_seconds_bucket{{method=\"GET\",route=\"/users/{{id}}\",le=\"{le}\"}} ")
612        };
613        assert!(text.contains(&format!("{}1\n", b("0.005"))));
614        assert!(text.contains(&format!("{}2\n", b("0.025"))));
615        assert!(text.contains(&format!("{}3\n", b("0.5"))));
616        assert!(text.contains(&format!("{}3\n", b("+Inf"))));
617        assert!(text.contains(
618            "http_request_duration_seconds_count{method=\"GET\",route=\"/users/{id}\"} 3\n"
619        ));
620        // slower than every bucket: only +Inf counts it
621        assert!(text.contains("route=\"unmatched\",le=\"10\"} 0\n"));
622        assert!(text.contains("route=\"unmatched\",le=\"+Inf\"} 1\n"));
623        assert!(text.contains("# TYPE http_requests_total counter\n"));
624        assert!(text.contains("http_requests_in_flight 0\n"));
625    }
626
627    #[test]
628    fn label_values_are_escaped_and_series_are_capped() {
629        assert_eq!(escape_label("a\"b\\c\nd"), "a\\\"b\\\\c\\nd");
630
631        let m = Metrics::new();
632        for i in 0..(MAX_SERIES + 50) {
633            m.observe("GET", Some(&format!("/r{i}")), 200, 0.001);
634        }
635        let text = m.render();
636        let series = text
637            .lines()
638            .filter(|l| l.starts_with("http_requests_total{"))
639            .count();
640        assert!(series <= MAX_SERIES + 1, "{series}");
641        assert!(text.contains("route=\"other\""));
642        // existing series keep counting after the cap is reached
643        m.observe("GET", Some("/r0"), 200, 0.001);
644        assert!(m.render().contains("route=\"/r0\",status=\"200\"} 2\n"));
645    }
646}
ullun/observability.rs

ullun/
observability.rs
