taba_cli/tui/screens/topics/monitor/
mod.rs

1//! The charts of one queue (doc 34, TUI-MON-*, TUI-MSG-11): `Enter` on a topic opens three charts stacked on a
2//! common time axis. Each reading of `GET /stats/topics/{name}` becomes one point per line; the history is only
3//! what was sampled since the charts were opened.
4//!
5//! 1. the messages that are visible and those that are invisible (delivered and waiting for their ack);
6//! 2. the flow: published, delivered and acknowledged (deleted) per second;
7//! 3. the age of the oldest message of the queue and of its largest groups.
8
9use std::collections::HashMap;
10
11use crate::tui::api::TopicSample;
12use crate::tui::effect::Effect;
13use crate::tui::history::{Point, Ring};
14use crate::tui::i18n;
15use crate::tui::keymap::Action;
16use crate::tui::model::{Input, InputPurpose, Modal, Model, TopicMonitor, TopicTab, WINDOWS_MIN};
17use crate::tui::screens::cluster::{allowed_windows, moved, Unit};
18
19use super::{messages, refuse, selected};
20
21pub mod view;
22
23/// The most groups drawn in the chart of ages.
24pub const MAX_GROUPS: usize = 3;
25
26/// What a chart draws: a title and its lines (a key into `TopicMonitor::series` and the catalog key of the legend).
27pub struct ChartDef {
28    pub key: &'static str,
29    pub title: &'static str,
30    pub unit: Unit,
31    pub lines: &'static [(&'static str, &'static str)],
32    /// The largest groups are drawn too, one line each.
33    pub groups: bool,
34}
35
36pub const CHARTS: &[ChartDef] = &[
37    ChartDef {
38        key: "messages",
39        title: "topic-chart.messages",
40        unit: Unit::Count,
41        lines: &[
42            ("visible", "topic-chart.visible"),
43            ("invisible", "topic-chart.invisible"),
44        ],
45        groups: false,
46    },
47    ChartDef {
48        key: "flow",
49        title: "topic-chart.flow",
50        unit: Unit::PerSecond,
51        lines: &[
52            ("published", "topic-chart.published"),
53            ("consumed", "topic-chart.consumed"),
54            ("acked", "topic-chart.acked"),
55        ],
56        groups: false,
57    },
58    ChartDef {
59        key: "age",
60        title: "topic-chart.age",
61        unit: Unit::Seconds,
62        lines: &[("oldest", "topic-chart.oldest")],
63        groups: true,
64    },
65];
66
67/// A line of a chart: its series key and its legend text.
68pub struct LineDef {
69    pub key: String,
70    pub label: String,
71}
72
73/// The lines of a chart now: the fixed ones, then one per group being followed.
74pub fn lines_of(m: &Model, mon: &TopicMonitor, def: &ChartDef) -> Vec<LineDef> {
75    let tx = m.tx();
76    let mut lines: Vec<LineDef> = def
77        .lines
78        .iter()
79        .map(|(key, label)| LineDef {
80            key: (*key).to_owned(),
81            label: tx.t(label),
82        })
83        .collect();
84    if def.groups {
85        for group in &mon.groups {
86            lines.push(LineDef {
87                key: format!("g:{group}"),
88                label: tx.tw("topic-chart.group", &[("group", &group)]),
89            });
90        }
91    }
92    lines
93}
94
95/// `Enter` (the charts) or `x` (the messages) on the list: opens the selected topic. A distributor has no queue, so
96/// it has no charts: only its messages can be opened, to publish.
97pub fn open(m: &mut Model, tab: TopicTab) -> Vec<Effect> {
98    let Some(topic) = selected(m) else {
99        return refuse(m, "gate.no-topic");
100    };
101    let distributor = topic.is_distributor();
102    if distributor && tab == TopicTab::Charts {
103        return refuse(m, "gate.topic-no-charts");
104    }
105    let mut mon = TopicMonitor::new(topic.name, tab, distributor, topic.dead_letter);
106    // What the strip under the list already sampled is not thrown away.
107    if let Some(seen) = m.topics.previews.iter().find(|p| p.name == mon.name) {
108        mon.series = seen.series.clone();
109        mon.groups = seen.groups.clone();
110    }
111    m.topics.monitor = Some(mon);
112    Vec::new()
113}
114
115/// The window in milliseconds, never above what the history retains (TUI-MON-03).
116pub fn window_ms(m: &Model, mon: &TopicMonitor) -> u64 {
117    let index = mon
118        .monitor
119        .window
120        .min(allowed_windows(m.history.retention_ms()) - 1);
121    WINDOWS_MIN[index] * 60_000
122}
123
124/// The points of a line: the window, up to the instant the chart was frozen at.
125pub fn line_points(m: &Model, mon: &TopicMonitor, key: &str) -> Vec<Point> {
126    let at = mon.monitor.frozen.unwrap_or(m.now);
127    mon.series
128        .get(key)
129        .map(|ring| ring.window(at, window_ms(m, mon)))
130        .unwrap_or_default()
131}
132
133fn put(series: &mut HashMap<String, Ring>, cap: usize, key: &str, point: Point) {
134    series
135        .entry(key.to_owned())
136        .or_insert_with(|| Ring::new(cap))
137        .push(point);
138}
139
140/// Keeps a reading: one point per line. A queue with no messages has an oldest age of zero; a group that is not
141/// in the reading leaves a gap.
142pub fn record(mon: &mut TopicMonitor, cap: usize, now: u64, sample: &TopicSample) {
143    for (group, _) in &sample.groups {
144        if mon.groups.len() < MAX_GROUPS && !mon.groups.contains(group) {
145            mon.groups.push(group.clone());
146        }
147    }
148    let at = |value: f64| Point {
149        at: now,
150        value,
151        partial: sample.partial,
152    };
153    let series = &mut mon.series;
154    put(series, cap, "visible", at(sample.visible as f64));
155    put(series, cap, "invisible", at(sample.invisible as f64));
156    put(series, cap, "published", at(sample.publish_rate));
157    put(series, cap, "consumed", at(sample.consume_rate));
158    put(series, cap, "acked", at(sample.ack_rate));
159    put(series, cap, "oldest", at(sample.oldest_age.unwrap_or(0.0)));
160    for (group, age) in &sample.groups {
161        if mon.groups.contains(group) {
162            put(&mut mon.series, cap, &format!("g:{group}"), at(*age));
163        }
164    }
165}
166
167/// The keys of an open topic dialog: `Esc` closes it, the rest are those of the view being shown.
168pub fn on_action(m: &mut Model, action: Action) -> Vec<Effect> {
169    let Some(shown) = m.topics.monitor.as_ref().map(|mon| mon.tab) else {
170        return Vec::new();
171    };
172    if action == Action::Esc {
173        // On the full screen Esc goes back to the list of charts first.
174        match &mut m.topics.monitor {
175            Some(mon) if shown == TopicTab::Charts && mon.monitor.full => mon.monitor.full = false,
176            _ => m.topics.monitor = None,
177        }
178        return Vec::new();
179    }
180    if shown == TopicTab::Messages {
181        return messages::on_action(m, action);
182    }
183    charts_action(m, action)
184}
185
186/// The keys of the charts of a queue.
187fn charts_action(m: &mut Model, action: Action) -> Vec<Effect> {
188    let now = m.now;
189    let allowed = allowed_windows(m.history.retention_ms());
190    let Some(mon) = &mut m.topics.monitor else {
191        return Vec::new();
192    };
193    match action {
194        Action::Up => mon.monitor.chart = moved(mon.monitor.chart, -1, CHARTS.len()),
195        Action::Down => mon.monitor.chart = moved(mon.monitor.chart, 1, CHARTS.len()),
196        Action::First => mon.monitor.chart = 0,
197        Action::Last => mon.monitor.chart = CHARTS.len() - 1,
198        Action::ChartFull => mon.monitor.full = !mon.monitor.full,
199        Action::WindowWider => mon.monitor.window = (mon.monitor.window + 1).min(allowed - 1),
200        Action::WindowNarrower => mon.monitor.window = mon.monitor.window.saturating_sub(1),
201        Action::PauseChart => {
202            mon.monitor.frozen = match mon.monitor.frozen {
203                Some(_) => None,
204                None => Some(now),
205            }
206        }
207        Action::Scale => mon.monitor.fixed_scale = !mon.monitor.fixed_scale,
208        // One file with every line of the three charts, so the chart is not asked for.
209        Action::Export => {
210            let value = format!("taba-{}-charts.csv", crate::tui::model::file_tag(&mon.name));
211            m.modal = Some(Modal::Input(Input {
212                purpose: InputPurpose::ExportTopicCsv,
213                value,
214            }));
215        }
216        _ => {}
217    }
218    Vec::new()
219}
220
221/// The visible lines of the three charts as CSV (TUI-MON-07): the time in UTC and one column per line.
222pub fn export_csv(m: &Model) -> String {
223    let Some(mon) = &m.topics.monitor else {
224        return String::new();
225    };
226    let lines: Vec<(LineDef, Vec<Point>)> = CHARTS
227        .iter()
228        .flat_map(|def| lines_of(m, mon, def))
229        .map(|line| {
230            let points = line_points(m, mon, &line.key);
231            (line, points)
232        })
233        .collect();
234    let mut out = String::from("time");
235    for (line, _) in &lines {
236        out.push(',');
237        out.push_str(&line.label.replace(',', " "));
238    }
239    out.push('\n');
240    // Every reading is taken at one instant for all the lines, so the instants of any line are the rows.
241    let mut instants: Vec<u64> = lines
242        .iter()
243        .flat_map(|(_, points)| points.iter().map(|p| p.at))
244        .collect();
245    instants.sort_unstable();
246    instants.dedup();
247    let offset = m.wall_ms - i64::try_from(m.now).unwrap_or(0);
248    for at in instants {
249        let c = i18n::civil_from_ms(offset + i64::try_from(at).unwrap_or(0));
250        out.push_str(&format!(
251            "{:04}-{:02}-{:02}T{:02}:{:02}:{:02}Z",
252            c.year, c.month, c.day, c.hour, c.minute, c.second
253        ));
254        for (_, points) in &lines {
255            out.push(',');
256            if let Some(p) = points.iter().find(|p| p.at == at) {
257                out.push_str(&p.value.to_string());
258            }
259        }
260        out.push('\n');
261    }
262    out
263}
264
265#[cfg(test)]
266mod tests;
taba_cli/tui/screens/topics/monitor/mod.rs

taba_cli/tui/screens/topics/monitor/
mod.rs
