taba_cli/tui/history/
mod.rs

1//! The history of the session (doc 34, D-TUI-5 and section 7): circular buffers, one per series, filled by the
2//! sampler from the cluster reads. Nothing goes to disk; closing the app forgets everything.
3
4use std::collections::{HashMap, VecDeque};
5
6use crate::tui::api::ClusterStatus;
7use crate::tui::prom;
8
9/// The longest retention accepted and the shortest sampling interval (section 7).
10pub const MAX_RETENTION_SECS: u64 = 6 * 3600;
11pub const MIN_INTERVAL_SECS: u64 = 1;
12pub const DEFAULT_RETENTION_SECS: u64 = 3600;
13pub const DEFAULT_INTERVAL_SECS: u64 = 5;
14
15/// Checks the two settings of `[history]`.
16pub fn validate(retention_secs: u64, interval_secs: u64) -> Result<(), String> {
17    if interval_secs < MIN_INTERVAL_SECS {
18        return Err("history.interval: use 1s or more".to_owned());
19    }
20    if retention_secs > MAX_RETENTION_SECS {
21        return Err("history.retention: use 6h or less".to_owned());
22    }
23    if retention_secs < interval_secs {
24        return Err("history.retention: it must hold at least one sample".to_owned());
25    }
26    Ok(())
27}
28
29/// What is measured for a node.
30#[derive(Debug, Clone, Copy, PartialEq, Eq, Hash)]
31pub enum Series {
32    MemoryUsed,
33    MemoryLimit,
34    MemoryMessages,
35    DiskMessages,
36    ActiveRequests,
37    /// Requests refused with `429` per second, from the counter.
38    RateLimited,
39    /// The worst `p95` of the channels' round trip, in seconds.
40    ChannelRtt,
41}
42
43/// One sample: when (ms since the app opened), the value, and whether the read it came from was incomplete.
44#[derive(Debug, Clone, Copy, PartialEq)]
45pub struct Point {
46    pub at: u64,
47    pub value: f64,
48    pub partial: bool,
49}
50
51#[derive(Debug, Clone, PartialEq)]
52pub struct Ring {
53    cap: usize,
54    points: VecDeque<Point>,
55}
56
57impl Ring {
58    pub fn new(cap: usize) -> Ring {
59        Ring {
60            cap: cap.max(1),
61            points: VecDeque::new(),
62        }
63    }
64
65    pub fn push(&mut self, point: Point) {
66        if self.points.len() == self.cap {
67            self.points.pop_front();
68        }
69        self.points.push_back(point);
70    }
71
72    pub fn len(&self) -> usize {
73        self.points.len()
74    }
75
76    pub fn is_empty(&self) -> bool {
77        self.points.is_empty()
78    }
79
80    pub fn last(&self) -> Option<Point> {
81        self.points.back().copied()
82    }
83
84    /// The points of the `window_ms` that ends at `now`, oldest first (a frozen chart passes the instant it froze).
85    pub fn window(&self, now: u64, window_ms: u64) -> Vec<Point> {
86        let from = now.saturating_sub(window_ms);
87        self.points
88            .iter()
89            .copied()
90            .filter(|p| p.at >= from && p.at <= now)
91            .collect()
92    }
93
94    /// How long ago the oldest sample was taken (what the chart can honestly claim to show).
95    pub fn collected_ms(&self, now: u64) -> u64 {
96        self.points.front().map_or(0, |p| now.saturating_sub(p.at))
97    }
98}
99
100/// Smallest, largest and last value of some points, for the summary line of a chart.
101pub fn summary(points: &[Point]) -> Option<(f64, f64, f64)> {
102    let last = points.last()?.value;
103    let min = points.iter().map(|p| p.value).fold(f64::INFINITY, f64::min);
104    let max = points
105        .iter()
106        .map(|p| p.value)
107        .fold(f64::NEG_INFINITY, f64::max);
108    Some((min, max, last))
109}
110
111#[derive(Debug, Clone)]
112pub struct History {
113    cap: usize,
114    interval_ms: u64,
115    rings: HashMap<(String, Series), Ring>,
116    /// The last reading of each counter, for its rate.
117    counters: HashMap<String, (u64, f64)>,
118    last_sample: Option<u64>,
119}
120
121impl History {
122    pub fn new(retention_secs: u64, interval_secs: u64) -> History {
123        let interval = interval_secs.max(MIN_INTERVAL_SECS);
124        History {
125            cap: (retention_secs / interval).max(1) as usize,
126            interval_ms: interval * 1000,
127            rings: HashMap::new(),
128            counters: HashMap::new(),
129            last_sample: None,
130        }
131    }
132
133    pub fn capacity(&self) -> usize {
134        self.cap
135    }
136
137    pub fn interval_ms(&self) -> u64 {
138        self.interval_ms
139    }
140
141    /// The retention in milliseconds: no chart window may go beyond it.
142    pub fn retention_ms(&self) -> u64 {
143        self.cap as u64 * self.interval_ms
144    }
145
146    /// Forgets everything (a new profile).
147    pub fn clear(&mut self) {
148        self.rings.clear();
149        self.counters.clear();
150        self.last_sample = None;
151    }
152
153    pub fn series(&self, node: &str, series: Series) -> Option<&Ring> {
154        self.rings.get(&(node.to_owned(), series))
155    }
156
157    fn push(&mut self, node: &str, series: Series, point: Point) {
158        let cap = self.cap;
159        self.rings
160            .entry((node.to_owned(), series))
161            .or_insert_with(|| Ring::new(cap))
162            .push(point);
163    }
164
165    /// Takes a sample of every node of a status read, if the interval has passed since the last one. The
166    /// sample of an incomplete read is kept, marked as such.
167    pub fn sample(&mut self, now: u64, status: &ClusterStatus) -> bool {
168        if self.last_sample.is_some_and(|t| now < t + self.interval_ms) {
169            return false;
170        }
171        self.last_sample = Some(now);
172        for node in status.nodes.iter().filter(|n| n.reachable) {
173            let mk = |value: f64| Point {
174                at: now,
175                value,
176                partial: status.partial,
177            };
178            let id = node.node_id.as_str();
179            self.push(id, Series::MemoryUsed, mk(node.memory_used as f64));
180            self.push(id, Series::MemoryLimit, mk(node.memory_limit as f64));
181            self.push(id, Series::MemoryMessages, mk(node.memory_messages as f64));
182            self.push(id, Series::DiskMessages, mk(node.disk_messages as f64));
183            self.push(id, Series::ActiveRequests, mk(node.active_requests));
184            if let Some(rtt) = node.rtt_p95 {
185                self.push(id, Series::ChannelRtt, mk(rtt));
186            }
187            let previous = self
188                .counters
189                .insert(id.to_owned(), (now, node.rate_limited));
190            if let Some((then, was)) = previous {
191                let seconds = (now - then) as f64 / 1000.0;
192                // A counter that went down is a restart: no sample, never a negative rate.
193                if let Some(per_second) = prom::rate(was, node.rate_limited, seconds) {
194                    self.push(id, Series::RateLimited, mk(per_second));
195                }
196            }
197        }
198        true
199    }
200}
201
202#[cfg(test)]
203mod tests;
taba_cli/tui/history/mod.rs

taba_cli/tui/history/
mod.rs
