aster_forge_metrics/
instruments.rs

1//! Exporter-independent typed instruments. Handles carry their instance ownership.
2
3use crate::{MetricDescriptor, MetricKind, MetricsLimits, provider::metric_name};
4use hashbrown::{Equivalent, HashMap};
5use opentelemetry::{
6    KeyValue,
7    metrics::{BoundCounter, BoundGauge, BoundHistogram, Counter, Gauge, Histogram, Meter},
8};
9use parking_lot::Mutex;
10use std::{
11    hash::{Hash, Hasher},
12    sync::{
13        Arc,
14        atomic::{AtomicBool, AtomicU64, AtomicUsize, Ordering},
15    },
16};
17
18/// Stable, deliberately redacted errors: never contain labels or exporter credentials.
19#[derive(Clone, Copy, Debug, PartialEq, Eq, thiserror::Error)]
20pub enum MetricError {
21    #[error("invalid metrics configuration")]
22    InvalidConfiguration,
23    #[error("invalid metric descriptor")]
24    InvalidDescriptor,
25    #[error("duplicate metric registration")]
26    DuplicateRegistration,
27    #[error("metric instrument limit reached")]
28    InstrumentLimit,
29    #[error("metric series limit reached")]
30    SeriesLimit,
31    #[error("invalid metric labels")]
32    InvalidLabels,
33    #[error("wrong metric kind")]
34    WrongKind,
35    #[error("invalid metric value or counter overflow")]
36    InvalidValue,
37    #[error("metrics initialization failed")]
38    Initialization,
39    #[error("metrics exporter is unavailable")]
40    Unavailable,
41    #[error("metrics operation is busy")]
42    Busy,
43    #[error("metrics operation timed out")]
44    Timeout,
45    #[error("metrics export failed")]
46    Export,
47    #[error("metrics provider is closed")]
48    Closed,
49    #[error("metrics shutdown failed")]
50    Shutdown,
51}
52
53pub type MetricResult<T> = std::result::Result<T, MetricError>;
54
55/// Borrow the caller's labels for cache lookup; allocate only for new series.
56struct BorrowedLabels<'a>(&'a [&'a str]);
57impl Hash for BorrowedLabels<'_> {
58    fn hash<H: Hasher>(&self, state: &mut H) {
59        self.0.hash(state);
60    }
61}
62impl Equivalent<Vec<String>> for BorrowedLabels<'_> {
63    fn equivalent(&self, key: &Vec<String>) -> bool {
64        self.0.len() == key.len() && self.0.iter().zip(key).all(|(left, right)| *left == right)
65    }
66}
67
68enum Instrument {
69    Counter(Counter<u64>),
70    Gauge(Gauge<f64>),
71    Histogram(Histogram<f64>),
72}
73enum BoundInstrument {
74    Counter(BoundCounter<u64>),
75    Gauge(BoundGauge<f64>),
76    Histogram(BoundHistogram<f64>),
77}
78struct Series {
79    instrument: BoundInstrument,
80    counter: AtomicU64,
81    gauge: Mutex<f64>,
82}
83struct HandleInner {
84    descriptor: MetricDescriptor,
85    instrument: Instrument,
86    enabled: bool,
87    closed: Arc<AtomicBool>,
88    limits: MetricsLimits,
89    series_count: Arc<AtomicUsize>,
90    series: Mutex<HashMap<Vec<String>, Arc<Series>>>,
91}
92
93#[derive(Clone)]
94pub struct MetricHandle {
95    inner: Arc<HandleInner>,
96}
97
98impl MetricHandle {
99    pub(crate) fn new(
100        descriptor: MetricDescriptor,
101        meter: &Meter,
102        enabled: bool,
103        closed: Arc<AtomicBool>,
104        limits: MetricsLimits,
105        series_count: Arc<AtomicUsize>,
106    ) -> Self {
107        let name = metric_name(&descriptor);
108        let instrument = match descriptor.kind {
109            MetricKind::Counter => Instrument::Counter(
110                meter
111                    .u64_counter(name)
112                    .with_description(descriptor.help)
113                    .with_unit(descriptor.unit)
114                    .build(),
115            ),
116            MetricKind::Gauge => Instrument::Gauge(
117                meter
118                    .f64_gauge(name)
119                    .with_description(descriptor.help)
120                    .with_unit(descriptor.unit)
121                    .build(),
122            ),
123            MetricKind::Histogram => {
124                let builder = meter
125                    .f64_histogram(name)
126                    .with_description(descriptor.help)
127                    .with_unit(descriptor.unit);
128                Instrument::Histogram(if descriptor.buckets.is_empty() {
129                    builder.build()
130                } else {
131                    builder.with_boundaries(descriptor.buckets.to_vec()).build()
132                })
133            }
134        };
135        Self {
136            inner: Arc::new(HandleInner {
137                descriptor,
138                instrument,
139                enabled,
140                closed,
141                limits,
142                series_count,
143                series: Mutex::new(HashMap::new()),
144            }),
145        }
146    }
147
148    #[must_use]
149    pub fn kind(&self) -> MetricKind {
150        self.inner.descriptor.kind
151    }
152
153    fn active(&self) -> MetricResult<bool> {
154        if self.inner.closed.load(Ordering::Acquire) {
155            Err(MetricError::Closed)
156        } else {
157            Ok(self.inner.enabled)
158        }
159    }
160
161    fn series(&self, labels: &[&str]) -> MetricResult<Arc<Series>> {
162        if labels.len() != self.inner.descriptor.labels.len()
163            || labels
164                .iter()
165                .any(|label| label.len() > self.inner.limits.max_label_bytes)
166        {
167            return Err(MetricError::InvalidLabels);
168        }
169        let mut series = self.inner.series.lock();
170        if let Some(value) = series.get(&BorrowedLabels(labels)) {
171            return Ok(value.clone());
172        }
173        if series.len() >= self.inner.limits.max_series_per_instrument {
174            return Err(MetricError::SeriesLimit);
175        }
176        self.inner
177            .series_count
178            .try_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
179                (current < self.inner.limits.max_total_series).then_some(current + 1)
180            })
181            .map_err(|_| MetricError::SeriesLimit)?;
182        let attributes: Vec<KeyValue> = self
183            .inner
184            .descriptor
185            .labels
186            .iter()
187            .zip(labels)
188            .map(|(key, value)| KeyValue::new(*key, (*value).to_owned()))
189            .collect();
190        let instrument = match &self.inner.instrument {
191            Instrument::Counter(counter) => BoundInstrument::Counter(counter.bind(&attributes)),
192            Instrument::Gauge(gauge) => BoundInstrument::Gauge(gauge.bind(&attributes)),
193            Instrument::Histogram(histogram) => {
194                BoundInstrument::Histogram(histogram.bind(&attributes))
195            }
196        };
197        let value = Arc::new(Series {
198            instrument,
199            counter: AtomicU64::new(0),
200            gauge: Mutex::new(0.0),
201        });
202        series.insert(
203            labels.iter().map(|label| (*label).to_owned()).collect(),
204            value.clone(),
205        );
206        Ok(value)
207    }
208}
209
210#[derive(Clone)]
211pub struct ProductCounter {
212    handle: MetricHandle,
213}
214#[derive(Clone)]
215pub struct ProductGauge {
216    handle: MetricHandle,
217}
218#[derive(Clone)]
219pub struct ProductHistogram {
220    handle: MetricHandle,
221}
222
223macro_rules! typed {
224    ($name:ident, $kind:ident) => {
225        impl $name {
226            /// # Errors
227            /// Returns `WrongKind` if the handle is not the requested instrument kind.
228            pub fn from_handle(handle: MetricHandle) -> MetricResult<Self> {
229                if handle.kind() != MetricKind::$kind {
230                    return Err(MetricError::WrongKind);
231                }
232                Ok(Self { handle })
233            }
234        }
235    };
236}
237typed!(ProductCounter, Counter);
238typed!(ProductGauge, Gauge);
239typed!(ProductHistogram, Histogram);
240
241impl ProductCounter {
242    pub fn inc(&self, labels: &[&str], value: u64) {
243        diagnose(self.try_inc(labels, value));
244    }
245    /// # Errors
246    /// Returns an error for closed providers, invalid labels, series limits or counter overflow.
247    pub fn try_inc(&self, labels: &[&str], value: u64) -> MetricResult<()> {
248        if !self.handle.active()? {
249            return Ok(());
250        }
251        let series = self.handle.series(labels)?;
252        if let BoundInstrument::Counter(counter) = &series.instrument {
253            series
254                .counter
255                .try_update(Ordering::Relaxed, Ordering::Relaxed, |current| {
256                    current.checked_add(value)
257                })
258                .map_err(|_| MetricError::InvalidValue)?;
259            counter.add(value);
260            Ok(())
261        } else {
262            Err(MetricError::WrongKind)
263        }
264    }
265}
266
267impl ProductGauge {
268    pub fn set(&self, labels: &[&str], value: f64) {
269        diagnose(self.try_set(labels, value));
270    }
271    pub fn add(&self, labels: &[&str], value: f64) {
272        diagnose(self.try_add(labels, value));
273    }
274    /// # Errors
275    /// Returns an error for closed providers, invalid labels, series limits or non-finite values.
276    pub fn try_set(&self, labels: &[&str], value: f64) -> MetricResult<()> {
277        self.update(labels, value, false)
278    }
279    /// # Errors
280    /// Returns an error for closed providers, invalid labels, series limits or non-finite results.
281    pub fn try_add(&self, labels: &[&str], value: f64) -> MetricResult<()> {
282        self.update(labels, value, true)
283    }
284    fn update(&self, labels: &[&str], value: f64, add: bool) -> MetricResult<()> {
285        if !self.handle.active()? {
286            return Ok(());
287        }
288        if !value.is_finite() {
289            return Err(MetricError::InvalidValue);
290        }
291        let series = self.handle.series(labels)?;
292        let mut current = series.gauge.lock();
293        let next = if add { *current + value } else { value };
294        if !next.is_finite() {
295            return Err(MetricError::InvalidValue);
296        }
297        if let BoundInstrument::Gauge(gauge) = &series.instrument {
298            *current = next;
299            // Keep the record ordered with the update; concurrent add/set must
300            // not publish an older absolute value after a newer one.
301            gauge.record(next);
302            Ok(())
303        } else {
304            Err(MetricError::WrongKind)
305        }
306    }
307}
308
309impl ProductHistogram {
310    pub fn observe(&self, labels: &[&str], value: f64) {
311        diagnose(self.try_observe(labels, value));
312    }
313    /// # Errors
314    /// Returns an error for closed providers, invalid labels, series limits or invalid observations.
315    pub fn try_observe(&self, labels: &[&str], value: f64) -> MetricResult<()> {
316        if !self.handle.active()? {
317            return Ok(());
318        }
319        if !value.is_finite() || value < 0.0 {
320            return Err(MetricError::InvalidValue);
321        }
322        let series = self.handle.series(labels)?;
323        if let BoundInstrument::Histogram(histogram) = &series.instrument {
324            histogram.record(value);
325            Ok(())
326        } else {
327            Err(MetricError::WrongKind)
328        }
329    }
330}
331
332fn diagnose(result: MetricResult<()>) {
333    if let Err(error) = result {
334        tracing::debug!(%error, "metric observation skipped");
335    }
336}
337
338/// Declare one typed schema and register it atomically against an explicit provider.
339#[macro_export]
340macro_rules! product_metrics {
341    ($(#[$meta:meta])* $vis:vis struct $name:ident {
342        $($(#[$field_meta:meta])* $field:ident : $kind:ident ($($args:expr),* $(,)? ) $(.with_unit($unit:expr))?),* $(,)?
343    }) => {
344        $(#[$meta])*
345        $vis struct $name { $($(#[$field_meta])* pub $field: $crate::product_metrics!(@type $kind),)* }
346        impl $name {
347            /// Registers this complete schema atomically on one instance.
348            ///
349            /// # Errors
350            /// Returns a registration error without changing the catalog when any descriptor is rejected.
351            pub fn register(provider: &$crate::MetricsProvider) -> $crate::MetricResult<Self> {
352                let descriptors = [$($crate::MetricDescriptor::$kind($($args),*)$(.with_unit($unit))?),*];
353                let mut handles = provider.register_batch(&descriptors)?.into_iter();
354                Ok(Self { $($field: <$crate::product_metrics!(@type $kind)>::from_handle(
355                    handles.next().ok_or($crate::MetricError::InvalidDescriptor)?
356                )?,)* })
357            }
358        }
359    };
360    (@type counter) => { $crate::ProductCounter };
361    (@type gauge) => { $crate::ProductGauge };
362    (@type histogram) => { $crate::ProductHistogram };
363    (@type histogram_with_buckets) => { $crate::ProductHistogram };
364}
aster_forge_metrics/instruments.rs

aster_forge_metrics/
instruments.rs
