saluki_core/observability/metrics/
processor.rs

1//! Helper for processing [`AggregatedMetricsState`] into Prometheus text exposition format.
2//!
3//! [`TelemetryProcessor`] wraps a [`PrometheusRenderer`] and an optional set of [`RemapperRule`]s.
4//! It has two modes:
5//!
6//! - **No rules**: every metric in the state is rendered under its raw name and tags.
7//! - **With rules**: each metric is matched against the rules; only matched metrics are rendered,
8//!   under their remapped name and tags. Counters, gauges, and histograms are all supported in
9//!   both modes.
10
11use std::collections::BTreeMap;
12
13use prometheus_exposition::{MetricType, PrometheusRenderer};
14use stringtheory::MetaString;
15
16use super::aggregated::{AggregatedMetricValue, AggregatedMetricsState};
17use super::remapper::RemapperRule;
18
19/// Processes an [`AggregatedMetricsState`] into a Prometheus text exposition payload.
20///
21/// Owns the underlying [`PrometheusRenderer`] (whose internal buffers are reused across calls) and
22/// an optional set of [`RemapperRule`]s. See the module-level docs for the two rendering modes.
23pub struct TelemetryProcessor {
24    renderer: PrometheusRenderer,
25    rules: Vec<RemapperRule>,
26}
27
28impl TelemetryProcessor {
29    /// Creates a new `TelemetryProcessor` with no remapper rules.
30    ///
31    /// In this mode, every metric in the state is rendered under its raw name and tags.
32    pub fn new() -> Self {
33        Self {
34            renderer: PrometheusRenderer::new(),
35            rules: Vec::new(),
36        }
37    }
38
39    /// Configures the set of [`RemapperRule`]s applied during rendering.
40    ///
41    /// In this mode, only metrics that match at least one rule are rendered; rule order matters
42    /// (see [`RemapperRule::with_continued_matching`] for how a single source metric can fan out
43    /// to multiple remapped metrics).
44    pub fn with_remapper_rules(mut self, rules: Vec<RemapperRule>) -> Self {
45        self.rules = rules;
46        self
47    }
48
49    /// Processes the given state and returns the resulting Prometheus exposition payload.
50    pub fn process(&mut self, state: &AggregatedMetricsState) -> String {
51        self.renderer.clear();
52
53        if self.rules.is_empty() {
54            self.process_all(state);
55        } else {
56            self.process_remapped(state);
57        }
58
59        self.renderer.output().to_string()
60    }
61
62    fn process_all(&mut self, state: &AggregatedMetricsState) {
63        // Group metrics first by name, then by sorted tag set. Duplicate (name, tags) entries are
64        // merged so counters sum, gauges take the latest, and histograms merge buckets/sum/count.
65        let mut groups: BTreeMap<MetaString, BTreeMap<Vec<MetaString>, AggregatedMetricValue>> = BTreeMap::new();
66
67        state.visit_metrics(|context, value| {
68            let mut tags: Vec<MetaString> = context.tags().into_iter().map(|tag| tag.clone().into_inner()).collect();
69            tags.sort();
70
71            let series = groups.entry(context.name().clone()).or_default();
72            series
73                .entry(tags)
74                .and_modify(|existing| existing.merge(value))
75                .or_insert_with(|| value.clone());
76        });
77
78        for (name, series) in &groups {
79            render_group(&mut self.renderer, name, None, series);
80        }
81    }
82
83    fn process_remapped(&mut self, state: &AggregatedMetricsState) {
84        // Collect, deduplicate, and group matched metrics by their remapped name and tags.
85        //
86        // Multiple source metrics can remap to the same (name, tags) identity. We merge those via
87        // `AggregatedMetricValue::merge`, and tags are sorted to normalize their order so identical
88        // tag sets in different orders deduplicate correctly.
89        let mut groups: BTreeMap<&'static str, BTreeMap<Vec<MetaString>, AggregatedMetricValue>> = BTreeMap::new();
90        let mut help_text: BTreeMap<&'static str, &'static str> = BTreeMap::new();
91
92        state.visit_metrics(|context, value| {
93            for rule in &self.rules {
94                if let Some(mut remapped) = rule.try_match_no_context(context) {
95                    remapped.tags.sort();
96
97                    let continue_matching = rule.should_continue_matching();
98                    if let Some(text) = rule.help_text() {
99                        help_text.entry(remapped.name).or_insert(text);
100                    }
101                    let series = groups.entry(remapped.name).or_default();
102                    series
103                        .entry(remapped.tags)
104                        .and_modify(|existing| existing.merge(value))
105                        .or_insert_with(|| value.clone());
106
107                    if !continue_matching {
108                        return;
109                    }
110                }
111            }
112        });
113
114        for (name, series) in &groups {
115            render_group(&mut self.renderer, name, help_text.get(name).copied(), series);
116        }
117    }
118}
119
120impl Default for TelemetryProcessor {
121    fn default() -> Self {
122        Self::new()
123    }
124}
125
126fn render_group(
127    renderer: &mut PrometheusRenderer, name: &str, help_text: Option<&str>,
128    series: &BTreeMap<Vec<MetaString>, AggregatedMetricValue>,
129) {
130    // Determine the metric type from the first series value.
131    let metric_type = match series.values().next() {
132        Some(AggregatedMetricValue::Counter(_)) => MetricType::Counter,
133        Some(AggregatedMetricValue::Gauge(_)) => MetricType::Gauge,
134        Some(AggregatedMetricValue::Histogram(_)) => MetricType::Histogram,
135        None => return,
136    };
137
138    match metric_type {
139        MetricType::Counter | MetricType::Gauge => {
140            let rendered = series.iter().map(|(tags, value)| (split_tags(tags), value.value()));
141            renderer.render_scalar_group(name, metric_type, help_text, rendered);
142        }
143        MetricType::Histogram => {
144            renderer.begin_group(name, metric_type, help_text);
145            for (tags, value) in series {
146                if let AggregatedMetricValue::Histogram(histogram) = value {
147                    renderer.write_histogram_series(
148                        split_tags(tags),
149                        histogram.buckets(),
150                        histogram.sum(),
151                        histogram.count(),
152                    );
153                }
154            }
155            renderer.finish_group();
156        }
157        MetricType::Summary => {}
158    }
159}
160
161/// Splits a sorted `key:value` tag list into Prometheus label pairs, dropping bare tags.
162fn split_tags(tags: &[MetaString]) -> impl Iterator<Item = (&str, &str)> {
163    tags.iter().filter_map(|tag| tag.as_ref().split_once(':'))
164}
165
166#[cfg(test)]
167mod tests {
168    use saluki_context::Context;
169
170    use super::super::aggregated::AggregatedMetricsProcessor;
171    use super::super::reflector::Processor as _;
172    use super::super::MetricsSnapshot;
173    use super::*;
174    use crate::data_model::event::{metric::Metric, Event};
175
176    fn process_all(metrics: Vec<Event>) -> AggregatedMetricsState {
177        let processor = AggregatedMetricsProcessor;
178        let state = processor.build_initial_state();
179        processor.process(
180            MetricsSnapshot {
181                upserts: metrics,
182                evictions: Vec::new(),
183            },
184            &state,
185        );
186        state
187    }
188
189    #[test]
190    fn renders_counter_and_gauge_groups_without_rules() {
191        let state = process_all(vec![
192            Event::Metric(Metric::counter(
193                Context::from_static_parts("adp.requests_total", &["method:get"]),
194                10.0,
195            )),
196            Event::Metric(Metric::counter(
197                Context::from_static_parts("adp.requests_total", &["method:post"]),
198                3.0,
199            )),
200            Event::Metric(Metric::gauge(
201                Context::from_static_parts("adp.queue_depth", &["queue:work"]),
202                5.0,
203            )),
204        ]);
205
206        let output = TelemetryProcessor::new().process(&state);
207
208        assert!(output.contains("# TYPE adp__requests_total counter"));
209        assert!(output.contains("adp__requests_total{method=\"get\"} 10"));
210        assert!(output.contains("adp__requests_total{method=\"post\"} 3"));
211        assert!(output.contains("# TYPE adp__queue_depth gauge"));
212        assert!(output.contains("adp__queue_depth{queue=\"work\"} 5"));
213    }
214
215    #[test]
216    fn renders_histogram_groups_without_rules() {
217        let state = process_all(vec![
218            Event::Metric(Metric::histogram(
219                Context::from_static_parts("adp.latency_seconds", &["op:read"]),
220                [0.001, 0.002, 0.5],
221            )),
222            Event::Metric(Metric::histogram(
223                Context::from_static_parts("adp.latency_seconds", &["op:write"]),
224                [0.01, 0.02],
225            )),
226        ]);
227
228        let output = TelemetryProcessor::new().process(&state);
229
230        assert!(output.contains("# TYPE adp__latency_seconds histogram"));
231        assert!(output.contains("adp__latency_seconds_bucket{op=\"read\","));
232        assert!(output.contains("adp__latency_seconds_bucket{op=\"write\","));
233        assert!(output.contains("le=\"+Inf\""));
234        assert!(output.contains("adp__latency_seconds_sum{op=\"read\"}"));
235        assert!(output.contains("adp__latency_seconds_count{op=\"read\"} 3"));
236        assert!(output.contains("adp__latency_seconds_count{op=\"write\"} 2"));
237    }
238
239    #[test]
240    fn renders_only_matched_metrics_with_rules() {
241        let state = process_all(vec![
242            Event::Metric(Metric::counter(
243                Context::from_static_parts("src.matched", &["component_id:x"]),
244                42.0,
245            )),
246            Event::Metric(Metric::counter(
247                Context::from_static_parts("src.unmatched", &["component_id:x"]),
248                100.0,
249            )),
250        ]);
251
252        let rules =
253            vec![RemapperRule::by_name("src.matched", "dst.renamed").with_help_text("Renamed counter help text")];
254
255        let output = TelemetryProcessor::new().with_remapper_rules(rules).process(&state);
256
257        assert!(output.contains("# HELP dst__renamed Renamed counter help text"));
258        assert!(output.contains("# TYPE dst__renamed counter"));
259        assert!(output.contains("dst__renamed 42"));
260        // The unmatched counter must not appear.
261        assert!(!output.contains("unmatched"));
262    }
263
264    #[test]
265    fn renders_histograms_through_rules() {
266        let state = process_all(vec![Event::Metric(Metric::histogram(
267            Context::from_static_parts("src.latency_seconds", &["op:read"]),
268            [0.001, 0.5],
269        ))]);
270
271        let rules = vec![RemapperRule::by_name("src.latency_seconds", "dst.latency_seconds")
272            .with_original_tags(["op"])
273            .with_help_text("Remapped latency")];
274
275        let output = TelemetryProcessor::new().with_remapper_rules(rules).process(&state);
276
277        assert!(output.contains("# HELP dst__latency_seconds Remapped latency"));
278        assert!(output.contains("# TYPE dst__latency_seconds histogram"));
279        assert!(output.contains("dst__latency_seconds_bucket{op=\"read\","));
280        assert!(output.contains("dst__latency_seconds_count{op=\"read\"} 2"));
281    }
282}
saluki_core/observability/metrics/processor.rs

saluki_core/observability/metrics/
processor.rs
