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.
137 ///
138 /// # Arguments
139 ///
140 /// * `event` - the event to record.
141 ///
142 /// # Returns
143 ///
144 /// `Some(event)` if it should be shipped, or `None` if it was dropped by the
145 /// threshold.
146 pub fn record(&mut self, event: Event) -> Option<Event> {
147 self.counts[event.level as usize] += 1;
148 if event.level >= self.threshold {
149 self.emitted += 1;
150 Some(event)
151 } else {
152 None
153 }
154 }
155
156 /// Returns how many events have been seen at `level`, shipped or not.
157 ///
158 /// # Arguments
159 ///
160 /// * `level` - the level to count.
161 ///
162 /// # Returns
163 ///
164 /// The number of events recorded at that level.
165 pub fn count(&self, level: Level) -> u32 {
166 self.counts[level as usize]
167 }
168
169 /// Returns the total number of events seen across all levels.
170 ///
171 /// # Returns
172 ///
173 /// The total count.
174 pub fn total(&self) -> u32 {
175 self.counts.iter().sum()
176 }
177
178 /// Returns how many events passed the threshold and were shipped.
179 ///
180 /// # Returns
181 ///
182 /// The emitted count.
183 pub fn emitted(&self) -> u32 {
184 self.emitted
185 }
186
187 /// Returns how many events were dropped by the threshold.
188 ///
189 /// # Returns
190 ///
191 /// The dropped count.
192 pub fn dropped(&self) -> u32 {
193 self.total() - self.emitted
194 }
195
196 /// Returns a snapshot of the counters to ship in place of the raw stream.
197 ///
198 /// # Returns
199 ///
200 /// A [`Snapshot`] of the per-level counts and the emitted and dropped totals.
201 pub fn snapshot(&self) -> Snapshot {
202 Snapshot {
203 by_level: self.counts,
204 emitted: self.emitted,
205 dropped: self.dropped(),
206 }
207 }
208}
209
210#[cfg(test)]
211mod tests {
212 use super::*;
213
214 #[test]
215 fn it_ships_at_or_above_the_threshold() {
216 let mut reporter = Reporter::new(Level::Info);
217 assert!(reporter.record(Event::debug("d")).is_none());
218 assert!(reporter.record(Event::info("i")).is_some());
219 assert!(reporter.record(Event::error("e")).is_some());
220 }
221
222 #[test]
223 fn it_counts_every_event_even_when_dropped() {
224 let mut reporter = Reporter::new(Level::Warn);
225 reporter.record(Event::debug("d"));
226 reporter.record(Event::debug("d"));
227 reporter.record(Event::error("e"));
228 assert_eq!(reporter.count(Level::Debug), 2);
229 assert_eq!(reporter.count(Level::Error), 1);
230 assert_eq!(reporter.total(), 3);
231 assert_eq!(reporter.emitted(), 1);
232 assert_eq!(reporter.dropped(), 2);
233 }
234
235 #[test]
236 fn link_cost_sets_the_threshold() {
237 let mut reporter = Reporter::new(Level::Trace);
238 reporter.adapt_to(LinkCost::Metered);
239 assert_eq!(reporter.threshold(), Level::Info);
240 reporter.adapt_to(LinkCost::Expensive);
241 assert_eq!(reporter.threshold(), Level::Warn);
242 reporter.adapt_to(LinkCost::Offline);
243 assert_eq!(reporter.threshold(), Level::Error);
244 reporter.adapt_to(LinkCost::Free);
245 assert_eq!(reporter.threshold(), Level::Trace);
246 }
247
248 #[test]
249 fn a_snapshot_summarizes_the_counters() {
250 let mut reporter = Reporter::new(Level::Info);
251 reporter.record(Event::trace("t"));
252 reporter.record(Event::info("i"));
253 reporter.record(Event::warn("w"));
254 let snapshot = reporter.snapshot();
255 assert_eq!(snapshot.by_level[Level::Trace as usize], 1);
256 assert_eq!(snapshot.by_level[Level::Info as usize], 1);
257 assert_eq!(snapshot.by_level[Level::Warn as usize], 1);
258 assert_eq!(snapshot.emitted, 2); // info and warn
259 assert_eq!(snapshot.dropped, 1); // trace
260 }
261}