diff --git a/foundations-metrics/Cargo.toml b/foundations-metrics/Cargo.toml index 23152ebb..a57ff8b9 100644 --- a/foundations-metrics/Cargo.toml +++ b/foundations-metrics/Cargo.toml @@ -16,6 +16,7 @@ serde = { workspace = true, features = ["derive"] } ryu = "1.0.23" parking_lot = { workspace = true } prost = { workspace = true } +prost-types = { workspace = true } [lints] workspace = true diff --git a/foundations-metrics/src/collect.rs b/foundations-metrics/src/collect.rs index a3c55e6f..ffec728f 100644 --- a/foundations-metrics/src/collect.rs +++ b/foundations-metrics/src/collect.rs @@ -447,10 +447,16 @@ mod tests { exemplars: vec![ Exemplar { label: vec![label("bad name", "bad")], + timestamp: Some(Default::default()), + ..Default::default() + }, + Exemplar { + label: vec![label("trace_id", "missing_timestamp")], ..Default::default() }, Exemplar { label: vec![label("trace_id", "good")], + timestamp: Some(Default::default()), ..Default::default() }, ], diff --git a/foundations-metrics/src/encoding/mod.rs b/foundations-metrics/src/encoding/mod.rs index dbd5c12d..59130a72 100644 --- a/foundations-metrics/src/encoding/mod.rs +++ b/foundations-metrics/src/encoding/mod.rs @@ -158,10 +158,16 @@ mod tests { exemplars: vec![ Exemplar { label: vec![label("bad#name", "nonstandard")], + timestamp: Some(Default::default()), + ..Default::default() + }, + Exemplar { + label: vec![label("trace_id", "missing_timestamp")], ..Default::default() }, Exemplar { label: vec![label("trace_id", "good")], + timestamp: Some(Default::default()), ..Default::default() }, ], diff --git a/foundations-metrics/src/encoding/text.rs b/foundations-metrics/src/encoding/text.rs index c3f75b2e..c629794e 100644 --- a/foundations-metrics/src/encoding/text.rs +++ b/foundations-metrics/src/encoding/text.rs @@ -302,9 +302,13 @@ fn write_sample( write_float(output, timestamp_ms as f64 / 1_000.0); } - if let Some(exemplar) = exemplar.filter(|exemplar| !exemplar.label.is_empty()) { + if let Some(exemplar) = exemplar { output.push_str(" # "); - write_labels(output, &exemplar.label, None); + if exemplar.label.is_empty() { + output.push_str("{}"); + } else { + write_labels(output, &exemplar.label, None); + } output.push(' '); write_float(output, exemplar.value.unwrap_or_default()); diff --git a/foundations-metrics/src/labels/serializer.rs b/foundations-metrics/src/labels/serializer.rs index 780aa058..4c70fc62 100644 --- a/foundations-metrics/src/labels/serializer.rs +++ b/foundations-metrics/src/labels/serializer.rs @@ -2,7 +2,7 @@ use std::fmt; use foundations_metrics_registry::proto::LabelPair; use serde::Serialize; -use serde::ser::{Impossible, SerializeStruct, Serializer}; +use serde::ser::{Impossible, SerializeSeq, SerializeStruct, SerializeTuple, Serializer}; use super::LabelError; use crate::validation::{NAME_REQUIREMENT, is_valid_name}; @@ -14,8 +14,8 @@ pub(super) struct LabelSetSerializer; impl Serializer for LabelSetSerializer { type Ok = Vec; type Error = LabelError; - type SerializeSeq = Impossible; - type SerializeTuple = Impossible; + type SerializeSeq = LabelSequenceSerializer; + type SerializeTuple = LabelTupleSerializer; type SerializeTupleStruct = Impossible; type SerializeTupleVariant = Impossible; type SerializeMap = Impossible; @@ -148,12 +148,21 @@ impl Serializer for LabelSetSerializer { Err(invalid_label_set()) } - fn serialize_seq(self, _len: Option) -> Result { - Err(invalid_label_set()) + fn serialize_seq(self, len: Option) -> Result { + Ok(LabelSequenceSerializer { + labels: Vec::with_capacity(len.unwrap_or_default()), + }) } - fn serialize_tuple(self, _len: usize) -> Result { - Err(invalid_label_set()) + fn serialize_tuple(self, len: usize) -> Result { + if len != 2 { + return Err(LabelError::new("label pairs must contain a name and value")); + } + + Ok(LabelTupleSerializer { + name: None, + value: None, + }) } fn serialize_tuple_struct( @@ -193,6 +202,77 @@ impl Serializer for LabelSetSerializer { } } +pub(super) struct LabelSequenceSerializer { + labels: Vec, +} + +impl SerializeSeq for LabelSequenceSerializer { + type Ok = Vec; + type Error = LabelError; + + fn serialize_element(&mut self, value: &T) -> Result<(), Self::Error> + where + T: Serialize + ?Sized, + { + let mut labels = value.serialize(LabelSetSerializer)?; + if labels.len() != 1 { + return Err(LabelError::new( + "label sequences must contain name-value pairs", + )); + } + self.labels + .push(labels.pop().expect("one label was encoded")); + Ok(()) + } + + fn end(self) -> Result { + Ok(self.labels) + } +} + +pub(super) struct LabelTupleSerializer { + name: Option, + value: Option, +} + +impl SerializeTuple for LabelTupleSerializer { + type Ok = Vec; + type Error = LabelError; + + fn serialize_element(&mut self, value: &T) -> Result<(), Self::Error> + where + T: Serialize + ?Sized, + { + if self.name.is_none() { + let name = value.serialize(LabelValueSerializer)?; + validate_label_name(&name)?; + self.name = Some(name); + } else if self.value.is_none() { + self.value = Some(value.serialize(LabelValueSerializer)?); + } else { + return Err(LabelError::new( + "label pairs must contain exactly one name and value", + )); + } + + Ok(()) + } + + fn end(self) -> Result { + let name = self + .name + .ok_or_else(|| LabelError::new("label pair is missing its name"))?; + let value = self + .value + .ok_or_else(|| LabelError::new("label pair is missing its value"))?; + + Ok(vec![LabelPair { + name: Some(name), + value: Some(value), + }]) + } +} + // Adapted from prometools' `serde::top::StructSerializer` // (https://github.com/nox/prometools, licensed MIT OR Apache-2.0). pub(super) struct LabelPairSerializer { @@ -468,6 +548,30 @@ mod tests { ); } + #[test] + fn serializes_legacy_name_value_sequences() { + let pairs = to_label_pairs(&vec![ + ("trace_id", "abc"), + ("span_id", "def"), + ("trace:id", "ghi"), + ]) + .unwrap(); + let values: Vec<_> = pairs + .iter() + .map(|pair| { + ( + pair.name.as_deref().unwrap(), + pair.value.as_deref().unwrap(), + ) + }) + .collect(); + + assert_eq!( + values, + [("trace_id", "abc"), ("span_id", "def"), ("trace:id", "ghi"),] + ); + } + #[test] fn unit_is_an_empty_label_set() { assert!(to_label_pairs(&()).unwrap().is_empty()); diff --git a/foundations-metrics/src/lib.rs b/foundations-metrics/src/lib.rs index 0b50e154..eb33dfe4 100644 --- a/foundations-metrics/src/lib.rs +++ b/foundations-metrics/src/lib.rs @@ -23,8 +23,9 @@ pub use foundations_metrics_registry::{ }; pub use labels::{LabelError, to_label_pairs}; pub use metrics::{ - Counter, CounterAtomic, Family, FamilyMetricGuard, Gauge, GaugeAtomic, GaugeGuard, Histogram, - HistogramBuilder, HistogramSnapshot, HistogramTimer, MetricConstructor, NativeHistogram, - NativeHistogramBuilder, RangeGauge, TimeHistogram, + Counter, CounterAtomic, CounterWithExemplar, Exemplar, Family, FamilyMetricGuard, Gauge, + GaugeAtomic, GaugeGuard, Histogram, HistogramBuilder, HistogramSnapshot, HistogramTimer, + HistogramWithExemplars, MetricConstructor, NativeHistogram, NativeHistogramBuilder, + NativeHistogramWithExemplars, RangeGauge, TimeHistogram, }; pub use registered::NamedMetric; diff --git a/foundations-metrics/src/metrics/counter.rs b/foundations-metrics/src/metrics/counter.rs index 31aaa84d..3b7d2eb9 100644 --- a/foundations-metrics/src/metrics/counter.rs +++ b/foundations-metrics/src/metrics/counter.rs @@ -135,7 +135,7 @@ impl Default for Counter { /// /// The name/help are left empty here; they are populated at registration and /// encode time. -fn encode_counter(value: f64) -> Vec { +pub(super) fn encode_counter(value: f64) -> Vec { vec![MetricFamily { name: Some(String::new()), help: None, diff --git a/foundations-metrics/src/metrics/exemplar.rs b/foundations-metrics/src/metrics/exemplar.rs new file mode 100644 index 00000000..6d8a84db --- /dev/null +++ b/foundations-metrics/src/metrics/exemplar.rs @@ -0,0 +1,710 @@ +use std::collections::HashMap; +use std::marker::PhantomData; +use std::sync::Arc; +use std::sync::atomic::AtomicU64; +use std::time::SystemTime; + +use foundations_metrics_registry::proto; +use parking_lot::{MappedRwLockReadGuard, RwLock, RwLockReadGuard}; +use prost_types::Timestamp; +use serde::Serialize; + +use super::counter::encode_counter; +use super::histogram::encode_snapshot; +use super::{ + CounterAtomic, Histogram, HistogramBuilder, IntoF64, MetricConstructor, NativeHistogram, + NativeHistogramBuilder, +}; +use crate::diagnostics::report_collect_error; +use crate::labels::to_label_pairs; +use crate::validation::EXEMPLAR_SERIALIZATION_ERROR_LABEL; +use crate::{MetricFamily, value::EncodeMetricValue}; + +/// Labels and a sampled value associated with a metric observation. +/// +/// Exemplar values are exposed through [`CounterWithExemplar::get`]. Their fields +/// remain private; exemplars are created by recording labeled metric updates. +#[derive(Debug)] +pub struct Exemplar { + label_set: Arc, + value: V, + timestamp: Option, +} + +impl Exemplar { + /// Returns the exemplar's label set. + pub fn label_set(&self) -> &S { + &self.label_set + } + + /// Returns the sampled increment or observation. + pub fn value(&self) -> &V { + &self.value + } +} + +impl Exemplar { + fn new(label_set: S, value: V, timestamp: Option) -> Self { + Self { + label_set: Arc::new(label_set), + value, + timestamp, + } + } +} + +impl Clone for Exemplar { + fn clone(&self) -> Self { + Self { + label_set: Arc::clone(&self.label_set), + value: self.value.clone(), + timestamp: self.timestamp, + } + } +} + +impl Exemplar { + fn encode(&self) -> Result + where + S: Serialize, + V: Clone + IntoF64, + { + Ok(proto::Exemplar { + label: to_label_pairs(self.label_set.as_ref()).map_err(|error| error.to_string())?, + value: Some(self.value.clone().into_f64()), + timestamp: self.timestamp, + }) + } +} + +fn finish_exemplar(exemplar: Option>) -> Option { + match exemplar { + Some(Ok(exemplar)) => Some(exemplar), + // Defer reporting to validation, after any enclosing Family lock has + // been released. + Some(Err(error)) => Some(proto::Exemplar { + label: vec![proto::LabelPair { + name: Some(EXEMPLAR_SERIALIZATION_ERROR_LABEL.to_owned()), + value: Some(error), + }], + ..Default::default() + }), + None => None, + } +} + +/// A monotonically increasing counter that records an exemplar with an update. +/// +/// Calling [`inc_by`](Self::inc_by) with a label set replaces the previous +/// exemplar. Calling it without a label set clears the previous exemplar. The +/// exemplar value is the increment. [`get`](Self::get) returns the cumulative +/// counter value and read-only access to the current exemplar. Clones share both +/// values. +/// +/// Exemplar labels are serialized with [`serde::Serialize`] during collection. +#[derive(Debug)] +pub struct CounterWithExemplar { + state: Arc>>, + marker: PhantomData, +} + +#[derive(Debug)] +struct CounterWithExemplarState { + value: A, + exemplar: Option>, +} + +impl CounterWithExemplar +where + N: Clone, + A: CounterAtomic, +{ + /// Increments the counter by `value`, returning the previous total. + /// + /// A provided label set replaces the stored exemplar. `None` clears it. + pub fn inc_by(&self, value: N, label_set: Option) -> N { + let exemplar = label_set.map(|label_set| Exemplar::new(label_set, value.clone(), None)); + let mut state = self.state.write(); + state.exemplar = exemplar; + state.value.inc_by(value) + } + + /// Returns the cumulative counter value and the current exemplar. + /// + /// The exemplar guard keeps the counter read-locked until it is dropped. + pub fn get(&self) -> (N, MappedRwLockReadGuard<'_, Option>>) { + let state = self.state.read(); + let value = state.value.get(); + let exemplar = RwLockReadGuard::map(state, |state| &state.exemplar); + (value, exemplar) + } + + /// Returns read-only access to the underlying counter storage. + /// + /// The returned guard prevents updates until it is dropped. + pub fn inner(&self) -> MappedRwLockReadGuard<'_, A> { + RwLockReadGuard::map(self.state.read(), |state| &state.value) + } +} + +impl Clone for CounterWithExemplar { + fn clone(&self) -> Self { + Self { + state: Arc::clone(&self.state), + marker: PhantomData, + } + } +} + +impl Default for CounterWithExemplar { + fn default() -> Self { + Self { + state: Arc::new(RwLock::new(CounterWithExemplarState { + value: A::default(), + exemplar: None, + })), + marker: PhantomData, + } + } +} + +impl EncodeMetricValue for CounterWithExemplar +where + S: Serialize + Send + Sync + 'static, + N: Clone + IntoF64, + A: CounterAtomic + Send + Sync + 'static, +{ + fn encode_metric_value(&self) -> Vec { + let state = self.state.read(); + let value = state.value.get().into_f64(); + let exemplar = state.exemplar.clone(); + drop(state); + + let mut families = encode_counter(value); + let counter = families[0].metric[0] + .counter + .as_mut() + .expect("counter encoder always creates a counter value"); + counter.exemplar = finish_exemplar(exemplar.as_ref().map(Exemplar::encode)); + families + } +} + +/// A classic fixed-bucket histogram that retains one exemplar per bucket. +/// +/// A labeled observation replaces the exemplar in its bucket. An unlabeled +/// observation updates the histogram without clearing an existing exemplar. +/// Clones share histogram and exemplar storage. +#[derive(Clone, Debug)] +pub struct HistogramWithExemplars { + state: Arc>>, +} + +#[derive(Debug)] +struct HistogramWithExemplarsState { + histogram: Histogram, + exemplars: HashMap>, +} + +impl HistogramWithExemplars { + /// Creates a histogram with the provided inclusive upper bounds. + /// + /// Bounds are sorted and a terminal `f64::MAX` bucket is appended. + pub fn new(bounds: impl IntoIterator) -> Self { + Self { + state: Arc::new(RwLock::new(HistogramWithExemplarsState { + histogram: Histogram::new(bounds), + exemplars: HashMap::new(), + })), + } + } + + /// Records an observation and optionally associates it with an exemplar. + pub fn observe(&self, value: f64, label_set: Option) { + let exemplar = label_set.map(|label_set| Exemplar::new(label_set, value, None)); + let mut state = self.state.write(); + let bucket = state.histogram.observe_and_bucket(value); + + if let (Some(bucket), Some(exemplar)) = (bucket, exemplar) { + state.exemplars.insert(bucket, exemplar); + } + } +} + +impl EncodeMetricValue for HistogramWithExemplars +where + S: Serialize + Send + Sync + 'static, +{ + fn encode_metric_value(&self) -> Vec { + let state = self.state.read(); + let snapshot = state.histogram.snapshot(); + let exemplars: Vec<_> = state + .exemplars + .iter() + .map(|(&index, exemplar)| (index, exemplar.clone())) + .collect(); + drop(state); + + let mut families = encode_snapshot(snapshot); + let buckets = &mut families[0].metric[0] + .histogram + .as_mut() + .expect("histogram encoder always creates a histogram value") + .bucket; + + for (index, exemplar) in exemplars { + if let Some(bucket) = buckets.get_mut(index) { + bucket.exemplar = finish_exemplar(Some(exemplar.encode())); + } + } + + families + } +} + +impl MetricConstructor> for HistogramBuilder { + fn new_metric(&self) -> HistogramWithExemplars { + HistogramWithExemplars::new(self.buckets.iter().copied()) + } +} + +/// A native histogram that retains its latest labeled observation as an exemplar. +/// +/// Native histogram exemplars require timestamps in the Prometheus protobuf +/// model. The timestamp is captured when the labeled observation is recorded. +/// Native histograms require protobuf exposition; OpenMetrics text encoding +/// does not expose their sparse buckets or exemplars. +#[derive(Clone, Debug)] +pub struct NativeHistogramWithExemplars { + state: Arc>>, +} + +#[derive(Debug)] +struct NativeHistogramWithExemplarsState { + histogram: NativeHistogram, + exemplar: Option>, +} + +impl NativeHistogramWithExemplars { + /// Creates a native histogram with the given bucket growth `factor`. + /// + /// # Panics + /// + /// Panics if `factor` is not greater than `1.0`. + pub fn new(factor: f64) -> Self { + NativeHistogramBuilder::new(factor).new_metric() + } + + /// Records an observation and optionally retains it as the latest exemplar. + pub fn observe(&self, value: f64, label_set: Option) { + let exemplar = label_set.map(|label_set| Exemplar::new(label_set, value, None)); + let mut state = self.state.write(); + state.histogram.observe(value); + + if let Some(mut exemplar) = exemplar { + exemplar.timestamp = Some(SystemTime::now().into()); + state.exemplar = Some(exemplar); + } + } +} + +impl EncodeMetricValue for NativeHistogramWithExemplars +where + S: Serialize + Send + Sync + 'static, +{ + fn encode_metric_value(&self) -> Vec { + let state = self.state.read(); + let families = state.histogram.try_encode_metric_value(); + let exemplar = state.exemplar.clone(); + drop(state); + + let mut families = match families { + Ok(families) => families, + Err(error) => { + report_collect_error(format_args!( + "non-fatal error while collecting metrics: skipped a native histogram; protobuf encoding failed: {error}" + )); + return Vec::new(); + } + }; + + if let Some(exemplar) = finish_exemplar(exemplar.as_ref().map(Exemplar::encode)) { + for histogram in families + .iter_mut() + .flat_map(|family| &mut family.metric) + .filter_map(|metric| metric.histogram.as_mut()) + { + histogram.exemplars.push(exemplar.clone()); + } + } + + families + } +} + +impl MetricConstructor> for NativeHistogramBuilder { + fn new_metric(&self) -> NativeHistogramWithExemplars { + NativeHistogramWithExemplars { + state: Arc::new(RwLock::new(NativeHistogramWithExemplarsState { + histogram: NativeHistogramBuilder::new_metric(self), + exemplar: None, + })), + } + } +} + +#[cfg(test)] +mod tests { + use foundations_metrics_registry::proto::MetricType; + use prost::Message; + use serde::Serialize; + + use super::*; + use crate::{ + CollectionOptions, EncodeMetric, Family, NamedMetric, RegistrationMetadata, + ServiceNameFormat, collect, encode_to_protobuf, encode_to_text, register, + }; + + #[derive(Clone, Debug, Serialize)] + struct TraceLabels { + trace_id: &'static str, + } + + fn trace_id(exemplar: &proto::Exemplar) -> Option<&str> { + exemplar + .label + .iter() + .find(|label| label.name.as_deref() == Some("trace_id")) + .and_then(|label| label.value.as_deref()) + } + + #[test] + fn counter_replaces_and_clears_exemplars_and_clones_share_storage() { + let counter = CounterWithExemplar::::default(); + let clone = counter.clone(); + + assert_eq!( + counter.inc_by(2, Some(TraceLabels { trace_id: "first" })), + 0 + ); + assert_eq!(clone.inc_by(3, Some(TraceLabels { trace_id: "latest" })), 2); + + let families = counter.encode_metric_value(); + let encoded = families[0].metric[0].counter.as_ref().unwrap(); + let exemplar = encoded.exemplar.as_ref().unwrap(); + assert_eq!(encoded.value, Some(5.0)); + assert_eq!(exemplar.value, Some(3.0)); + assert_eq!(trace_id(exemplar), Some("latest")); + assert!(exemplar.timestamp.is_none()); + + let (value, exemplar) = counter.get(); + assert_eq!(value, 5); + assert!(exemplar.is_some()); + drop(exemplar); + assert_eq!( + counter.inner().load(std::sync::atomic::Ordering::Relaxed), + 5 + ); + + assert_eq!(counter.inc_by(1, None), 5); + assert_eq!(counter.get().0, 6); + assert!( + counter.encode_metric_value()[0].metric[0] + .counter + .as_ref() + .unwrap() + .exemplar + .is_none() + ); + } + + #[test] + fn classic_histogram_retains_latest_exemplar_per_bucket() { + let histogram = HistogramWithExemplars::new([1.0, 2.0]); + histogram.observe(0.5, Some(TraceLabels { trace_id: "first" })); + histogram.observe( + 0.75, + Some(TraceLabels { + trace_id: "replacement", + }), + ); + histogram.observe(0.8, None); + histogram.observe( + 1.5, + Some(TraceLabels { + trace_id: "second_bucket", + }), + ); + + let families = histogram.encode_metric_value(); + let encoded = families[0].metric[0].histogram.as_ref().unwrap(); + assert_eq!(encoded.sample_count, Some(4)); + assert_eq!(encoded.sample_sum, Some(3.55)); + assert_eq!( + encoded + .bucket + .iter() + .map(|bucket| bucket.cumulative_count) + .collect::>(), + [Some(3), Some(4), Some(4)] + ); + assert_eq!( + trace_id(encoded.bucket[0].exemplar.as_ref().unwrap()), + Some("replacement") + ); + assert_eq!( + encoded.bucket[0].exemplar.as_ref().unwrap().value, + Some(0.75) + ); + assert_eq!( + trace_id(encoded.bucket[1].exemplar.as_ref().unwrap()), + Some("second_bucket") + ); + assert!(encoded.bucket[2].exemplar.is_none()); + } + + #[test] + fn native_histogram_retains_latest_timestamped_exemplar() { + let histogram = NativeHistogramWithExemplars::new(1.1); + let clone = histogram.clone(); + histogram.observe(0.5, Some(TraceLabels { trace_id: "first" })); + clone.observe(2.0, None); + clone.observe(3.0, Some(TraceLabels { trace_id: "latest" })); + + let families = histogram.encode_metric_value(); + let encoded = families[0].metric[0].histogram.as_ref().unwrap(); + assert_eq!(encoded.sample_count, Some(3)); + assert_eq!(encoded.sample_sum, Some(5.5)); + assert!(!encoded.positive_span.is_empty()); + assert_eq!(encoded.exemplars.len(), 1); + assert_eq!(encoded.exemplars[0].value, Some(3.0)); + assert_eq!(trace_id(&encoded.exemplars[0]), Some("latest")); + assert!(encoded.exemplars[0].timestamp.is_some()); + } + + #[test] + fn exemplar_label_failure_drops_only_the_exemplar() { + let counter = CounterWithExemplar::<&'static str>::default(); + counter.inc_by(2, Some("not a label set")); + + let families = NamedMetric::new("serialization_failure", "", counter).encode(); + let sentinel = &families[0].metric[0] + .counter + .as_ref() + .unwrap() + .exemplar + .as_ref() + .unwrap() + .label[0]; + assert_eq!( + sentinel.name.as_deref(), + Some(EXEMPLAR_SERIALIZATION_ERROR_LABEL) + ); + assert_eq!( + sentinel.value.as_deref(), + Some("metric labels must serialize as a struct or unit") + ); + + let payload = encode_to_protobuf(&families); + let encoded = MetricFamily::decode_length_delimited(payload.as_slice()).unwrap(); + let encoded = encoded.metric[0].counter.as_ref().unwrap(); + assert_eq!(encoded.value, Some(2.0)); + assert!(encoded.exemplar.is_none()); + } + + #[test] + fn empty_exemplar_label_sets_are_retained_in_text() { + let counter = CounterWithExemplar::<()>::default(); + counter.inc_by(2, Some(())); + + let families = NamedMetric::new("empty_exemplar", "", counter).encode(); + assert!(encode_to_text(&families).contains("empty_exemplar 2.0 # {} 2.0\n")); + } + + #[test] + fn accepts_legacy_sequence_label_sets() { + let counter = CounterWithExemplar::>::default(); + counter.inc_by(1, Some(vec![("trace_id", "legacy")])); + + let families = counter.encode_metric_value(); + let exemplar = families[0].metric[0] + .counter + .as_ref() + .unwrap() + .exemplar + .as_ref() + .unwrap(); + assert_eq!(trace_id(exemplar), Some("legacy")); + } + + #[test] + fn updates_do_not_require_serializable_label_sets() { + struct OpaqueLabels; + + let counter = CounterWithExemplar::::default(); + counter.inc_by(1, Some(OpaqueLabels)); + assert_eq!(counter.get().0, 1); + + let classic = HistogramWithExemplars::new([1.0]); + classic.observe(0.5, Some(OpaqueLabels)); + + let native = NativeHistogramWithExemplars::new(1.1); + native.observe(0.5, Some(OpaqueLabels)); + } + + #[test] + fn families_keep_series_and_exemplar_labels_separate() { + #[derive(Clone, Debug, Eq, Hash, PartialEq, Serialize)] + struct SeriesLabels { + method: &'static str, + } + + let family = Family::>::default(); + family + .get_or_create(&SeriesLabels { method: "GET" }) + .inc_by(1, Some(TraceLabels { trace_id: "abc" })); + + let families = family.encode_metric_value(); + let metric = &families[0].metric[0]; + assert!(metric.label.iter().any(|label| { + label.name.as_deref() == Some("method") && label.value.as_deref() == Some("GET") + })); + assert_eq!( + trace_id(metric.counter.as_ref().unwrap().exemplar.as_ref().unwrap()), + Some("abc") + ); + } + + #[test] + fn histogram_builders_construct_exemplar_metrics() { + let classic: HistogramWithExemplars = HistogramBuilder { + buckets: &[0.5, 1.0], + } + .new_metric(); + classic.observe( + 0.75, + Some(TraceLabels { + trace_id: "classic", + }), + ); + assert!( + classic.encode_metric_value()[0].metric[0] + .histogram + .as_ref() + .unwrap() + .bucket[1] + .exemplar + .is_some() + ); + + let native: NativeHistogramWithExemplars = NativeHistogramBuilder::new(1.1) + .with_max_buckets(160) + .new_metric(); + native.observe(0.75, Some(TraceLabels { trace_id: "native" })); + assert_eq!( + native.encode_metric_value()[0].metric[0] + .histogram + .as_ref() + .unwrap() + .exemplars + .len(), + 1 + ); + } + + #[test] + fn registered_exemplar_counter_is_collected_and_encoded() { + let counter = CounterWithExemplar::::default(); + counter.inc_by( + 4, + Some(TraceLabels { + trace_id: "registered", + }), + ); + register( + Box::new(NamedMetric::new( + "registered_counter_with_exemplar", + "A registered exemplar counter.", + counter, + )) as Box, + RegistrationMetadata::default(), + ); + + let families = collect(CollectionOptions { + include_optional: false, + service_name: None, + service_name_format: ServiceNameFormat::MetricPrefix, + }); + let family = families + .iter() + .find(|family| family.name.as_deref() == Some("registered_counter_with_exemplar")) + .expect("registered exemplar counter is collected"); + assert_eq!(family.r#type, Some(MetricType::Counter as i32)); + assert_eq!( + trace_id( + family.metric[0] + .counter + .as_ref() + .unwrap() + .exemplar + .as_ref() + .unwrap() + ), + Some("registered") + ); + + let text = encode_to_text(std::slice::from_ref(family)); + assert!( + text.contains("registered_counter_with_exemplar 4.0 # {trace_id=\"registered\"} 4.0\n") + ); + } + + #[test] + fn registered_native_exemplar_family_round_trips_through_protobuf() { + #[derive(Clone, Debug, Eq, Hash, PartialEq, Serialize)] + struct SeriesLabels { + method: &'static str, + } + + let family = Family::< + SeriesLabels, + NativeHistogramWithExemplars, + NativeHistogramBuilder, + >::new_with_constructor(NativeHistogramBuilder::new(1.1)); + family + .get_or_create(&SeriesLabels { method: "GET" }) + .observe(0.25, Some(TraceLabels { trace_id: "native" })); + register( + Box::new(NamedMetric::new( + "registered_native_histogram_with_exemplars", + "A registered native histogram with exemplars.", + family, + )) as Box, + RegistrationMetadata::default(), + ); + + let families = collect(CollectionOptions { + include_optional: false, + service_name: None, + service_name_format: ServiceNameFormat::MetricPrefix, + }); + let family = families + .iter() + .find(|family| { + family.name.as_deref() == Some("registered_native_histogram_with_exemplars") + }) + .expect("registered native histogram family is collected"); + let payload = encode_to_protobuf(std::slice::from_ref(family)); + let mut bytes = payload.as_slice(); + let decoded = MetricFamily::decode_length_delimited(&mut bytes).unwrap(); + let metric = &decoded.metric[0]; + assert!(metric.label.iter().any(|label| { + label.name.as_deref() == Some("method") && label.value.as_deref() == Some("GET") + })); + let exemplars = &metric.histogram.as_ref().unwrap().exemplars; + assert_eq!(exemplars.len(), 1); + assert_eq!(trace_id(&exemplars[0]), Some("native")); + assert!(exemplars[0].timestamp.is_some()); + assert!(bytes.is_empty()); + } +} diff --git a/foundations-metrics/src/metrics/histogram.rs b/foundations-metrics/src/metrics/histogram.rs index e58d71e9..0de302f0 100644 --- a/foundations-metrics/src/metrics/histogram.rs +++ b/foundations-metrics/src/metrics/histogram.rs @@ -66,21 +66,27 @@ impl Histogram { /// Records an observed value. pub fn observe(&self, value: f64) { + self.observe_and_bucket(value); + } + + pub(super) fn observe_and_bucket(&self, value: f64) -> Option { let mut state = self.state.lock(); state.sum += value; state.count = state.count.wrapping_add(1); - let bucket = if value.is_nan() { - None - } else { - let index = state - .buckets - .partition_point(|(upper_bound, _)| *upper_bound < value); - state.buckets.get_mut(index) - }; + if value.is_nan() { + return None; + } - if let Some((_, count)) = bucket { + let index = state + .buckets + .partition_point(|(upper_bound, _)| *upper_bound < value); + + if let Some((_, count)) = state.buckets.get_mut(index) { *count = count.wrapping_add(1); + Some(index) + } else { + None } } @@ -102,7 +108,7 @@ impl EncodeMetricValue for Histogram { } } -fn encode_snapshot(snapshot: HistogramSnapshot) -> Vec { +pub(super) fn encode_snapshot(snapshot: HistogramSnapshot) -> Vec { let mut cumulative_count = 0_u64; let buckets = snapshot .buckets diff --git a/foundations-metrics/src/metrics/mod.rs b/foundations-metrics/src/metrics/mod.rs index d7eda4d8..72a6176b 100644 --- a/foundations-metrics/src/metrics/mod.rs +++ b/foundations-metrics/src/metrics/mod.rs @@ -35,12 +35,16 @@ use std::sync::atomic::{AtomicU64, Ordering}; mod counter; +mod exemplar; mod family; mod gauge; mod histogram; mod native_histogram; pub use counter::{Counter, CounterAtomic}; +pub use exemplar::{ + CounterWithExemplar, Exemplar, HistogramWithExemplars, NativeHistogramWithExemplars, +}; pub use family::{Family, FamilyMetricGuard, MetricConstructor}; pub use gauge::{Gauge, GaugeAtomic, GaugeGuard, RangeGauge}; pub use histogram::{ diff --git a/foundations-metrics/src/metrics/native_histogram.rs b/foundations-metrics/src/metrics/native_histogram.rs index 812be320..3efee69a 100644 --- a/foundations-metrics/src/metrics/native_histogram.rs +++ b/foundations-metrics/src/metrics/native_histogram.rs @@ -56,6 +56,16 @@ impl NativeHistogram { pub fn observe(&self, value: f64) { self.inner.observe(value); } + + pub(super) fn try_encode_metric_value(&self) -> Result, std::fmt::Error> { + // prometheus_client keeps native bucket state private. A cloned histogram shares + // storage, so a temporary registry can drive its protobuf encoder. + let mut registry = Registry::default(); + registry.register("native_histogram", "", self.inner.clone()); + + prometheus_protobuf::encode(®istry) + .map(|families| families.into_iter().map(convert_native_family).collect()) + } } /// Constructs [`NativeHistogram`]s with a fixed configuration. @@ -164,13 +174,8 @@ impl MetricConstructor for NativeHistogramBuilder { impl EncodeMetricValue for NativeHistogram { fn encode_metric_value(&self) -> Vec { - // Upstream keeps native bucket state private. A cloned histogram shares - // storage, so a temporary registry can drive its protobuf encoder. - let mut registry = Registry::default(); - registry.register("native_histogram", "", self.inner.clone()); - - match prometheus_protobuf::encode(®istry) { - Ok(families) => families.into_iter().map(convert_native_family).collect(), + match self.try_encode_metric_value() { + Ok(families) => families, Err(error) => { report_collect_error(format_args!( "non-fatal error while collecting metrics: skipped a native histogram; protobuf encoding failed: {error}" @@ -325,7 +330,7 @@ mod tests { #[test] fn builder_applies_configuration() { - let histogram = NativeHistogramBuilder::new(1.5) + let histogram: NativeHistogram = NativeHistogramBuilder::new(1.5) .with_zero_threshold(0.001) .with_max_buckets(160) .new_metric(); diff --git a/foundations-metrics/src/validation.rs b/foundations-metrics/src/validation.rs index a3c43dd4..243912a8 100644 --- a/foundations-metrics/src/validation.rs +++ b/foundations-metrics/src/validation.rs @@ -5,6 +5,8 @@ use foundations_metrics_registry::proto::{Exemplar, LabelPair, Metric, MetricFam use crate::diagnostics::report_collect_error; pub(crate) const NAME_REQUIREMENT: &str = "a non-empty UTF-8 string without NUL bytes"; +pub(crate) const EXEMPLAR_SERIALIZATION_ERROR_LABEL: &str = + "\0foundations_metrics_exemplar_serialization_error"; #[derive(Clone, Copy)] pub(crate) enum ValidationContext { @@ -115,6 +117,11 @@ enum LabelIssue<'a> { Reserved(&'a str), } +enum ExemplarIssue<'a> { + Label(LabelIssue<'a>), + Serialization(&'a str), +} + fn find_label_issue<'a>( labels: &'a [LabelPair], reserved_label: Option<&str>, @@ -166,12 +173,16 @@ fn exemplars_are_valid(metric: &Metric) -> bool { .bucket .iter() .all(|bucket| bucket.exemplar.as_ref().is_none_or(exemplar_is_valid)) - && histogram.exemplars.iter().all(exemplar_is_valid) + && histogram.exemplars.iter().all(native_exemplar_is_valid) }) } fn exemplar_is_valid(exemplar: &Exemplar) -> bool { - find_label_issue(&exemplar.label, None).is_none() + find_exemplar_issue(&exemplar.label).is_none() +} + +fn native_exemplar_is_valid(exemplar: &Exemplar) -> bool { + exemplar.timestamp.is_some() && exemplar_is_valid(exemplar) } fn sanitize_exemplars(metric: &mut Metric, family_name: &str, context: ValidationContext) { @@ -189,9 +200,15 @@ fn sanitize_exemplars(metric: &mut Metric, family_name: &str, context: Validatio ); } histogram.exemplars.retain(|exemplar| { - if let Some(issue) = find_label_issue(&exemplar.label, None) { + if let Some(issue) = find_exemplar_issue(&exemplar.label) { report_exemplar_drop(context, family_name, "native histogram", issue); false + } else if exemplar.timestamp.is_none() { + report_collect_error(format_args!( + "non-fatal error while {}: dropped native histogram exemplar in metric family {family_name:?} without a required timestamp", + context.action() + )); + false } else { true } @@ -207,7 +224,7 @@ fn sanitize_exemplar_slot( ) { let Some(issue) = exemplar .as_ref() - .and_then(|exemplar| find_label_issue(&exemplar.label, None)) + .and_then(|exemplar| find_exemplar_issue(&exemplar.label)) else { return; }; @@ -220,21 +237,38 @@ fn report_exemplar_drop( context: ValidationContext, family_name: &str, kind: &str, - issue: LabelIssue<'_>, + issue: ExemplarIssue<'_>, ) { match issue { - LabelIssue::Invalid(name) => report_collect_error(format_args!( + ExemplarIssue::Label(LabelIssue::Invalid(name)) => report_collect_error(format_args!( "non-fatal error while {}: dropped {kind} exemplar in metric family {family_name:?} with invalid label name {name:?}; expected {NAME_REQUIREMENT}", context.action() )), - LabelIssue::Duplicate(name) => report_collect_error(format_args!( + ExemplarIssue::Label(LabelIssue::Duplicate(name)) => report_collect_error(format_args!( "non-fatal error while {}: dropped {kind} exemplar in metric family {family_name:?} with duplicate label name {name:?}", context.action() )), - LabelIssue::Reserved(_) => unreachable!("exemplar labels have no reserved names"), + ExemplarIssue::Serialization(error) => report_collect_error(format_args!( + "non-fatal error while {}: dropped {kind} exemplar in metric family {family_name:?}; label serialization failed: {error}", + context.action() + )), + ExemplarIssue::Label(LabelIssue::Reserved(_)) => { + unreachable!("exemplar labels have no reserved names") + } } } +fn find_exemplar_issue(labels: &[LabelPair]) -> Option> { + if labels.len() == 1 + && labels[0].name.as_deref() == Some(EXEMPLAR_SERIALIZATION_ERROR_LABEL) + && let Some(error) = labels[0].value.as_deref() + { + return Some(ExemplarIssue::Serialization(error)); + } + + find_label_issue(labels, None).map(ExemplarIssue::Label) +} + #[cfg(test)] mod tests { use super::*; @@ -260,4 +294,17 @@ mod tests { assert!(!is_valid_name(name)); } } + + #[test] + fn preserves_exemplar_serialization_errors_for_diagnostics() { + let labels = [LabelPair { + name: Some(EXEMPLAR_SERIALIZATION_ERROR_LABEL.to_owned()), + value: Some("root cause".to_owned()), + }]; + + assert!(matches!( + find_exemplar_issue(&labels), + Some(ExemplarIssue::Serialization("root cause")) + )); + } }