iota_metrics/
peak.rs

1// Copyright (c) 2026 IOTA Stiftung
2// SPDX-License-Identifier: Apache-2.0
3
4//! Gauges that keep the exact maximum over a sliding window.
5//!
6//! [`PeakGauge`] keeps the exact maximum over a window of the same length as
7//! that of the quantile gauges, for values that move faster than the scrape
8//! interval and would otherwise only ever be sampled at one arbitrary instant.
9//! [`PeakGaugeVec`] keeps one such peak for each value of a label.
10//! [`IntGaugeWithPeakGauge`] pairs an `IntGauge` with such a peak.
11
12use std::{
13    collections::HashMap,
14    sync::{
15        Arc,
16        atomic::{AtomicU64, Ordering},
17    },
18    time::Instant,
19};
20
21use parking_lot::{Mutex, MutexGuard, RwLock};
22use prometheus_filtered::{
23    MetricLevel, Opts, Registry,
24    core::{Collector, Desc},
25    prometheus::{IntGauge, IntGaugeVec, proto::MetricFamily},
26};
27
28use crate::quantile_gauge::{WINDOW_SLOT, WINDOW_SLOTS};
29
30/// The highest value over a sliding window, kept exactly per slot. The window
31/// has the same length as that of the quantile gauges. Observing takes no lock while the
32/// slot does not change; a change of slot and a read take a lock.
33struct PeakWindow {
34    start: Instant,
35    /// The number of the slot being filled: the time since `start` in slots.
36    current_slot: AtomicU64,
37    /// The maxima of the last `WINDOW_SLOTS` slots, slot `n` at `n %
38    /// WINDOW_SLOTS`.
39    slots: [AtomicU64; WINDOW_SLOTS],
40    rotation: Mutex<()>,
41}
42
43impl PeakWindow {
44    fn new(now: Instant) -> Self {
45        Self {
46            start: now,
47            current_slot: AtomicU64::new(0),
48            slots: std::array::from_fn(|_| AtomicU64::new(0)),
49            rotation: Mutex::new(()),
50        }
51    }
52
53    fn slot_number(&self, now: Instant) -> u64 {
54        (now.saturating_duration_since(self.start).as_nanos() / WINDOW_SLOT.as_nanos()) as u64
55    }
56
57    /// Records `value` at `now`. `now` is not later than the current time. A
58    /// value older than the window is dropped.
59    fn observe_at(&self, value: u64, now: Instant) {
60        let value_slot = self.slot_number(now);
61        let current_slot = self.current_slot.load(Ordering::Acquire);
62        if value_slot == current_slot {
63            let slot_max = self.slot_max(value_slot);
64            if value > slot_max.load(Ordering::Relaxed) {
65                slot_max.fetch_max(value, Ordering::Relaxed);
66            }
67            return;
68        }
69        let rotation = self.rotation.lock();
70        self.advance_to(&rotation, value_slot);
71        let current_slot = self.current_slot.load(Ordering::Relaxed);
72        let still_in_window = value_slot + WINDOW_SLOTS as u64 > current_slot;
73        if still_in_window {
74            self.slot_max(value_slot)
75                .fetch_max(value, Ordering::Relaxed);
76        }
77    }
78
79    fn slot_max(&self, slot: u64) -> &AtomicU64 {
80        &self.slots[slot as usize % WINDOW_SLOTS]
81    }
82
83    /// Moves to `target_slot` and clears the slots it takes over.
84    fn advance_to(&self, _rotation: &MutexGuard<'_, ()>, target_slot: u64) {
85        let current_slot = self.current_slot.load(Ordering::Relaxed);
86        if target_slot <= current_slot {
87            return;
88        }
89        let first_to_clear =
90            (current_slot + 1).max(target_slot.saturating_sub(WINDOW_SLOTS as u64 - 1));
91        for slot in first_to_clear..=target_slot {
92            self.slot_max(slot).store(0, Ordering::Relaxed);
93        }
94        self.current_slot.store(target_slot, Ordering::Release);
95    }
96
97    fn max_at(&self, now: Instant) -> u64 {
98        {
99            let rotation = self.rotation.lock();
100            self.advance_to(&rotation, self.slot_number(now));
101        }
102        self.slots
103            .iter()
104            .map(|slot| slot.load(Ordering::Relaxed))
105            .max()
106            .unwrap_or(0)
107    }
108
109    fn observe(&self, value: u64) {
110        self.observe_at(value, Instant::now());
111    }
112
113    fn max(&self) -> u64 {
114        self.max_at(Instant::now())
115    }
116}
117
118/// The highest value observed over the window, exposed as `<name>`. The value
119/// is exact. It is a scalar from [`PeakGauge::register`], or the gauge of one
120/// label value from [`PeakGaugeVec::with_label_values`].
121///
122/// Register with [`PeakGauge::register`], feed it with [`observe`], and the
123/// registry's scrape reports the peak over the current window. A window with
124/// no observation reports zero rather than dropping the series, so an idle
125/// period is visible as such.
126///
127/// For an `IntGauge` that also exports its peak, see [`IntGaugeWithPeakGauge`].
128///
129/// [`observe`]: PeakGauge::observe
130#[derive(Clone)]
131pub struct PeakGauge {
132    gauge: IntGauge,
133    window: Arc<PeakWindow>,
134    /// If set, the peak is never reported below the current value of this gauge.
135    floor: Option<IntGauge>,
136}
137
138impl PeakGauge {
139    /// # Panics
140    ///
141    /// Panics if a metric of this name is already registered.
142    pub fn register(
143        name: &str,
144        help: &str,
145        module: &str,
146        registry: &Registry,
147        level: MetricLevel,
148    ) -> Self {
149        Self::register_with_floor(name, help, module, registry, level, None)
150    }
151
152    /// Like [`register`](Self::register), and with a `floor` the peak is never
153    /// reported below the current value of `floor`.
154    fn register_with_floor(
155        name: &str,
156        help: &str,
157        module: &str,
158        registry: &Registry,
159        level: MetricLevel,
160        floor: Option<IntGauge>,
161    ) -> Self {
162        let gauge = IntGauge::with_opts(Opts::new(name, help)).expect("valid gauge options");
163        let this = Self {
164            gauge,
165            window: Arc::new(PeakWindow::new(Instant::now())),
166            floor,
167        };
168        registry
169            .register_filtered(name, module, level, this)
170            .expect("peak gauge registers without collision")
171    }
172
173    pub fn observe(&self, value: u64) {
174        self.window.observe(value);
175    }
176}
177
178impl Collector for PeakGauge {
179    fn desc(&self) -> Vec<&Desc> {
180        self.gauge.desc()
181    }
182
183    fn collect(&self) -> Vec<MetricFamily> {
184        let current = self
185            .floor
186            .as_ref()
187            .map_or(0, |gauge| gauge.get().max(0) as u64);
188        self.gauge
189            .set(i64::try_from(self.window.max().max(current)).unwrap_or(i64::MAX));
190        self.gauge.collect()
191    }
192}
193
194/// The highest value observed over the window per value of one label, exposed
195/// as `<name>{<label>="..."}`. A series stays after its window empties and then
196/// reports zero, so the label values must be a small fixed set.
197#[derive(Clone)]
198pub struct PeakGaugeVec {
199    gauge: IntGaugeVec,
200    windows: Arc<RwLock<HashMap<String, Arc<PeakWindow>>>>,
201}
202
203impl PeakGaugeVec {
204    /// # Panics
205    ///
206    /// Panics if a metric of this name is already registered.
207    pub fn register(
208        name: &str,
209        help: &str,
210        label: &str,
211        module: &str,
212        registry: &Registry,
213        level: MetricLevel,
214    ) -> Self {
215        let gauge = IntGaugeVec::new(Opts::new(name, help), &[label]).expect("valid gauge options");
216        let this = Self {
217            gauge,
218            windows: Arc::new(RwLock::new(HashMap::new())),
219        };
220        registry
221            .register_filtered(name, module, level, this)
222            .expect("peak gauge registers without collision")
223    }
224
225    /// The peak of `label_values`, created if needed. Its window is shared by
226    /// all peaks of the same label values.
227    ///
228    /// # Panics
229    ///
230    /// Panics if `label_values` does not hold exactly one value.
231    pub fn with_label_values(&self, label_values: &[&str]) -> PeakGauge {
232        let [label_value] = label_values else {
233            panic!("a peak gauge vector has one label");
234        };
235        PeakGauge {
236            gauge: self.gauge.with_label_values(label_values),
237            window: self.window(label_value),
238            floor: None,
239        }
240    }
241
242    fn window(&self, label_value: &str) -> Arc<PeakWindow> {
243        if let Some(window) = self.windows.read().get(label_value) {
244            return window.clone();
245        }
246        self.windows
247            .write()
248            .entry(label_value.to_owned())
249            .or_insert_with(|| Arc::new(PeakWindow::new(Instant::now())))
250            .clone()
251    }
252}
253
254impl Collector for PeakGaugeVec {
255    fn desc(&self) -> Vec<&Desc> {
256        self.gauge.desc()
257    }
258
259    fn collect(&self) -> Vec<MetricFamily> {
260        let windows: Vec<_> = self
261            .windows
262            .read()
263            .iter()
264            .map(|(label, window)| (label.clone(), window.clone()))
265            .collect();
266        for (label_value, window) in windows {
267            self.gauge
268                .with_label_values(&[label_value.as_str()])
269                .set(i64::try_from(window.max()).unwrap_or(i64::MAX));
270        }
271        self.gauge.collect()
272    }
273}
274
275/// An `IntGauge` and its peak over the window, as a [`PeakGauge`]. The peak is
276/// observed at each increment, and a scrape reports at least the current value
277/// of the gauge.
278/// A decrement that runs between an increment and the observation that
279/// follows it can make the peak miss the value the increment gave.
280#[derive(Clone)]
281pub struct IntGaugeWithPeakGauge {
282    gauge: IntGauge,
283    peak: PeakGauge,
284}
285
286impl IntGaugeWithPeakGauge {
287    /// Registers the gauge and its peak, each as `(name, help)`.
288    ///
289    /// # Panics
290    ///
291    /// Panics if a metric of one of the names is already registered.
292    pub fn register(
293        (name, help): (&str, &str),
294        (peak_name, peak_help): (&str, &str),
295        module: &str,
296        registry: &Registry,
297        level: MetricLevel,
298    ) -> Self {
299        let gauge = IntGauge::with_opts(Opts::new(name, help)).expect("valid gauge options");
300        let gauge = registry
301            .register_filtered(name, module, level, gauge)
302            .expect("gauge registers without collision");
303        let peak = PeakGauge::register_with_floor(
304            peak_name,
305            peak_help,
306            module,
307            registry,
308            level,
309            Some(gauge.clone()),
310        );
311        Self { gauge, peak }
312    }
313
314    /// Adds one and observes the new value as a peak.
315    pub fn inc(&self) {
316        self.gauge.inc();
317        self.peak.observe(self.gauge.get().max(0) as u64);
318    }
319
320    /// Subtracts one. The peak does not change.
321    pub fn dec(&self) {
322        self.gauge.dec();
323    }
324}
325
326#[cfg(test)]
327mod tests {
328    use std::time::Duration;
329
330    use super::*;
331    use crate::test_utils::MetricsReader;
332
333    #[test]
334    fn peak_is_zero_before_any_observation() {
335        let window = PeakWindow::new(Instant::now());
336        assert_eq!(window.max_at(Instant::now()), 0);
337    }
338
339    #[test]
340    fn observing_zero_leaves_the_peak_at_zero() {
341        let t0 = Instant::now();
342        let window = PeakWindow::new(t0);
343        window.observe_at(0, t0);
344        assert_eq!(window.max_at(t0), 0);
345    }
346
347    #[test]
348    fn peak_is_exact() {
349        let t0 = Instant::now();
350        for value in [32768, 1500, 1_000_003, u64::MAX / 2] {
351            let window = PeakWindow::new(t0);
352            window.observe_at(value, t0);
353            assert_eq!(window.max_at(t0), value);
354        }
355    }
356
357    #[test]
358    fn peak_is_the_highest_value_across_slots() {
359        let t0 = Instant::now();
360        let window = PeakWindow::new(t0);
361        window.observe_at(3, t0);
362        window.observe_at(70, t0 + WINDOW_SLOT);
363        window.observe_at(5, t0 + WINDOW_SLOT * 2);
364        assert_eq!(window.max_at(t0 + WINDOW_SLOT * 2), 70);
365    }
366
367    #[test]
368    fn peak_drops_values_older_than_the_window() {
369        let t0 = Instant::now();
370        let window = PeakWindow::new(t0);
371        window.observe_at(90, t0);
372        // The slot of the value is the oldest one still in the window.
373        assert_eq!(
374            window.max_at(t0 + WINDOW_SLOT * (WINDOW_SLOTS as u32 - 1)),
375            90
376        );
377        assert_eq!(window.max_at(t0 + WINDOW_SLOT * WINDOW_SLOTS as u32), 0);
378        window.observe_at(2, t0 + WINDOW_SLOT * WINDOW_SLOTS as u32);
379        assert_eq!(window.max_at(t0 + WINDOW_SLOT * WINDOW_SLOTS as u32), 2);
380    }
381
382    #[test]
383    fn peak_forgets_everything_after_a_gap_longer_than_the_window() {
384        let t0 = Instant::now();
385        let window = PeakWindow::new(t0);
386        for i in 0..WINDOW_SLOTS as u32 {
387            window.observe_at(100 + i as u64, t0 + WINDOW_SLOT * i);
388        }
389        assert_eq!(
390            window.max_at(t0 + WINDOW_SLOT * (WINDOW_SLOTS as u32 - 1)),
391            111
392        );
393        assert_eq!(window.max_at(t0 + WINDOW_SLOT * 100), 0);
394    }
395
396    #[test]
397    fn peak_accepts_a_late_value_for_a_slot_still_in_the_window() {
398        let t0 = Instant::now();
399        let window = PeakWindow::new(t0);
400        window.observe_at(1, t0 + WINDOW_SLOT * 3);
401        window.observe_at(9, t0 + WINDOW_SLOT);
402        assert_eq!(window.max_at(t0 + WINDOW_SLOT * 3), 9);
403    }
404
405    #[test]
406    fn concurrent_observations_end_with_the_highest_value() {
407        let window = Arc::new(PeakWindow::new(Instant::now()));
408        let threads: Vec<_> = (0..8u64)
409            .map(|t| {
410                let window = window.clone();
411                std::thread::spawn(move || {
412                    for i in 0..10_000u64 {
413                        window.observe(t * 10_000 + i);
414                    }
415                })
416            })
417            .collect();
418        for thread in threads {
419            thread.join().unwrap();
420        }
421        assert_eq!(window.max(), 79_999);
422    }
423
424    #[test]
425    fn observe_at_uses_the_given_time() {
426        let t0 = Instant::now();
427        let window = PeakWindow::new(t0);
428        window.observe_at(9, t0);
429        let later = t0 + WINDOW_SLOT * (WINDOW_SLOTS as u32 + 5);
430        window.observe_at(4, later);
431        assert_eq!(window.max_at(later), 4, "the earlier value left the window");
432    }
433
434    #[test]
435    fn a_late_value_at_the_edge_of_the_window_is_dropped() {
436        let t0 = Instant::now();
437        let window = PeakWindow::new(t0);
438        let edge = t0 + WINDOW_SLOT * WINDOW_SLOTS as u32;
439        window.observe_at(1, edge);
440        window.observe_at(9, t0);
441        assert_eq!(window.max_at(edge), 1);
442    }
443
444    #[test]
445    fn observing_in_the_current_slot_takes_no_lock() {
446        let t0 = Instant::now();
447        let window = Arc::new(PeakWindow::new(t0));
448        let _held = window.rotation.lock();
449        let (sender, receiver) = std::sync::mpsc::channel();
450        let observer = window.clone();
451        std::thread::spawn(move || {
452            observer.observe_at(1, t0);
453            sender.send(()).unwrap();
454        });
455        receiver
456            .recv_timeout(Duration::from_secs(5))
457            .expect("the observation waited for the rotation lock");
458    }
459
460    #[test]
461    fn peak_gauge_reports_the_exact_peak() {
462        let registry = Registry::new();
463        let gauge = PeakGauge::register(
464            "test_peak_gauge",
465            "help",
466            module_path!(),
467            &registry,
468            MetricLevel::Warn,
469        );
470        gauge.observe(32768);
471        gauge.observe(1500);
472        let families = registry.gather();
473        let family = families
474            .iter()
475            .find(|f| f.name() == "test_peak_gauge")
476            .unwrap();
477        assert_eq!(family.get_metric()[0].get_gauge().value(), 32768.0);
478    }
479
480    #[test]
481    fn int_gauge_with_peak_gauge_reports_the_peak_and_the_gauge() {
482        let registry = Registry::new();
483        let gauge = IntGaugeWithPeakGauge::register(
484            ("test_gauge", "help"),
485            ("test_gauge_peak", "help"),
486            module_path!(),
487            &registry,
488            MetricLevel::Warn,
489        );
490        let reader = MetricsReader::new(&registry);
491        for _ in 0..3 {
492            gauge.inc();
493        }
494        gauge.dec();
495        gauge.dec();
496        assert_eq!(reader.value("test_gauge", &[]), 1.0);
497        assert_eq!(reader.value("test_gauge_peak", &[]), 3.0);
498    }
499
500    #[test]
501    fn the_peak_of_a_gauge_is_never_below_the_gauge() {
502        let registry = Registry::new();
503        let gauge = IntGaugeWithPeakGauge::register(
504            ("test_gauge", "help"),
505            ("test_gauge_peak", "help"),
506            module_path!(),
507            &registry,
508            MetricLevel::Warn,
509        );
510        // The window of the peak holds no observation of this value.
511        gauge.gauge.set(5);
512        assert_eq!(
513            MetricsReader::new(&registry).value("test_gauge_peak", &[]),
514            5.0
515        );
516    }
517
518    #[test]
519    fn peak_with_a_floor_reports_the_peak_and_never_less_than_the_gauge() {
520        let registry = Registry::new();
521        let gauge = IntGauge::new("test_gauge", "help").unwrap();
522        let peak = PeakGauge::register_with_floor(
523            "test_gauge_peak",
524            "help",
525            module_path!(),
526            &registry,
527            MetricLevel::Warn,
528            Some(gauge.clone()),
529        );
530        let reader = MetricsReader::new(&registry);
531        for _ in 0..3 {
532            gauge.inc();
533            peak.observe(gauge.get() as u64);
534        }
535        gauge.dec();
536        gauge.dec();
537        assert_eq!(reader.value("test_gauge_peak", &[]), 3.0);
538
539        peak.window
540            .observe_at(0, Instant::now() + WINDOW_SLOT * (WINDOW_SLOTS as u32 + 1));
541        assert_eq!(
542            reader.value("test_gauge_peak", &[]),
543            1.0,
544            "the window is empty, so the peak is the current gauge"
545        );
546    }
547
548    #[test]
549    fn peak_gauge_vec_reports_the_peak_per_label_value() {
550        let registry = Registry::new();
551        let gauge = PeakGaugeVec::register(
552            "test_gauge_vec_peak",
553            "help",
554            "kind",
555            module_path!(),
556            &registry,
557            MetricLevel::Warn,
558        );
559        let (a, b) = (
560            gauge.with_label_values(&["a"]),
561            gauge.with_label_values(&["b"]),
562        );
563        a.observe(2);
564        a.observe(1);
565        b.observe(7);
566        let reader = MetricsReader::new(&registry);
567        assert_eq!(reader.value("test_gauge_vec_peak", &[("kind", "a")]), 2.0);
568        assert_eq!(reader.value("test_gauge_vec_peak", &[("kind", "b")]), 7.0);
569        assert_eq!(
570            gauge.with_label_values(&["a"]).window.max(),
571            2,
572            "a second peak of the label value shares the window"
573        );
574    }
575
576    #[test]
577    fn threads_that_get_the_same_peak_share_one_window() {
578        let registry = Registry::new();
579        let gauge = PeakGaugeVec::register(
580            "test_gauge_vec_threads",
581            "help",
582            "kind",
583            module_path!(),
584            &registry,
585            MetricLevel::Warn,
586        );
587        for round in 0..200 {
588            let label = round.to_string();
589            let barrier = std::sync::Barrier::new(8);
590            std::thread::scope(|scope| {
591                for value in 1..=8 {
592                    let (gauge, barrier, label) = (&gauge, &barrier, label.as_str());
593                    scope.spawn(move || {
594                        barrier.wait();
595                        gauge.with_label_values(&[label]).observe(value);
596                    });
597                }
598            });
599            assert_eq!(
600                gauge.with_label_values(&[&label]).window.max(),
601                8,
602                "round {round}"
603            );
604        }
605    }
606}
iota_metrics/peak.rs

iota_metrics/
peak.rs
