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

spate_core/metrics/
labels.rs
