spate_core/metrics/
labels.rs

1//! Shared building blocks for the per-stage handle structs: the standard
2//! label set, its typed `Counter`/`Gauge`/`Histogram` constructors, and the
3//! dynamic per-partition gauge family.
4//!
5//! The stage constructors (`counter`, `counter1`, ...) are `pub(crate)` so
6//! each stage module (`source`, `sink`, `checkpoint`, ...) resolves its
7//! handles through one code path that always attaches the three standard
8//! labels. The dynamic-arity `register_*` path backing the public
9//! [`Meter`](super::Meter) lives here too (also `pub(crate)`), so
10//! connector- and user-owned metric families inherit the same labels.
11
12use super::names;
13use crate::error::ErrorClass;
14use crate::record::PartitionId;
15use metrics::{
16    Counter, Gauge, Histogram, Key, Label, Level, Metadata, SharedString, counter, gauge,
17    histogram, with_recorder,
18};
19use std::collections::HashMap;
20use std::sync::Mutex;
21
22/// The standard label set attached to every framework metric.
23#[derive(Clone, Debug)]
24pub struct ComponentLabels {
25    /// Pipeline name.
26    pub pipeline: SharedString,
27    /// Component instance id from config/builder (e.g. `orders_kafka`).
28    pub component: SharedString,
29    /// Component implementation (e.g. `kafka`, `clickhouse`, `map`).
30    pub component_type: SharedString,
31}
32
33impl ComponentLabels {
34    /// Build the standard label set.
35    pub fn new(
36        pipeline: impl Into<SharedString>,
37        component: impl Into<SharedString>,
38        component_type: impl Into<SharedString>,
39    ) -> Self {
40        ComponentLabels {
41            pipeline: pipeline.into(),
42            component: component.into(),
43            component_type: component_type.into(),
44        }
45    }
46
47    pub(crate) fn counter(&self, name: &'static str) -> Counter {
48        counter!(name,
49            names::L_PIPELINE => self.pipeline.clone(),
50            names::L_COMPONENT => self.component.clone(),
51            names::L_COMPONENT_TYPE => self.component_type.clone(),
52        )
53    }
54
55    pub(crate) fn counter1(
56        &self,
57        name: &'static str,
58        k: &'static str,
59        v: impl Into<SharedString>,
60    ) -> Counter {
61        counter!(name,
62            names::L_PIPELINE => self.pipeline.clone(),
63            names::L_COMPONENT => self.component.clone(),
64            names::L_COMPONENT_TYPE => self.component_type.clone(),
65            k => v.into(),
66        )
67    }
68
69    pub(crate) fn counter2(
70        &self,
71        name: &'static str,
72        k1: &'static str,
73        v1: impl Into<SharedString>,
74        k2: &'static str,
75        v2: impl Into<SharedString>,
76    ) -> Counter {
77        counter!(name,
78            names::L_PIPELINE => self.pipeline.clone(),
79            names::L_COMPONENT => self.component.clone(),
80            names::L_COMPONENT_TYPE => self.component_type.clone(),
81            k1 => v1.into(),
82            k2 => v2.into(),
83        )
84    }
85
86    pub(crate) fn gauge(&self, name: &'static str) -> Gauge {
87        gauge!(name,
88            names::L_PIPELINE => self.pipeline.clone(),
89            names::L_COMPONENT => self.component.clone(),
90            names::L_COMPONENT_TYPE => self.component_type.clone(),
91        )
92    }
93
94    pub(crate) fn gauge1(
95        &self,
96        name: &'static str,
97        k: &'static str,
98        v: impl Into<SharedString>,
99    ) -> Gauge {
100        gauge!(name,
101            names::L_PIPELINE => self.pipeline.clone(),
102            names::L_COMPONENT => self.component.clone(),
103            names::L_COMPONENT_TYPE => self.component_type.clone(),
104            k => v.into(),
105        )
106    }
107
108    pub(crate) fn gauge2(
109        &self,
110        name: &'static str,
111        k1: &'static str,
112        v1: impl Into<SharedString>,
113        k2: &'static str,
114        v2: impl Into<SharedString>,
115    ) -> Gauge {
116        gauge!(name,
117            names::L_PIPELINE => self.pipeline.clone(),
118            names::L_COMPONENT => self.component.clone(),
119            names::L_COMPONENT_TYPE => self.component_type.clone(),
120            k1 => v1.into(),
121            k2 => v2.into(),
122        )
123    }
124
125    pub(crate) fn histogram(&self, name: &'static str) -> Histogram {
126        histogram!(name,
127            names::L_PIPELINE => self.pipeline.clone(),
128            names::L_COMPONENT => self.component.clone(),
129            names::L_COMPONENT_TYPE => self.component_type.clone(),
130        )
131    }
132
133    pub(crate) fn histogram1(
134        &self,
135        name: &'static str,
136        k: &'static str,
137        v: impl Into<SharedString>,
138    ) -> Histogram {
139        histogram!(name,
140            names::L_PIPELINE => self.pipeline.clone(),
141            names::L_COMPONENT => self.component.clone(),
142            names::L_COMPONENT_TYPE => self.component_type.clone(),
143            k => v.into(),
144        )
145    }
146
147    pub(crate) fn histogram2(
148        &self,
149        name: &'static str,
150        k1: &'static str,
151        v1: impl Into<SharedString>,
152        k2: &'static str,
153        v2: impl Into<SharedString>,
154    ) -> Histogram {
155        histogram!(name,
156            names::L_PIPELINE => self.pipeline.clone(),
157            names::L_COMPONENT => self.component.clone(),
158            names::L_COMPONENT_TYPE => self.component_type.clone(),
159            k1 => v1.into(),
160            k2 => v2.into(),
161        )
162    }
163
164    /// Build the metric key for a dynamic-arity family: the three standard
165    /// labels first (so every family joins cleanly against the framework's
166    /// series), then the caller's `extra` labels in order.
167    ///
168    /// This mirrors what the `counter!`/`gauge!`/`histogram!` macros lower to
169    /// for runtime label values (`Key::from_parts` over a `Vec<Label>`); the
170    /// stage constructors above use the macro form because their labels are
171    /// fixed, while [`Meter`](super::Meter) needs a runtime-sized slice and a
172    /// name assembled from its namespace at build time.
173    fn family_key(&self, name: SharedString, extra: &[(&'static str, SharedString)]) -> Key {
174        validate_extra_labels(&name, extra);
175        let mut labels = Vec::with_capacity(3 + extra.len());
176        labels.push(Label::new(names::L_PIPELINE, self.pipeline.clone()));
177        labels.push(Label::new(names::L_COMPONENT, self.component.clone()));
178        labels.push(Label::new(
179            names::L_COMPONENT_TYPE,
180            self.component_type.clone(),
181        ));
182        for (k, v) in extra {
183            labels.push(Label::new(*k, v.clone()));
184        }
185        Key::from_parts(name, labels)
186    }
187
188    /// Resolve a counter carrying the three standard labels plus `extra`.
189    /// `name` is the fully-qualified `spate_<namespace>_...` name the
190    /// [`Meter`](super::Meter) assembled. Build-time (cold path) only.
191    pub(crate) fn register_counter(
192        &self,
193        name: SharedString,
194        extra: &[(&'static str, SharedString)],
195    ) -> Counter {
196        let key = self.family_key(name, extra);
197        with_recorder(|recorder| recorder.register_counter(&key, &FAMILY_METADATA))
198    }
199
200    /// Resolve a gauge carrying the three standard labels plus `extra`.
201    pub(crate) fn register_gauge(
202        &self,
203        name: SharedString,
204        extra: &[(&'static str, SharedString)],
205    ) -> Gauge {
206        let key = self.family_key(name, extra);
207        with_recorder(|recorder| recorder.register_gauge(&key, &FAMILY_METADATA))
208    }
209
210    /// Resolve a histogram carrying the three standard labels plus `extra`.
211    pub(crate) fn register_histogram(
212        &self,
213        name: SharedString,
214        extra: &[(&'static str, SharedString)],
215    ) -> Histogram {
216        let key = self.family_key(name, extra);
217        with_recorder(|recorder| recorder.register_histogram(&key, &FAMILY_METADATA))
218    }
219}
220
221/// A gauge that publishes only for the handle set that **owns** its series.
222///
223/// Registration still happens for a shadow — the key is the same, so it is the
224/// same series and nothing extra renders — but every write is dropped, leaving
225/// the owner's reading intact. Ownership is decided once, at construction (see
226/// [`ownership`](super::ownership)); this is a plain bool test, not a lock.
227///
228/// Wrapping the handle rather than gating each setter is deliberate: a stage
229/// struct's initial publishes run in its constructor, and those are exactly
230/// the writes that clobber a live owner. Routing them through this type makes
231/// the suppression impossible to forget.
232#[derive(Clone, Debug)]
233pub(crate) struct OwnedGauge {
234    gauge: Gauge,
235    owned: bool,
236}
237
238impl OwnedGauge {
239    pub(crate) fn new(gauge: Gauge, owned: bool) -> Self {
240        OwnedGauge { gauge, owned }
241    }
242
243    #[inline]
244    pub(crate) fn set(&self, value: f64) {
245        if self.owned {
246            self.gauge.set(value);
247        }
248    }
249
250    #[inline]
251    pub(crate) fn increment(&self, value: f64) {
252        if self.owned {
253            self.gauge.increment(value);
254        }
255    }
256}
257
258/// Metadata attached to connector- and user-owned metric families. The
259/// framework's own stage metrics register through the `metrics` macros, which
260/// stamp `module_path!()` here; families registered through
261/// [`Meter`](super::Meter) inherit this module's path, which is adequate — the
262/// exporters this framework installs do not surface metadata.
263const FAMILY_METADATA: Metadata<'static> =
264    Metadata::new(module_path!(), Level::INFO, Some(module_path!()));
265
266/// Guard the caller's `extra` labels at build time (cold path): none may
267/// shadow a standard label key (they are attached automatically) or repeat
268/// another `extra` key. Panics — a construction-time wiring mistake, caught
269/// at startup before any data flows, like a bad sink topology. The name's
270/// namespace is validated up front by [`Meter`](super::Meter), so it needs no
271/// check here.
272fn validate_extra_labels(name: &str, extra: &[(&'static str, SharedString)]) {
273    for (i, (k, _)) in extra.iter().enumerate() {
274        assert!(
275            *k != names::L_PIPELINE && *k != names::L_COMPONENT && *k != names::L_COMPONENT_TYPE,
276            "custom label `{k}` on `{name}` shadows a standard label \
277             (pipeline/component/component_type are attached automatically)"
278        );
279        assert!(
280            !extra[..i].iter().any(|(prev, _)| prev == k),
281            "custom label `{k}` is repeated on `{name}`"
282        );
283    }
284}
285
286/// Why a `Meter` namespace token was rejected. Lets the panicking constructor
287/// path ([`validate_namespace`]) and the non-panicking runtime path
288/// (`Meter::for_component`) share one rule set while phrasing the outcome
289/// differently — a hard error for an explicit author call, a silent opt-out or
290/// a warning for a component default.
291#[derive(Clone, Copy, Debug, PartialEq, Eq)]
292pub(crate) enum NamespaceRejection {
293    /// Empty string.
294    Empty,
295    /// Not a lowercase `[a-z][a-z0-9_]*` segment (so not a legal metric-name
296    /// segment) — e.g. it contains an uppercase letter or a hyphen, or leads
297    /// with a digit.
298    Malformed,
299    /// A framework stage root (`source`, `sink`, …); a custom family here would
300    /// collide with the taxonomy.
301    Reserved,
302}
303
304impl NamespaceRejection {
305    /// A short reason phrase for a diagnostic (`… because {reason}`).
306    pub(crate) fn reason(self) -> &'static str {
307        match self {
308            NamespaceRejection::Empty => "it is empty",
309            NamespaceRejection::Malformed => "it is not a lowercase `[a-z][a-z0-9_]*` segment",
310            NamespaceRejection::Reserved => "it is a reserved framework stage root",
311        }
312    }
313}
314
315/// Classify a `Meter` namespace token (the `<ns>` in the `spate_<ns>_` prefix
316/// every one of its metrics gets): `Ok(())` if it is a usable, non-reserved
317/// segment, else why it was rejected. The single source of truth behind both
318/// the panicking [`validate_namespace`] and the non-panicking
319/// `Meter::for_component`; the reserved case is what makes custom names
320/// collision-proof against the framework taxonomy.
321pub(crate) fn classify_namespace(namespace: &str) -> Result<(), NamespaceRejection> {
322    if namespace.is_empty() {
323        return Err(NamespaceRejection::Empty);
324    }
325    let well_formed = namespace
326        .bytes()
327        .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'_')
328        && namespace.as_bytes()[0].is_ascii_lowercase();
329    if !well_formed {
330        return Err(NamespaceRejection::Malformed);
331    }
332    if names::RESERVED_ROOTS.contains(&namespace) {
333        return Err(NamespaceRejection::Reserved);
334    }
335    Ok(())
336}
337
338/// Validate a `Meter` namespace token, panicking with a specific message on
339/// rejection. The construction-time wiring check behind
340/// [`Meter::with_namespace`](super::Meter::with_namespace).
341pub(crate) fn validate_namespace(namespace: &str) {
342    match classify_namespace(namespace) {
343        Ok(()) => {}
344        Err(NamespaceRejection::Empty) => panic!(
345            "Meter namespace must not be empty (it becomes the `spate_<namespace>_` \
346             segment on every metric); use `\"custom\"` or your connector's name"
347        ),
348        Err(NamespaceRejection::Malformed) => panic!(
349            "Meter namespace `{namespace}` must be a lowercase `[a-z][a-z0-9_]*` \
350             segment (it becomes part of the `spate_<namespace>_` metric prefix)"
351        ),
352        Err(NamespaceRejection::Reserved) => panic!(
353            "Meter namespace `{namespace}` is a reserved framework root; custom \
354             families would collide with `spate_{namespace}_*`. Use `\"custom\"` or \
355             a connector segment like `\"kafka\"`."
356        ),
357    }
358}
359
360impl ErrorClass {
361    pub(crate) fn label(self) -> &'static str {
362        match self {
363            ErrorClass::Retryable => "retryable",
364            ErrorClass::RecordLevel => "record_level",
365            ErrorClass::Fatal => "fatal",
366        }
367    }
368}
369
370/// Dynamic per-partition gauge family, gated by `per_partition_detail`.
371/// Registration happens on the control plane (rebalance/commit paths), so a
372/// mutex is acceptable; the hot path never touches this.
373#[derive(Debug)]
374pub(crate) struct PartitionGauges {
375    pub(crate) name: &'static str,
376    pub(crate) labels: ComponentLabels,
377    pub(crate) gauges: Mutex<HashMap<u32, Gauge>>,
378    /// Whether the owning handle set owns this series — see [`OwnedGauge`].
379    /// A shadow registers nothing here and publishes nothing: the owner is
380    /// the one member whose per-partition figures are real.
381    pub(crate) owned: bool,
382}
383
384impl PartitionGauges {
385    pub(crate) fn set(&self, partition: PartitionId, value: f64) {
386        if !self.owned {
387            return;
388        }
389        let mut gauges = self.gauges.lock().expect("partition gauge lock");
390        gauges
391            .entry(partition.0)
392            .or_insert_with(|| {
393                self.labels
394                    .gauge1(self.name, names::L_PARTITION, partition.0.to_string())
395            })
396            .set(value);
397    }
398
399    /// Zeroes and then drops the handles for partitions this component no
400    /// longer owns.
401    ///
402    /// The zeroing is the load-bearing half. The `metrics` facade has no
403    /// deletion and no idle timeout is configured (see `configured_builder`),
404    /// so dropping a handle is invisible to the exporter: the series keeps
405    /// rendering its last value for the life of the process. A reader that
406    /// aggregates across members would then count a partition twice — once
407    /// frozen here, once live on the member that now owns it. Setting `0`
408    /// first makes this component's contribution honest, because `0` is the
409    /// truth: it owns none of that partition.
410    ///
411    /// This is why absence and `0` mean different things for a per-partition
412    /// series. Absent is "never measured"; `0` here is "measured, not ours".
413    pub(crate) fn retain(&self, keep: &[PartitionId]) {
414        let mut gauges = self.gauges.lock().expect("partition gauge lock");
415        gauges.retain(|p, gauge| {
416            let kept = keep.iter().any(|k| k.0 == *p);
417            if !kept {
418                gauge.set(0.0);
419            }
420            kept
421        });
422    }
423}
spate_core/metrics/labels.rs

spate_core/metrics/
labels.rs
