dialcache/
prometheus.rs

1//! Prometheus metric exporter (feature `prometheus`).
2//!
3//! [`PrometheusObserver`] registers the twenty DialCache collectors on a
4//! [`Registry`] under the names, help text, label sets and histogram buckets
5//! of the TypeScript adapter, and feeds every [`Event`] to them.
6//!
7//! # Sharing one registry
8//!
9//! One observer owns one set of collectors. Instances that export to the same
10//! registry share the observer: it is `Clone`, and every clone feeds the same
11//! series. The `prometheus` crate reports a duplicate registration without a
12//! handle to the existing collector, so a second `new` for the same registry
13//! and prefix is a [`PrometheusError::Conflict`], as is an externally
14//! registered collector under one of the DialCache names. A failed
15//! constructor unregisters every collector it had registered, so no series is
16//! left behind; the `prometheus` crate nevertheless remembers each such name
17//! with the DialCache schema for the registry's lifetime, so only a collector
18//! with the same schema can take that name later.
19
20use std::fmt;
21use std::sync::Arc;
22
23use ::prometheus::core::Collector;
24use ::prometheus::{HistogramOpts, HistogramVec, IntCounterVec, Opts, Registry};
25
26use crate::metrics::MetricKind;
27use crate::observe::{Event, Observer};
28
29const TIMER_BUCKETS: &[f64] = &[
30    0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 2.5, 5.0, 10.0,
31];
32const SIZE_BUCKETS: &[f64] = &[
33    100.0,
34    1_000.0,
35    10_000.0,
36    100_000.0,
37    1_000_000.0,
38    10_000_000.0,
39];
40const RATIO_BUCKETS: &[f64] = &[0.05, 0.1, 0.2, 0.3, 0.5, 0.7, 0.9, 1.0];
41/// Value ages span seconds to the 365-day TTL ceiling: 1s..15m, then 1h, 3h, 12h, 1d, 3d, 7d.
42const VALUE_AGE_BUCKETS: &[f64] = &[
43    1.0, 5.0, 15.0, 60.0, 300.0, 900.0, 3_600.0, 10_800.0, 43_200.0, 86_400.0, 259_200.0, 604_800.0,
44];
45const FUTURE_TIMESTAMP_OFFSET_BUCKETS: &[f64] = &[
46    0.001, 0.005, 0.01, 0.025, 0.05, 0.1, 0.25, 0.5, 1.0, 5.0, 15.0, 60.0, 300.0, 900.0, 3_600.0,
47    10_800.0, 43_200.0,
48];
49
50/// The wire schema of one collector: what a scrape exposes.
51#[derive(Debug, Clone, PartialEq)]
52pub struct CollectorSchema {
53    /// The event kind the collector receives.
54    pub kind: MetricKind,
55    /// The fully qualified metric name, prefix included.
56    pub name: String,
57    /// The Prometheus help text, spelled as in the TypeScript adapter.
58    pub help: &'static str,
59    /// Label names in wire order.
60    pub labels: &'static [&'static str],
61    /// Histogram bucket upper bounds; empty for counters.
62    pub buckets: &'static [f64],
63}
64
65impl CollectorSchema {
66    /// `true` for the counter vectors; every other collector is a histogram.
67    pub fn is_counter(&self) -> bool {
68        self.kind.is_counter()
69    }
70}
71
72/// The twenty collectors DialCache registers, in registration order, with
73/// their names under `prefix`.
74pub fn schemas(prefix: &str) -> Vec<CollectorSchema> {
75    let schema = |kind: MetricKind,
76                  suffix: &str,
77                  help: &'static str,
78                  buckets: &'static [f64]|
79     -> CollectorSchema {
80        CollectorSchema {
81            kind,
82            name: format!("{prefix}dialcache_{suffix}"),
83            help,
84            labels: kind.label_names(),
85            buckets,
86        }
87    };
88    vec![
89        schema(
90            MetricKind::Disabled,
91            "disabled_counter",
92            "Requests where DialCache skipped a cache layer.",
93            &[],
94        ),
95        schema(
96            MetricKind::Miss,
97            "miss_counter",
98            "DialCache cache misses.",
99            &[],
100        ),
101        schema(
102            MetricKind::Request,
103            "request_counter",
104            "Total DialCache cache-layer requests.",
105            &[],
106        ),
107        schema(
108            MetricKind::Error,
109            "error_counter",
110            "Errors during DialCache cache operations or fallback execution.",
111            &[],
112        ),
113        schema(
114            MetricKind::Invalidation,
115            "invalidation_counter",
116            "DialCache invalidation calls by key type and layer.",
117            &[],
118        ),
119        schema(
120            MetricKind::Coalesced,
121            "coalesced_counter",
122            "DialCache requests coalesced onto in-flight work by sharing scope.",
123            &[],
124        ),
125        schema(
126            MetricKind::ShadowValidation,
127            "shadow_validation_counter",
128            "Sampled DialCache Redis shadow-validation outcomes.",
129            &[],
130        ),
131        schema(
132            MetricKind::ShadowValueAge,
133            "shadow_value_age_histogram",
134            "Age in seconds of the validated Redis value at DialCache shadow verdict time.",
135            VALUE_AGE_BUCKETS,
136        ),
137        schema(
138            MetricKind::ShadowMismatchLogging,
139            "shadow_mismatch_logging_counter",
140            "DialCache confirmed shadow-mismatch warning outcomes.",
141            &[],
142        ),
143        schema(
144            MetricKind::FutureTimestampOffset,
145            "future_timestamp_offset_histogram",
146            "Positive offset in seconds of Redis frames dated after the observing DialCache process clock.",
147            FUTURE_TIMESTAMP_OFFSET_BUCKETS,
148        ),
149        schema(
150            MetricKind::StaleRecovery,
151            "stale_recovery_counter",
152            "DialCache stale-on-error Redis recovery outcomes.",
153            &[],
154        ),
155        schema(
156            MetricKind::StaleRecoveryValueAge,
157            "stale_recovery_value_age_histogram",
158            "Age in seconds of Redis values served by DialCache stale-on-error recovery.",
159            VALUE_AGE_BUCKETS,
160        ),
161        schema(
162            MetricKind::Compression,
163            "compression_counter",
164            "DialCache Redis payload compression and decompression outcomes.",
165            &[],
166        ),
167        schema(
168            MetricKind::Get,
169            "get_timer",
170            "DialCache cache get latency in seconds.",
171            TIMER_BUCKETS,
172        ),
173        schema(
174            MetricKind::Fallback,
175            "fallback_timer",
176            "Time DialCache waited for the fallback function in seconds.",
177            TIMER_BUCKETS,
178        ),
179        schema(
180            MetricKind::Serialization,
181            "serialization_timer",
182            "DialCache serialization latency in seconds.",
183            TIMER_BUCKETS,
184        ),
185        schema(
186            MetricKind::Size,
187            "size_histogram",
188            "Serialized DialCache value sizes in bytes.",
189            SIZE_BUCKETS,
190        ),
191        schema(
192            MetricKind::StoredSize,
193            "stored_size_histogram",
194            "Stored DialCache payload sizes in bytes, after compression and escaping.",
195            SIZE_BUCKETS,
196        ),
197        schema(
198            MetricKind::CompressionRatio,
199            "compression_ratio_histogram",
200            "Compressed-to-original DialCache payload size ratio for compressed writes.",
201            RATIO_BUCKETS,
202        ),
203        schema(
204            MetricKind::CompressionDuration,
205            "compression_timer",
206            "DialCache payload compression and decompression latency in seconds.",
207            TIMER_BUCKETS,
208        ),
209    ]
210}
211
212/// Construction failures of a [`PrometheusObserver`].
213#[derive(Debug)]
214pub enum PrometheusError {
215    /// A collector with this name is already registered: by another observer
216    /// (clone that observer instead of constructing a second one), by someone
217    /// else, or with another schema. Use a unique prefix or another registry.
218    /// The collectors registered before the collision were unregistered
219    /// again; their names stay bound to the DialCache schema in that registry.
220    Conflict {
221        /// The fully qualified collector name that collided.
222        name: String,
223        /// The registry's rejection.
224        source: ::prometheus::Error,
225    },
226    /// The `prometheus` crate rejected a collector definition, which only a
227    /// prefix that is not a valid metric name fragment can cause.
228    InvalidCollector {
229        /// The fully qualified collector name that was rejected.
230        name: String,
231        /// The crate's rejection.
232        source: ::prometheus::Error,
233    },
234}
235
236impl fmt::Display for PrometheusError {
237    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
238        match self {
239            PrometheusError::Conflict { name, source } => write!(
240                f,
241                "Prometheus collector {name:?} already exists with an incompatible or externally \
242                 owned schema; use a unique prefix or registry: {source}"
243            ),
244            PrometheusError::InvalidCollector { name, source } => {
245                write!(f, "invalid Prometheus collector {name:?}: {source}")
246            }
247        }
248    }
249}
250
251impl std::error::Error for PrometheusError {
252    fn source(&self) -> Option<&(dyn std::error::Error + 'static)> {
253        match self {
254            PrometheusError::Conflict { source, .. }
255            | PrometheusError::InvalidCollector { source, .. } => Some(source),
256        }
257    }
258}
259
260#[derive(Clone)]
261enum Vector {
262    Counter(IntCounterVec),
263    Histogram(HistogramVec),
264}
265
266impl Vector {
267    fn boxed(&self) -> Box<dyn Collector> {
268        match self {
269            Vector::Counter(vector) => Box::new(vector.clone()),
270            Vector::Histogram(vector) => Box::new(vector.clone()),
271        }
272    }
273}
274
275/// The collectors of one (registry, prefix) pair, indexed by [`MetricKind::index`].
276struct Group {
277    vectors: Vec<Vector>,
278}
279
280impl Group {
281    fn build(prefix: &str) -> Result<Group, PrometheusError> {
282        let mut vectors: Vec<Option<Vector>> = vec![None; MetricKind::ALL.len()];
283        for schema in schemas(prefix) {
284            let vector = if schema.is_counter() {
285                IntCounterVec::new(Opts::new(&schema.name, schema.help), schema.labels)
286                    .map(Vector::Counter)
287            } else {
288                HistogramVec::new(
289                    HistogramOpts::new(&schema.name, schema.help).buckets(schema.buckets.to_vec()),
290                    schema.labels,
291                )
292                .map(Vector::Histogram)
293            }
294            .map_err(|source| PrometheusError::InvalidCollector {
295                name: schema.name.clone(),
296                source,
297            })?;
298            vectors[schema.kind.index()] = Some(vector);
299        }
300        Ok(Group {
301            vectors: vectors
302                .into_iter()
303                .map(|vector| vector.expect("every kind has a schema"))
304                .collect(),
305        })
306    }
307
308    /// Register every collector on `registry`; roll back on the first failure.
309    fn register(&self, registry: &Registry, prefix: &str) -> Result<(), PrometheusError> {
310        let mut registered: Vec<&Vector> = Vec::new();
311        for schema in schemas(prefix) {
312            let vector = &self.vectors[schema.kind.index()];
313            if let Err(source) = registry.register(vector.boxed()) {
314                for done in registered {
315                    let _ = registry.unregister(done.boxed());
316                }
317                return Err(PrometheusError::Conflict {
318                    name: schema.name,
319                    source,
320                });
321            }
322            registered.push(vector);
323        }
324        Ok(())
325    }
326}
327
328/// Publishes DialCache diagnostics to Prometheus collectors.
329///
330/// Clones share the collectors; hand one observer to every instance that
331/// exports to the same registry.
332#[derive(Clone)]
333pub struct PrometheusObserver {
334    group: Arc<Group>,
335    prefix: String,
336}
337
338impl fmt::Debug for PrometheusObserver {
339    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
340        f.debug_struct("PrometheusObserver")
341            .field("prefix", &self.prefix)
342            .finish_non_exhaustive()
343    }
344}
345
346impl PrometheusObserver {
347    /// Register the DialCache collectors named `<prefix>dialcache_*` on
348    /// `registry`. Registering the same names twice on one registry is a
349    /// [`PrometheusError::Conflict`]: clone the observer instead.
350    pub fn new(registry: &Registry, prefix: &str) -> Result<PrometheusObserver, PrometheusError> {
351        let group = Arc::new(Group::build(prefix)?);
352        group.register(registry, prefix)?;
353        Ok(PrometheusObserver {
354            group,
355            prefix: prefix.to_string(),
356        })
357    }
358
359    /// The wire schemas of this observer's collectors.
360    pub fn schemas(&self) -> Vec<CollectorSchema> {
361        schemas(&self.prefix)
362    }
363}
364
365impl Observer for PrometheusObserver {
366    fn observe(&self, event: &Event) {
367        let kind = MetricKind::of(event);
368        let labels = MetricKind::labels(event);
369        let values: Vec<&str> = labels.iter().map(|(_, value)| value.as_str()).collect();
370        let result = match &self.group.vectors[kind.index()] {
371            Vector::Counter(vector) => vector
372                .get_metric_with_label_values(&values)
373                .map(|metric| metric.inc()),
374            Vector::Histogram(vector) => vector
375                .get_metric_with_label_values(&values)
376                .map(|metric| metric.observe(MetricKind::value(event))),
377        };
378        if let Err(error) = result {
379            log::warn!(
380                target: "dialcache",
381                "Dropped DialCache {} metric with labels {:?}: {error}",
382                kind.as_str(),
383                labels
384            );
385        }
386    }
387
388    fn observes_shadow_outcomes(&self) -> bool {
389        true
390    }
391}
392
393#[cfg(test)]
394mod tests {
395    use super::*;
396
397    #[test]
398    fn schemas_cover_every_kind_once_with_matching_label_names() {
399        let all = schemas("p_");
400        assert_eq!(all.len(), MetricKind::ALL.len());
401        let mut seen = std::collections::HashSet::new();
402        for schema in &all {
403            assert!(seen.insert(schema.kind), "{:?} twice", schema.kind);
404            assert!(schema.name.starts_with("p_dialcache_"));
405            assert_eq!(schema.labels, schema.kind.label_names());
406            assert_eq!(schema.buckets.is_empty(), schema.is_counter());
407        }
408    }
409
410    #[test]
411    fn invalid_prefix_is_an_error_not_a_panic() {
412        let registry = Registry::new();
413        let error = PrometheusObserver::new(&registry, "bad prefix ").unwrap_err();
414        assert!(
415            matches!(error, PrometheusError::InvalidCollector { .. }),
416            "{error}"
417        );
418        assert!(registry.gather().is_empty());
419    }
420}
dialcache/prometheus.rs

dialcache/
prometheus.rs
