spectra_core/sinks/
recording.rs

1//! In-memory [`SpectraSink`] for tests.
2
3use std::sync::{Arc, Mutex};
4
5use serde_json::Value;
6
7use crate::sink::SpectraSink;
8
9/// Captured counter increment.
10#[derive(Debug, Clone, PartialEq, Eq)]
11pub struct RecordedCounter {
12    pub name: String,
13    pub labels: Vec<(String, String)>,
14    pub delta: i64,
15}
16
17/// Captured gauge sample.
18#[derive(Debug, Clone, PartialEq)]
19pub struct RecordedGauge {
20    pub name: String,
21    pub labels: Vec<(String, String)>,
22    pub value: f64,
23}
24
25/// Captured structured event row.
26#[derive(Debug, Clone, PartialEq, Eq)]
27pub struct RecordedEvent {
28    pub table: String,
29    pub fields: Value,
30}
31
32#[derive(Debug, Default)]
33struct Inner {
34    counters: Vec<RecordedCounter>,
35    gauges: Vec<RecordedGauge>,
36    events: Vec<RecordedEvent>,
37}
38
39/// Append-only in-memory sink for assertions in unit and integration tests.
40///
41/// Clones share the same captured data. Use the typed accessors to inspect all emits or the
42/// matching helpers to select one metric name, label subset, or event table.
43///
44/// # Examples
45///
46/// ```
47/// use spectra_core::{RecordingSink, SpectraSink};
48///
49/// let sink = RecordingSink::new();
50/// sink.record_counter("cache_hits", &[("region", "us")], 2);
51///
52/// let hits = sink.recorded_counters_matching("cache_hits", &[("region", "us")]);
53/// assert_eq!(hits.len(), 1);
54/// assert_eq!(hits[0].delta, 2);
55///
56/// sink.clear();
57/// assert!(sink.counters().is_empty());
58/// ```
59#[derive(Debug, Clone)]
60pub struct RecordingSink {
61    inner: Arc<Mutex<Inner>>,
62}
63
64impl Default for RecordingSink {
65    fn default() -> Self {
66        Self::new()
67    }
68}
69
70impl RecordingSink {
71    /// Creates an empty recording sink.
72    pub fn new() -> Self {
73        Self {
74            inner: Arc::new(Mutex::new(Inner::default())),
75        }
76    }
77
78    /// Clears all recorded emits.
79    pub fn clear(&self) {
80        let mut g = self.inner.lock().expect("recording sink lock");
81        g.counters.clear();
82        g.gauges.clear();
83        g.events.clear();
84    }
85
86    /// Returns all recorded counter increments.
87    pub fn counters(&self) -> Vec<RecordedCounter> {
88        self.inner
89            .lock()
90            .expect("recording sink lock")
91            .counters
92            .clone()
93    }
94
95    /// Returns all recorded gauge samples.
96    pub fn gauges(&self) -> Vec<RecordedGauge> {
97        self.inner
98            .lock()
99            .expect("recording sink lock")
100            .gauges
101            .clone()
102    }
103
104    /// Returns all recorded events.
105    pub fn events(&self) -> Vec<RecordedEvent> {
106        self.inner
107            .lock()
108            .expect("recording sink lock")
109            .events
110            .clone()
111    }
112
113    /// Returns counters matching a name and label subset.
114    pub fn recorded_counters_matching(
115        &self,
116        name: &str,
117        label_subset: &[(&str, &str)],
118    ) -> Vec<RecordedCounter> {
119        self.counters()
120            .into_iter()
121            .filter(|c| c.name == name && labels_contain(&c.labels, label_subset))
122            .collect()
123    }
124
125    /// Returns events recorded for a specific table.
126    pub fn recorded_events_for(&self, table: &str) -> Vec<RecordedEvent> {
127        self.events()
128            .into_iter()
129            .filter(|e| e.table == table)
130            .collect()
131    }
132}
133
134fn labels_contain(labels: &[(String, String)], subset: &[(&str, &str)]) -> bool {
135    subset.iter().all(|(k, v)| {
136        labels
137            .iter()
138            .any(|(lk, lv)| lk.as_str() == *k && lv.as_str() == *v)
139    })
140}
141
142impl SpectraSink for RecordingSink {
143    fn record_counter(&self, name: &str, labels: &[(&str, &str)], delta: i64) {
144        let labels: Vec<(String, String)> = labels
145            .iter()
146            .map(|(k, v)| (k.to_string(), v.to_string()))
147            .collect();
148        self.inner
149            .lock()
150            .expect("recording sink lock")
151            .counters
152            .push(RecordedCounter {
153                name: name.to_string(),
154                labels,
155                delta,
156            });
157    }
158
159    fn record_gauge(&self, name: &str, labels: &[(&str, &str)], value: f64) {
160        let labels: Vec<(String, String)> = labels
161            .iter()
162            .map(|(k, v)| (k.to_string(), v.to_string()))
163            .collect();
164        self.inner
165            .lock()
166            .expect("recording sink lock")
167            .gauges
168            .push(RecordedGauge {
169                name: name.to_string(),
170                labels,
171                value,
172            });
173    }
174
175    fn log_event(&self, table: &str, fields: &Value) {
176        self.inner
177            .lock()
178            .expect("recording sink lock")
179            .events
180            .push(RecordedEvent {
181                table: table.to_string(),
182                fields: fields.clone(),
183            });
184    }
185}
186
187#[cfg(test)]
188mod tests {
189    use super::*;
190    use serde_json::json;
191
192    #[test]
193    fn captures_counters_and_events() {
194        let sink = RecordingSink::new();
195        sink.record_counter("example_db_reads", &[("table", "t"), ("op", "get")], 1);
196        sink.log_event("example_error_log", &json!({"source": "database"}));
197
198        assert_eq!(sink.counters().len(), 1);
199        assert_eq!(sink.events().len(), 1);
200        let hits = sink.recorded_counters_matching("example_db_reads", &[("op", "get")]);
201        assert_eq!(hits.len(), 1);
202    }
203}
spectra_core/sinks/recording.rs

spectra_core/sinks/
recording.rs
