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 come first (so every family joins cleanly against the
166    /// framework's 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; [`Meter`](super::Meter)
171    /// needs a runtime-sized slice and a name assembled from its namespace at
172    /// 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)); the check is a plain bool test, not a lock.
227///
228/// The handle is wrapped rather than each setter gated. A stage struct's
229/// initial publishes run in its constructor, and those are the writes that
230/// clobber a live owner.
231#[derive(Clone, Debug)]
232pub(crate) struct OwnedGauge {
233    gauge: Gauge,
234    owned: bool,
235}
236
237impl OwnedGauge {
238    pub(crate) fn new(gauge: Gauge, owned: bool) -> Self {
239        OwnedGauge { gauge, owned }
240    }
241
242    #[inline]
243    pub(crate) fn set(&self, value: f64) {
244        if self.owned {
245            self.gauge.set(value);
246        }
247    }
248
249    #[inline]
250    pub(crate) fn increment(&self, value: f64) {
251        if self.owned {
252            self.gauge.increment(value);
253        }
254    }
255}
256
257/// Metadata attached to connector- and user-owned metric families. The
258/// framework's own stage metrics register through the `metrics` macros, which
259/// stamp `module_path!()` here; families registered through
260/// [`Meter`](super::Meter) inherit this module's path. The exporters this
261/// framework installs do not surface metadata.
262const FAMILY_METADATA: Metadata<'static> =
263    Metadata::new(module_path!(), Level::INFO, Some(module_path!()));
264
265/// Guard the caller's `extra` labels at build time (cold path). None may
266/// shadow a standard label key (those are attached automatically) or repeat
267/// another `extra` key. Panics on a violation, at startup before any data
268/// flows. The name's namespace is validated up front by
269/// [`Meter`](super::Meter), so it needs no check here.
270fn validate_extra_labels(name: &str, extra: &[(&'static str, SharedString)]) {
271    for (i, (k, _)) in extra.iter().enumerate() {
272        assert!(
273            *k != names::L_PIPELINE && *k != names::L_COMPONENT && *k != names::L_COMPONENT_TYPE,
274            "custom label `{k}` on `{name}` shadows a standard label \
275             (pipeline/component/component_type are attached automatically)"
276        );
277        assert!(
278            !extra[..i].iter().any(|(prev, _)| prev == k),
279            "custom label `{k}` is repeated on `{name}`"
280        );
281    }
282}
283
284/// Why a `Meter` namespace token was rejected. The panicking constructor path
285/// ([`validate_namespace`]) and the non-panicking runtime path
286/// (`Meter::for_component`) share one rule set and phrase the outcome
287/// differently. An explicit author call gets a hard error; a component default
288/// gets a silent opt-out or a warning.
289#[derive(Clone, Copy, Debug, PartialEq, Eq)]
290pub(crate) enum NamespaceRejection {
291    /// Empty string.
292    Empty,
293    /// Not a lowercase `[a-z][a-z0-9_]*` segment, and so not a legal
294    /// metric-name segment. For example it contains an uppercase letter or a
295    /// hyphen, or leads with a digit.
296    Malformed,
297    /// A framework stage root (`source`, `sink`, …); a custom family here would
298    /// collide with the taxonomy.
299    Reserved,
300}
301
302impl NamespaceRejection {
303    /// A short reason phrase for a diagnostic (`… because {reason}`).
304    pub(crate) fn reason(self) -> &'static str {
305        match self {
306            NamespaceRejection::Empty => "it is empty",
307            NamespaceRejection::Malformed => "it is not a lowercase `[a-z][a-z0-9_]*` segment",
308            NamespaceRejection::Reserved => "it is a reserved framework stage root",
309        }
310    }
311}
312
313/// Classify a `Meter` namespace token (the `<ns>` in the `spate_<ns>_` prefix
314/// every one of its metrics gets). Returns `Ok(())` if it is a usable,
315/// non-reserved segment, else why it was rejected. Both the panicking
316/// [`validate_namespace`] and the non-panicking `Meter::for_component` resolve
317/// through this function; rejecting a reserved root keeps custom names from
318/// colliding with the framework taxonomy.
319pub(crate) fn classify_namespace(namespace: &str) -> Result<(), NamespaceRejection> {
320    if namespace.is_empty() {
321        return Err(NamespaceRejection::Empty);
322    }
323    let well_formed = namespace
324        .bytes()
325        .all(|b| b.is_ascii_lowercase() || b.is_ascii_digit() || b == b'_')
326        && namespace.as_bytes()[0].is_ascii_lowercase();
327    if !well_formed {
328        return Err(NamespaceRejection::Malformed);
329    }
330    if names::RESERVED_ROOTS.contains(&namespace) {
331        return Err(NamespaceRejection::Reserved);
332    }
333    Ok(())
334}
335
336/// Validate a `Meter` namespace token, panicking with a specific message on
337/// rejection. The construction-time wiring check behind
338/// [`Meter::with_namespace`](super::Meter::with_namespace).
339pub(crate) fn validate_namespace(namespace: &str) {
340    match classify_namespace(namespace) {
341        Ok(()) => {}
342        Err(NamespaceRejection::Empty) => panic!(
343            "Meter namespace must not be empty (it becomes the `spate_<namespace>_` \
344             segment on every metric); use `\"custom\"` or your connector's name"
345        ),
346        Err(NamespaceRejection::Malformed) => panic!(
347            "Meter namespace `{namespace}` must be a lowercase `[a-z][a-z0-9_]*` \
348             segment (it becomes part of the `spate_<namespace>_` metric prefix)"
349        ),
350        Err(NamespaceRejection::Reserved) => panic!(
351            "Meter namespace `{namespace}` is a reserved framework root; custom \
352             families would collide with `spate_{namespace}_*`. Use `\"custom\"` or \
353             a connector segment like `\"kafka\"`."
354        ),
355    }
356}
357
358impl ErrorClass {
359    pub(crate) fn label(self) -> &'static str {
360        match self {
361            ErrorClass::Retryable => "retryable",
362            ErrorClass::RecordLevel => "record_level",
363            ErrorClass::Fatal => "fatal",
364        }
365    }
366}
367
368/// Dynamic per-partition gauge family, gated by `per_partition_detail`.
369/// Registration happens on the control plane (rebalance/commit paths); the hot
370/// path never touches this.
371#[derive(Debug)]
372pub(crate) struct PartitionGauges {
373    pub(crate) name: &'static str,
374    pub(crate) labels: ComponentLabels,
375    pub(crate) gauges: Mutex<HashMap<u32, Gauge>>,
376    /// Whether the owning handle set owns this series (see [`OwnedGauge`]).
377    /// A shadow registers nothing here and publishes nothing.
378    pub(crate) owned: bool,
379}
380
381impl PartitionGauges {
382    pub(crate) fn set(&self, partition: PartitionId, value: f64) {
383        if !self.owned {
384            return;
385        }
386        let mut gauges = self.gauges.lock().expect("partition gauge lock");
387        gauges
388            .entry(partition.0)
389            .or_insert_with(|| {
390                self.labels
391                    .gauge1(self.name, names::L_PARTITION, partition.0.to_string())
392            })
393            .set(value);
394    }
395
396    /// Zeroes and then drops the handles for partitions this component no
397    /// longer owns.
398    ///
399    /// The `metrics` facade has no deletion and no idle timeout is configured
400    /// (see `configured_builder`), so dropping a handle is invisible to the
401    /// exporter. The series keeps rendering its last value for the life of the
402    /// process. Without the zeroing, a reader that aggregates across members
403    /// counts a partition twice, once frozen here and once live on the member
404    /// that now owns it.
405    ///
406    /// Absence and `0` therefore mean different things for a per-partition
407    /// series. Absent is "never measured"; `0` here is "measured, not ours".
408    pub(crate) fn retain(&self, keep: &[PartitionId]) {
409        let mut gauges = self.gauges.lock().expect("partition gauge lock");
410        gauges.retain(|p, gauge| {
411            let kept = keep.iter().any(|k| k.0 == *p);
412            if !kept {
413                gauge.set(0.0);
414            }
415            kept
416        });
417    }
418}
spate_core/metrics/labels.rs

spate_core/metrics/
labels.rs
