rust_web_server/metrics/
http.rs

1//! `http.server.request.duration` — the OpenTelemetry HTTP semantic-convention
2//! server metric, recorded by the server core for every request on every
3//! protocol (no layer needed).
4//!
5//! Attributes follow the conventions: `http.request.method` (the known methods,
6//! anything else `_OTHER`), `http.response.status_code`, `url.scheme` and
7//! `network.protocol.version` — no path, so the series count stays bounded.
8//! The duration runs from the moment the request is handed to the app until
9//! its response is ready to write (a streamed body's transfer isn't included).
10//!
11//! Each series keeps the convention's explicit buckets (Prometheus text format,
12//! OTLP) and an exponential, schema-3 copy for Prometheus native histograms
13//! (the protobuf format, [`super::exposition`]). Observations go to one of
14//! [`SHARDS`] locks picked per thread, so threads rarely meet on one.
15
16use std::collections::HashMap;
17use std::sync::atomic::{AtomicUsize, Ordering};
18use std::sync::{Mutex, OnceLock};
19use std::time::Instant;
20
21/// The semantic convention's advised bucket boundaries, in seconds.
22pub(crate) const BOUNDS: [f64; 14] = [0.005, 0.01, 0.025, 0.05, 0.075, 0.1, 0.25, 0.5, 0.75, 1.0, 2.5, 5.0, 7.5, 10.0];
23
24/// Native histogram resolution: 8 buckets per power of two (growth 2^(1/8) ≈ 9%),
25/// client_golang's default.
26pub(crate) const SCHEMA: i32 = 3;
27
28/// Observations at or below this count in the zero bucket (client_golang's default).
29pub(crate) const ZERO_THRESHOLD: f64 = 2.938735877055719e-39; // 2^-128
30
31/// Native bucket indexes are kept within [-200, 160]: about 30 ns to 1 000 000 s.
32const MIN_INDEX: i32 = -200;
33const MAX_INDEX: i32 = 160;
34
35const METHODS: [&str; 9] = ["CONNECT", "DELETE", "GET", "HEAD", "OPTIONS", "PATCH", "POST", "PUT", "TRACE"];
36const OTHER: &str = "_OTHER";
37
38/// The series a request belongs to.
39#[derive(Clone, Copy, PartialEq, Eq, PartialOrd, Ord, Debug)]
40pub(crate) struct Series {
41    /// Index into `METHODS`, or `METHODS.len()` for `_OTHER`.
42    method: u8,
43    pub(crate) status: i16,
44    pub(crate) https: bool,
45    /// `network.protocol.version`: 0 = 1.0, 1 = 1.1, 2 = 2, 3 = 3.
46    version: u8,
47}
48
49impl Series {
50    pub(crate) fn new(method: &str, status: i16, https: bool, http_version: &str) -> Series {
51        let method = METHODS.iter().position(|m| *m == method).unwrap_or(METHODS.len()) as u8;
52        let version = match http_version {
53            "HTTP/1.0" => 0,
54            "HTTP/2" | "HTTP/2.0" => 2,
55            "HTTP/3" | "HTTP/3.0" => 3,
56            _ => 1,
57        };
58        Series { method, status, https, version }
59    }
60
61    /// All four fields in one number (what the store hashes).
62    fn key(&self) -> u32 {
63        (self.method as u32) << 24 | (self.status as u16 as u32) << 8 | (self.https as u32) << 4 | self.version as u32
64    }
65
66    pub(crate) fn method(&self) -> &'static str {
67        METHODS.get(self.method as usize).copied().unwrap_or(OTHER)
68    }
69
70    pub(crate) fn scheme(&self) -> &'static str {
71        if self.https { "https" } else { "http" }
72    }
73
74    pub(crate) fn protocol_version(&self) -> &'static str {
75        ["1.0", "1.1", "2", "3"][self.version as usize]
76    }
77}
78
79/// One series' observations.
80#[derive(Clone, Debug, Default, PartialEq)]
81pub(crate) struct Histogram {
82    pub(crate) count: u64,
83    pub(crate) sum: f64,
84    /// Per explicit bucket (not cumulative): `counts[i]` is `(BOUNDS[i-1], BOUNDS[i]]`,
85    /// the last one `(10, +Inf)`.
86    pub(crate) counts: [u64; BOUNDS.len() + 1],
87    pub(crate) zero_count: u64,
88    /// Native buckets `native_offset ..`, each `(2^((i-1)/8), 2^(i/8)]`.
89    pub(crate) native_offset: i32,
90    pub(crate) native: Vec<u64>,
91}
92
93impl Histogram {
94    pub(crate) fn observe(&mut self, secs: f64) {
95        self.count += 1;
96        self.sum += secs;
97        let slot = BOUNDS.iter().position(|b| secs <= *b).unwrap_or(BOUNDS.len());
98        self.counts[slot] += 1;
99        if secs <= ZERO_THRESHOLD {
100            self.zero_count += 1;
101            return;
102        }
103        self.add_native(native_index(secs), 1);
104    }
105
106    fn add_native(&mut self, index: i32, n: u64) {
107        if self.native.is_empty() {
108            self.native_offset = index;
109            self.native.push(0);
110        } else if index < self.native_offset {
111            let grow = (self.native_offset - index) as usize;
112            self.native.splice(0..0, std::iter::repeat_n(0, grow));
113            self.native_offset = index;
114        } else if index >= self.native_offset + self.native.len() as i32 {
115            self.native.resize((index - self.native_offset + 1) as usize, 0);
116        }
117        self.native[(index - self.native_offset) as usize] += n;
118    }
119
120    fn merge(&mut self, other: &Histogram) {
121        self.count += other.count;
122        self.sum += other.sum;
123        for (mine, theirs) in self.counts.iter_mut().zip(other.counts.iter()) {
124            *mine += theirs;
125        }
126        self.zero_count += other.zero_count;
127        for (i, n) in other.native.iter().enumerate() {
128            if *n == 0 {
129                continue;
130            }
131            self.add_native(other.native_offset + i as i32, *n);
132        }
133    }
134
135    /// The native buckets as Prometheus spans (offset, length) and count deltas:
136    /// runs of empty buckets longer than two start a new span.
137    pub(crate) fn native_spans(&self) -> (Vec<(i32, u32)>, Vec<i64>) {
138        let mut spans: Vec<(i32, u32)> = Vec::new();
139        let mut deltas = Vec::new();
140        let mut previous_count = 0i64;
141        let mut last_index: Option<i32> = None;
142        for (i, n) in self.native.iter().enumerate() {
143            if *n == 0 {
144                continue;
145            }
146            let index = self.native_offset + i as i32;
147            match last_index {
148                Some(last) if index - last <= 3 => {
149                    // Fill a short gap with zero buckets inside the span.
150                    for _ in last + 1..index {
151                        deltas.push(-previous_count);
152                        previous_count = 0;
153                    }
154                    if let Some(span) = spans.last_mut() {
155                        span.1 += (index - last) as u32;
156                    }
157                }
158                Some(last) => spans.push((index - last - 1, 1)),
159                None => spans.push((index, 1)),
160            }
161            deltas.push(*n as i64 - previous_count);
162            previous_count = *n as i64;
163            last_index = Some(index);
164        }
165        (spans, deltas)
166    }
167}
168
169/// Upper bounds of the eight schema-3 buckets within one power of two:
170/// 2^(k/8) for k = 0..=8.
171const MANTISSA_BOUNDS: [f64; 9] = [
172    1.0,
173    1.0905077326652577,
174    1.189207115002721,
175    1.2968395546510096,
176    1.4142135623730951,
177    1.5422108254079407,
178    1.681792830507429,
179    1.8340080864093424,
180    2.0,
181];
182
183/// The schema-3 bucket holding `secs` (> 0): `ceil(log2(secs) · 8)`, clamped —
184/// from the float's exponent and a table lookup on its mantissa (no `log2`).
185pub(crate) fn native_index(secs: f64) -> i32 {
186    let bits = secs.to_bits();
187    let raw_exp = ((bits >> 52) & 0x7ff) as i32;
188    if raw_exp == 0 {
189        return MIN_INDEX; // subnormal: far below anything a request takes
190    }
191    // secs = mantissa · 2^exp, mantissa in [1, 2).
192    let exp = raw_exp - 1023;
193    let mantissa = f64::from_bits((bits & !(0x7ff << 52)) | (1023u64 << 52));
194    let within = MANTISSA_BOUNDS.iter().position(|bound| mantissa <= *bound).unwrap_or(8) as i32;
195    ((exp << SCHEMA) + within).clamp(MIN_INDEX, MAX_INDEX)
196}
197
198// ── store ─────────────────────────────────────────────────────────────────────
199
200const SHARDS: usize = 16;
201
202impl std::hash::Hash for Series {
203    fn hash<H: std::hash::Hasher>(&self, state: &mut H) {
204        state.write_u32(self.key());
205    }
206}
207
208/// Hashes a [`Series`] key with one multiplication (SipHash cost more than
209/// the rest of an observation).
210#[derive(Default)]
211struct SeriesHasher(u64);
212
213impl std::hash::Hasher for SeriesHasher {
214    fn write(&mut self, bytes: &[u8]) {
215        for b in bytes {
216            self.0 = (self.0 << 8 | *b as u64).wrapping_mul(0x9E37_79B9_7F4A_7C15);
217        }
218    }
219    fn write_u32(&mut self, n: u32) {
220        self.0 = (n as u64).wrapping_mul(0x9E37_79B9_7F4A_7C15);
221    }
222    fn finish(&self) -> u64 {
223        self.0
224    }
225}
226
227type Store = HashMap<Series, Histogram, std::hash::BuildHasherDefault<SeriesHasher>>;
228
229fn shards() -> &'static [Mutex<Store>] {
230    static STORE: OnceLock<Vec<Mutex<Store>>> = OnceLock::new();
231    STORE.get_or_init(|| (0..SHARDS).map(|_| Mutex::new(Store::default())).collect())
232}
233
234thread_local! {
235    static SHARD: usize = {
236        static NEXT: AtomicUsize = AtomicUsize::new(0);
237        NEXT.fetch_add(1, Ordering::Relaxed) % SHARDS
238    };
239}
240
241/// Record a finished request: `started` is when it was handed to the app.
242pub fn observe(method: &str, status: i16, https: bool, http_version: &str, started: Instant) {
243    let secs = started.elapsed().as_secs_f64();
244    record(Series::new(method, status, https, http_version), secs);
245}
246
247pub(crate) fn record(series: Series, secs: f64) {
248    let shard = SHARD.with(|s| *s);
249    let mut map = shards()[shard].lock().unwrap_or_else(|e| e.into_inner());
250    map.entry(series).or_default().observe(secs);
251}
252
253/// Every series, merged across shards, in a stable order.
254pub(crate) fn snapshot() -> Vec<(Series, Histogram)> {
255    let mut merged: HashMap<Series, Histogram> = HashMap::new();
256    for shard in shards() {
257        let map = shard.lock().unwrap_or_else(|e| e.into_inner());
258        for (series, histogram) in map.iter() {
259            merged.entry(*series).or_default().merge(histogram);
260        }
261    }
262    let mut all: Vec<_> = merged.into_iter().collect();
263    all.sort_by_key(|(s, _)| *s);
264    all
265}
266
267pub(crate) const NAME: &str = "http_server_request_duration_seconds";
268pub(crate) const HELP: &str = "Duration of HTTP server requests (OpenTelemetry http.server.request.duration)";
269
270/// The series' labels, Prometheus-style (dots become underscores).
271pub(crate) fn labels(series: &Series) -> Vec<(String, String)> {
272    vec![
273        ("http_request_method".to_string(), series.method().to_string()),
274        ("http_response_status_code".to_string(), series.status.to_string()),
275        ("network_protocol_version".to_string(), series.protocol_version().to_string()),
276        ("url_scheme".to_string(), series.scheme().to_string()),
277    ]
278}
279
280/// The classic-histogram text form.
281pub(crate) fn prometheus_text() -> String {
282    let all = snapshot();
283    if all.is_empty() {
284        return String::new();
285    }
286    let mut out = format!("# HELP {NAME} {HELP}\n# TYPE {NAME} histogram\n");
287    for (series, histogram) in &all {
288        let labels: String = labels(series).iter().map(|(k, v)| format!("{k}=\"{v}\",")).collect();
289        let mut cumulative = 0;
290        for (i, bound) in BOUNDS.iter().enumerate() {
291            cumulative += histogram.counts[i];
292            out.push_str(&format!("{NAME}_bucket{{{labels}le=\"{bound}\"}} {cumulative}\n"));
293        }
294        out.push_str(&format!("{NAME}_bucket{{{labels}le=\"+Inf\"}} {}\n", histogram.count));
295        let labels = labels.trim_end_matches(',');
296        out.push_str(&format!("{NAME}_sum{{{labels}}} {}\n", histogram.sum));
297        out.push_str(&format!("{NAME}_count{{{labels}}} {}\n", histogram.count));
298    }
299    out
300}
301
302#[cfg(test)]
303mod tests;
rust_web_server/metrics/http.rs

rust_web_server/metrics/
http.rs
