leyline/trace/
metrics.rs

1use std::fmt;
2use std::sync::Arc;
3use std::sync::atomic::{AtomicU64, Ordering};
4use std::time::Duration;
5
6use super::{BodyEnd, BodyOutcome, Summary, Trace};
7use crate::ErrorCategory;
8
9const LATENCY_BOUNDS_MS: [u64; 9] = [10, 50, 100, 250, 500, 1_000, 2_500, 5_000, 10_000];
10const LATENCY_BUCKETS: usize = LATENCY_BOUNDS_MS.len() + 1;
11const STATUS_CLASSES: usize = 5;
12const CATEGORIES: usize = ErrorCategory::ALL.len();
13
14#[derive(Clone, Copy)]
15enum BodyKind {
16    Complete,
17    Failed,
18    Dropped,
19}
20
21impl BodyKind {
22    const ALL: [Self; 3] = [Self::Complete, Self::Failed, Self::Dropped];
23
24    fn of(outcome: &BodyOutcome<'_>) -> Self {
25        match outcome {
26            BodyOutcome::Complete => Self::Complete,
27            BodyOutcome::Failed(_) => Self::Failed,
28            BodyOutcome::Dropped => Self::Dropped,
29        }
30    }
31
32    fn label(self) -> &'static str {
33        match self {
34            Self::Complete => "complete",
35            Self::Failed => "failed",
36            Self::Dropped => "dropped",
37        }
38    }
39}
40
41#[derive(Debug, Default)]
42pub struct Metrics {
43    requests: AtomicU64,
44    attempts: AtomicU64,
45    status: [AtomicU64; STATUS_CLASSES],
46    errors: [AtomicU64; CATEGORIES],
47    bodies: [AtomicU64; BodyKind::ALL.len()],
48    latency: [AtomicU64; LATENCY_BUCKETS],
49}
50
51#[derive(Debug, Clone, Default, PartialEq, Eq)]
52pub struct MetricsSnapshot {
53    requests: u64,
54    attempts: u64,
55    status: [u64; STATUS_CLASSES],
56    errors: [u64; CATEGORIES],
57    bodies: [u64; BodyKind::ALL.len()],
58    latency: [u64; LATENCY_BUCKETS],
59}
60
61fn bump(counter: &AtomicU64, by: u64) {
62    counter.fetch_add(by, Ordering::Relaxed);
63}
64
65fn load<const N: usize>(counters: &[AtomicU64; N]) -> [u64; N] {
66    std::array::from_fn(|i| counters[i].load(Ordering::Relaxed))
67}
68
69fn latency_bucket(elapsed: Duration) -> usize {
70    let ms = u64::try_from(elapsed.as_millis()).unwrap_or(u64::MAX);
71    LATENCY_BOUNDS_MS
72        .iter()
73        .position(|bound| ms <= *bound)
74        .unwrap_or(LATENCY_BOUNDS_MS.len())
75}
76
77impl Metrics {
78    pub fn new() -> Arc<Self> {
79        Arc::new(Self::default())
80    }
81
82    pub fn snapshot(&self) -> MetricsSnapshot {
83        MetricsSnapshot {
84            requests: self.requests.load(Ordering::Relaxed),
85            attempts: self.attempts.load(Ordering::Relaxed),
86            status: load(&self.status),
87            errors: load(&self.errors),
88            bodies: load(&self.bodies),
89            latency: load(&self.latency),
90        }
91    }
92}
93
94impl Trace for Metrics {
95    fn summary(&self, ev: &Summary<'_>) {
96        bump(&self.requests, 1);
97        bump(&self.attempts, u64::from(ev.attempts));
98        bump(&self.latency[latency_bucket(ev.elapsed)], 1);
99        if let Some(class) = ev
100            .status
101            .map(|s| usize::from(s.as_u16() / 100))
102            .filter(|class| (1..=STATUS_CLASSES).contains(class))
103        {
104            bump(&self.status[class - 1], 1);
105        }
106        if let Err(error) = ev.outcome {
107            bump(&self.errors[error.category() as usize], 1);
108        }
109    }
110
111    fn body(&self, ev: &BodyEnd<'_>) {
112        bump(&self.bodies[BodyKind::of(&ev.outcome) as usize], 1);
113    }
114}
115
116impl MetricsSnapshot {
117    pub fn requests(&self) -> u64 {
118        self.requests
119    }
120
121    pub fn attempts(&self) -> u64 {
122        self.attempts
123    }
124
125    pub fn status_class(&self, class: u8) -> u64 {
126        usize::from(class)
127            .checked_sub(1)
128            .and_then(|i| self.status.get(i))
129            .copied()
130            .unwrap_or(0)
131    }
132
133    pub fn errors(&self, category: ErrorCategory) -> u64 {
134        self.errors[category as usize]
135    }
136
137    pub fn errors_total(&self) -> u64 {
138        self.errors.iter().sum()
139    }
140
141    pub fn bodies_complete(&self) -> u64 {
142        self.bodies[BodyKind::Complete as usize]
143    }
144
145    pub fn bodies_failed(&self) -> u64 {
146        self.bodies[BodyKind::Failed as usize]
147    }
148
149    pub fn bodies_dropped(&self) -> u64 {
150        self.bodies[BodyKind::Dropped as usize]
151    }
152
153    pub fn latency(&self) -> Vec<(Option<Duration>, u64)> {
154        LATENCY_BOUNDS_MS
155            .iter()
156            .map(|ms| Some(Duration::from_millis(*ms)))
157            .chain(std::iter::once(None))
158            .zip(self.latency)
159            .collect()
160    }
161}
162
163impl fmt::Display for MetricsSnapshot {
164    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
165        write!(f, "requests={} attempts={}", self.requests, self.attempts)?;
166        for (i, count) in self.status.iter().enumerate() {
167            write!(f, " status_{}xx={count}", i + 1)?;
168        }
169        for category in ErrorCategory::ALL {
170            let count = self.errors(category);
171            if count > 0 {
172                write!(f, " error_{}={count}", category.as_str())?;
173            }
174        }
175        for kind in BodyKind::ALL {
176            write!(f, " body_{}={}", kind.label(), self.bodies[kind as usize])?;
177        }
178        for (bound, count) in self.latency() {
179            match bound {
180                Some(bound) => write!(f, " latency_le_{}ms={count}", bound.as_millis())?,
181                None => write!(f, " latency_le_inf={count}")?,
182            }
183        }
184        Ok(())
185    }
186}
leyline/trace/metrics.rs

leyline/trace/
metrics.rs
