pamoja_telemetry/
reporter.rs

1//! Recording events, filtering them for the link, and summarizing the counts.
2
3use crate::event::{Event, Level};
4
5// The number of [`Level`] variants, the width of the per-level counters.
6const LEVEL_COUNT: usize = 5;
7
8/// How costly the current link is, which sets how selective telemetry should be.
9///
10/// The cost maps to the level [`threshold`](LinkCost::threshold) a
11/// [`Reporter`] should use: a free link ships everything, while an expensive or
12/// absent link ships only what is worth its bytes.
13#[derive(Clone, Copy, Debug, PartialEq, Eq)]
14pub enum LinkCost {
15    /// A free or local link: ship all detail.
16    Free,
17    /// A metered link: skip routine detail.
18    Metered,
19    /// An expensive link, such as satellite: ship only warnings and errors.
20    Expensive,
21    /// No link: hold back everything but errors worth buffering.
22    Offline,
23}
24
25impl LinkCost {
26    /// Returns the level threshold this link cost calls for.
27    ///
28    /// # Returns
29    ///
30    /// The minimum [`Level`] a reporter should ship at this link cost.
31    pub fn threshold(self) -> Level {
32        match self {
33            LinkCost::Free => Level::Trace,
34            LinkCost::Metered => Level::Info,
35            LinkCost::Expensive => Level::Warn,
36            LinkCost::Offline => Level::Error,
37        }
38    }
39}
40
41/// A point-in-time summary of a reporter's counters.
42///
43/// This is what a node ships periodically in place of the raw event stream: a few
44/// integers that capture how many events occurred at each level and how many were
45/// shipped versus dropped, cheap to send even on a metered link.
46#[derive(Clone, Copy, Debug, PartialEq, Eq)]
47pub struct Snapshot {
48    /// The number of events seen at each level, indexed by the level's order from
49    /// [`Trace`](Level::Trace) to [`Error`](Level::Error).
50    pub by_level: [u32; LEVEL_COUNT],
51    /// How many events passed the filter and were shipped.
52    pub emitted: u32,
53    /// How many events were dropped by the filter.
54    pub dropped: u32,
55}
56
57/// Records telemetry events, ships the ones worth their bytes, and counts them all.
58///
59/// A reporter keeps a level threshold and forwards only events at or above it, while
60/// counting every event it sees - dropped or not - so the aggregate picture survives
61/// even when the link is too costly to ship detail. Call
62/// [`adapt_to`](Reporter::adapt_to) as the link cost changes to raise or lower the
63/// bar, and ship a [`snapshot`](Reporter::snapshot) of the counters periodically
64/// instead of the full stream.
65///
66/// # Examples
67///
68/// ```
69/// use pamoja_telemetry::{Event, Level, LinkCost, Reporter};
70///
71/// let mut reporter = Reporter::new(Level::Trace);
72///
73/// // On a metered link, routine debug events are dropped but a warning still ships.
74/// reporter.adapt_to(LinkCost::Metered);
75/// assert!(reporter.record(Event::debug("loop.tick")).is_none());
76/// assert!(reporter.record(Event::warn("battery.low")).is_some());
77///
78/// // Both events were still counted.
79/// assert_eq!(reporter.total(), 2);
80/// assert_eq!(reporter.dropped(), 1);
81/// ```
82pub struct Reporter {
83    threshold: Level,
84    counts: [u32; LEVEL_COUNT],
85    emitted: u32,
86}
87
88impl Reporter {
89    /// Creates a reporter that ships events at or above `threshold`.
90    ///
91    /// # Arguments
92    ///
93    /// * `threshold` - the minimum level to ship.
94    ///
95    /// # Returns
96    ///
97    /// A reporter with empty counters.
98    pub fn new(threshold: Level) -> Self {
99        Self {
100            threshold,
101            counts: [0; LEVEL_COUNT],
102            emitted: 0,
103        }
104    }
105
106    /// Returns the current ship threshold.
107    ///
108    /// # Returns
109    ///
110    /// The minimum level currently being shipped.
111    pub fn threshold(&self) -> Level {
112        self.threshold
113    }
114
115    /// Sets the ship threshold directly.
116    ///
117    /// # Arguments
118    ///
119    /// * `threshold` - the new minimum level to ship.
120    pub fn set_threshold(&mut self, threshold: Level) {
121        self.threshold = threshold;
122    }
123
124    /// Raises or lowers the threshold to match the current link cost.
125    ///
126    /// # Arguments
127    ///
128    /// * `cost` - how costly the link currently is.
129    pub fn adapt_to(&mut self, cost: LinkCost) {
130        self.threshold = cost.threshold();
131    }
132
133    /// Records an event, returning it to ship if it clears the threshold.
134    ///
135    /// The event is counted whether or not it is shipped, so the aggregate counts
136    /// stay complete even while detail is held back. A count that reaches `u32::MAX`
137    /// stays there rather than wrapping to zero, so a node that runs for years reports
138    /// a full counter, never a small one.
139    ///
140    /// # Arguments
141    ///
142    /// * `event` - the event to record.
143    ///
144    /// # Returns
145    ///
146    /// `Some(event)` if it should be shipped, or `None` if it was dropped by the
147    /// threshold.
148    pub fn record(&mut self, event: Event) -> Option<Event> {
149        let count = &mut self.counts[event.level as usize];
150        *count = count.saturating_add(1);
151        if event.level >= self.threshold {
152            self.emitted = self.emitted.saturating_add(1);
153            Some(event)
154        } else {
155            None
156        }
157    }
158
159    /// Returns how many events have been seen at `level`, shipped or not.
160    ///
161    /// # Arguments
162    ///
163    /// * `level` - the level to count.
164    ///
165    /// # Returns
166    ///
167    /// The number of events recorded at that level.
168    pub fn count(&self, level: Level) -> u32 {
169        self.counts[level as usize]
170    }
171
172    /// Returns the total number of events seen across all levels.
173    ///
174    /// # Returns
175    ///
176    /// The total count, held at `u32::MAX` once it gets there.
177    pub fn total(&self) -> u32 {
178        self.counts
179            .iter()
180            .fold(0u32, |total, count| total.saturating_add(*count))
181    }
182
183    /// Returns how many events passed the threshold and were shipped.
184    ///
185    /// # Returns
186    ///
187    /// The emitted count.
188    pub fn emitted(&self) -> u32 {
189        self.emitted
190    }
191
192    /// Returns how many events were dropped by the threshold.
193    ///
194    /// # Returns
195    ///
196    /// The dropped count.
197    pub fn dropped(&self) -> u32 {
198        self.total().saturating_sub(self.emitted)
199    }
200
201    /// Returns a snapshot of the counters to ship in place of the raw stream.
202    ///
203    /// # Returns
204    ///
205    /// A [`Snapshot`] of the per-level counts and the emitted and dropped totals.
206    pub fn snapshot(&self) -> Snapshot {
207        Snapshot {
208            by_level: self.counts,
209            emitted: self.emitted,
210            dropped: self.dropped(),
211        }
212    }
213}
214
215#[cfg(test)]
216mod tests {
217    use super::*;
218
219    #[test]
220    fn it_ships_at_or_above_the_threshold() {
221        let mut reporter = Reporter::new(Level::Info);
222        assert!(reporter.record(Event::debug("d")).is_none());
223        assert!(reporter.record(Event::info("i")).is_some());
224        assert!(reporter.record(Event::error("e")).is_some());
225    }
226
227    #[test]
228    fn it_counts_every_event_even_when_dropped() {
229        let mut reporter = Reporter::new(Level::Warn);
230        reporter.record(Event::debug("d"));
231        reporter.record(Event::debug("d"));
232        reporter.record(Event::error("e"));
233        assert_eq!(reporter.count(Level::Debug), 2);
234        assert_eq!(reporter.count(Level::Error), 1);
235        assert_eq!(reporter.total(), 3);
236        assert_eq!(reporter.emitted(), 1);
237        assert_eq!(reporter.dropped(), 2);
238    }
239
240    #[test]
241    fn link_cost_sets_the_threshold() {
242        let mut reporter = Reporter::new(Level::Trace);
243        reporter.adapt_to(LinkCost::Metered);
244        assert_eq!(reporter.threshold(), Level::Info);
245        reporter.adapt_to(LinkCost::Expensive);
246        assert_eq!(reporter.threshold(), Level::Warn);
247        reporter.adapt_to(LinkCost::Offline);
248        assert_eq!(reporter.threshold(), Level::Error);
249        reporter.adapt_to(LinkCost::Free);
250        assert_eq!(reporter.threshold(), Level::Trace);
251    }
252
253    #[test]
254    fn a_snapshot_summarizes_the_counters() {
255        let mut reporter = Reporter::new(Level::Info);
256        reporter.record(Event::trace("t"));
257        reporter.record(Event::info("i"));
258        reporter.record(Event::warn("w"));
259        let snapshot = reporter.snapshot();
260        assert_eq!(snapshot.by_level[Level::Trace as usize], 1);
261        assert_eq!(snapshot.by_level[Level::Info as usize], 1);
262        assert_eq!(snapshot.by_level[Level::Warn as usize], 1);
263        assert_eq!(snapshot.emitted, 2); // info and warn
264        assert_eq!(snapshot.dropped, 1); // trace
265    }
266
267    #[test]
268    fn a_full_counter_holds_rather_than_wrapping() {
269        let mut reporter = Reporter::new(Level::Info);
270        reporter.counts = [0, u32::MAX - 1, u32::MAX - 1, 0, 0];
271        reporter.emitted = u32::MAX - 1;
272        reporter.record(Event::debug("d"));
273        reporter.record(Event::debug("d"));
274        reporter.record(Event::info("i"));
275        reporter.record(Event::info("i"));
276        assert_eq!(reporter.count(Level::Debug), u32::MAX);
277        assert_eq!(reporter.count(Level::Info), u32::MAX);
278        assert_eq!(reporter.emitted(), u32::MAX);
279        assert_eq!(reporter.total(), u32::MAX);
280        assert_eq!(reporter.dropped(), 0);
281    }
282}
pamoja_telemetry/reporter.rs

pamoja_telemetry/
reporter.rs
