diff --git a/foundations-metrics-registry/src/proto/mod.rs b/foundations-metrics-registry/src/proto/mod.rs index 6a56004f..50f85e21 100644 --- a/foundations-metrics-registry/src/proto/mod.rs +++ b/foundations-metrics-registry/src/proto/mod.rs @@ -10,7 +10,7 @@ mod model; pub use model::{ - BucketSpan, Counter, Gauge, Histogram, LabelPair, Metric, MetricFamily, MetricType, + Bucket, BucketSpan, Counter, Gauge, Histogram, LabelPair, Metric, MetricFamily, MetricType, }; #[cfg(test)] diff --git a/foundations-metrics/Cargo.toml b/foundations-metrics/Cargo.toml index 34a36d91..48090317 100644 --- a/foundations-metrics/Cargo.toml +++ b/foundations-metrics/Cargo.toml @@ -8,7 +8,10 @@ license.workspace = true [dependencies] foundations-metrics-registry = { workspace = true } -prometheus-client = "0.25.0" +prometheus-client = { version = "0.25.0", features = [ + "prometheus_protobuf", + "protobuf-protox", +] } serde = { workspace = true, features = ["derive"] } ryu = "1.0.23" parking_lot = { workspace = true } diff --git a/foundations-metrics/src/lib.rs b/foundations-metrics/src/lib.rs index 14215e42..9e2b2e73 100644 --- a/foundations-metrics/src/lib.rs +++ b/foundations-metrics/src/lib.rs @@ -18,7 +18,8 @@ pub use foundations_metrics_registry::{ }; pub use labels::{LabelError, to_label_pairs}; pub use metrics::{ - Counter, CounterAtomic, Family, FamilyMetricGuard, Gauge, GaugeAtomic, GaugeGuard, - MetricConstructor, RangeGauge, + Counter, CounterAtomic, Family, FamilyMetricGuard, Gauge, GaugeAtomic, GaugeGuard, Histogram, + HistogramBuilder, HistogramSnapshot, HistogramTimer, MetricConstructor, NativeHistogram, + NativeHistogramBuilder, RangeGauge, TimeHistogram, }; pub use registered::NamedMetric; diff --git a/foundations-metrics/src/metrics/histogram.rs b/foundations-metrics/src/metrics/histogram.rs new file mode 100644 index 00000000..e58d71e9 --- /dev/null +++ b/foundations-metrics/src/metrics/histogram.rs @@ -0,0 +1,608 @@ +use std::iter::once; +use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::time::{Duration, Instant}; + +use foundations_metrics_registry::proto::{self, Bucket, MetricType}; +use parking_lot::Mutex; + +use crate::{MetricFamily, value::EncodeMetricValue}; + +use super::MetricConstructor; + +/// A classic fixed-bucket histogram. +/// +/// Each observation contributes to the sum and count and increments the first +/// bucket whose inclusive upper bound contains it. Clones share the same +/// storage. +/// +/// # Examples +/// +/// ``` +/// use foundations_metrics::Histogram; +/// +/// let response_size = Histogram::new([100.0, 1_000.0, 10_000.0]); +/// response_size.observe(750.0); +/// response_size.observe(2_500.0); +/// ``` +#[derive(Clone, Debug)] +pub struct Histogram { + state: Arc>, +} + +#[derive(Debug)] +struct HistogramState { + sum: f64, + count: u64, + buckets: Vec<(f64, u64)>, +} + +/// A point-in-time view of a histogram's sum, count, and buckets. +#[derive(Debug, PartialEq)] +pub struct HistogramSnapshot { + sum: f64, + count: u64, + buckets: Vec<(f64, u64)>, +} + +impl Histogram { + /// Creates a histogram with the provided inclusive upper bounds. + /// + /// Bounds may be given in any order; they are sorted ascending. A terminal + /// `f64::MAX` bucket is appended automatically. + pub fn new(bounds: impl IntoIterator) -> Self { + let mut buckets: Vec<_> = bounds.into_iter().map(|bound| (bound, 0)).collect(); + buckets.push((f64::MAX, 0)); + buckets.sort_by(|(a, _), (b, _)| a.total_cmp(b)); + + Self { + state: Arc::new(Mutex::new(HistogramState { + sum: 0.0, + count: 0, + buckets, + })), + } + } + + /// Records an observed value. + pub fn observe(&self, value: f64) { + 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 let Some((_, count)) = bucket { + *count = count.wrapping_add(1); + } + } + + /// Returns a point-in-time snapshot of this histogram's sum, count, and + /// buckets. + pub fn snapshot(&self) -> HistogramSnapshot { + let state = self.state.lock(); + HistogramSnapshot { + sum: state.sum, + count: state.count, + buckets: state.buckets.clone(), + } + } +} + +impl EncodeMetricValue for Histogram { + fn encode_metric_value(&self) -> Vec { + encode_snapshot(self.snapshot()) + } +} + +fn encode_snapshot(snapshot: HistogramSnapshot) -> Vec { + let mut cumulative_count = 0_u64; + let buckets = snapshot + .buckets + .into_iter() + .map(|(upper_bound, count)| { + cumulative_count = cumulative_count.wrapping_add(count); + Bucket { + cumulative_count: Some(cumulative_count), + upper_bound: Some(upper_bound), + ..Default::default() + } + }) + .collect(); + + vec![MetricFamily { + name: Some(String::new()), + help: None, + r#type: Some(MetricType::Histogram as i32), + metric: vec![proto::Metric { + histogram: Some(proto::Histogram { + sample_count: Some(snapshot.count), + sample_sum: Some(snapshot.sum), + bucket: buckets, + ..Default::default() + }), + ..Default::default() + }], + unit: None, + }] +} + +/// Constructs classic and time histograms with a fixed set of buckets. +#[derive(Clone, Debug)] +pub struct HistogramBuilder { + /// Inclusive upper bounds for the histogram buckets. + pub buckets: &'static [f64], +} + +impl MetricConstructor for HistogramBuilder { + fn new_metric(&self) -> Histogram { + Histogram::new(self.buckets.iter().copied()) + } +} + +impl MetricConstructor for HistogramBuilder { + fn new_metric(&self) -> TimeHistogram { + TimeHistogram::new(self.buckets.iter().copied()) + } +} + +/// A faster, lock-free histogram for tracking time. +// Adapted from prometools' `histogram::TimeHistogram` +// (https://github.com/nox/prometools, licensed MIT OR Apache-2.0). +#[derive(Debug)] +pub struct TimeHistogram { + state: Arc, +} + +/// Timer to measure and record the duration of an event. +/// +/// This timer can be stopped and observed at most once, either automatically +/// (when it goes out of scope) or manually. Alternatively, it can be manually +/// stopped and discarded in order to not record its value. +// Adapted from prometools' `histogram::HistogramTimer` +// (https://github.com/nox/prometools, licensed MIT OR Apache-2.0). +#[must_use = "HistogramTimer measures on Drop so should be assigned to named variable"] +pub struct HistogramTimer { + histogram: TimeHistogram, + observed: bool, + start: Option, + accumulated: Duration, +} + +#[derive(Debug)] +struct TimeHistogramState { + sum: AtomicU64, + count: AtomicU64, + buckets: Vec<(f64, AtomicU64)>, +} + +impl HistogramTimer { + /// Pauses elapsed-time tracking. + /// + /// Calling this while the timer is already paused has no effect. + pub fn pause(&mut self) { + self.accumulated += self.start.map_or(Duration::ZERO, |value| { + Instant::now().saturating_duration_since(value) + }); + self.start = None + } + + /// Resumes elapsed-time tracking. + /// + /// Calling this while the timer is already running has no effect. + pub fn resume(&mut self) { + if self.start.is_none() { + self.start = Some(Instant::now()); + } + } + + /// Stops the timer, records its duration, and returns the duration. + pub fn stop_and_record(self) -> Duration { + let mut timer = self; + timer.observe(true) + } + + /// Stops the timer without recording and returns its duration. + pub fn stop_and_discard(self) -> Duration { + let mut timer = self; + timer.observe(false) + } + + fn observe(&mut self, record: bool) -> Duration { + let elapsed_since_start: Duration = self.start.map_or(Duration::ZERO, |value| { + Instant::now().saturating_duration_since(value) + }); + let elapsed = elapsed_since_start + self.accumulated; + + self.observed = true; + if record { + self.histogram.observe(elapsed.as_nanos() as u64); + } + + elapsed + } +} + +impl Drop for HistogramTimer { + fn drop(&mut self) { + if !self.observed { + self.observe(true); + } + } +} + +impl Clone for TimeHistogram { + fn clone(&self) -> Self { + TimeHistogram { + state: Arc::clone(&self.state), + } + } +} + +impl TimeHistogram { + /// Creates a time histogram with inclusive bucket bounds in seconds. + /// + /// Bounds may be given in any order; they are sorted ascending. A terminal + /// `f64::MAX` bucket is appended automatically. + pub fn new(buckets: impl IntoIterator) -> Self { + let mut buckets: Vec<_> = buckets + .into_iter() + .chain(once(f64::MAX)) + .map(|upper_bound| (upper_bound, AtomicU64::new(0))) + .collect(); + buckets.sort_by(|(a, _), (b, _)| a.total_cmp(b)); + + Self { + state: Arc::new(TimeHistogramState { + sum: Default::default(), + count: Default::default(), + buckets, + }), + } + } + + /// Starts a timer that records its duration when stopped or dropped. + pub fn start_timer(&self) -> HistogramTimer { + HistogramTimer { + histogram: self.clone(), + observed: false, + start: Some(Instant::now()), + accumulated: Duration::new(0, 0), + } + } + + /// Records an observed duration in nanoseconds. + pub fn observe(&self, nanos: u64) { + self.observe_and_bucket(nanos); + } + + fn observe_and_bucket(&self, v: u64) -> Option { + self.state.sum.fetch_add(v, Ordering::Relaxed); + self.state.count.fetch_add(1, Ordering::Relaxed); + + let first_bucket = self + .state + .buckets + .iter() + .enumerate() + .find(|(_i, (upper_bound, _value))| upper_bound >= &(seconds(v))); + + match first_bucket { + Some((i, (_upper_bound, value))) => { + value.fetch_add(1, Ordering::Relaxed); + Some(i) + } + None => None, + } + } + + /// Returns a snapshot whose sum and bucket bounds are expressed in seconds. + pub fn snapshot(&self) -> HistogramSnapshot { + let sum = seconds(self.state.sum.load(Ordering::Relaxed)); + let count = self.state.count.load(Ordering::Relaxed); + let buckets = self + .state + .buckets + .iter() + .map(|(k, v)| (*k, v.load(Ordering::Relaxed))) + .collect(); + + HistogramSnapshot { + sum, + count, + buckets, + } + } +} + +impl HistogramSnapshot { + /// Returns the sum of all observations. + pub fn sum(&self) -> f64 { + self.sum + } + + /// Returns the number of observations. + pub fn count(&self) -> u64 { + self.count + } + + /// Returns each inclusive upper bound and its non-cumulative count. + pub fn buckets(&self) -> &[(f64, u64)] { + &self.buckets + } +} + +// Adapted from prometools' private `histogram::seconds` +// (https://github.com/nox/prometools, licensed MIT OR Apache-2.0). +#[inline(always)] +fn seconds(val: u64) -> f64 { + (val as f64) * 1E-9 +} + +impl EncodeMetricValue for TimeHistogram { + fn encode_metric_value(&self) -> Vec { + encode_snapshot(self.snapshot()) + } +} + +#[cfg(test)] +mod tests { + use serde::Serialize; + + use super::*; + use crate::Family; + + #[test] + fn appends_terminal_bucket_and_clones_share_storage() { + let histogram = Histogram::new([1.0, 2.0]); + let clone = histogram.clone(); + + clone.observe(0.5); + clone.observe(1.0); + histogram.observe(1.5); + histogram.observe(3.0); + + assert_eq!( + histogram.snapshot(), + HistogramSnapshot { + sum: 6.0, + count: 4, + buckets: vec![(1.0, 2), (2.0, 1), (f64::MAX, 1)], + } + ); + } + + #[test] + fn excludes_nan_and_sorts_unsorted_bounds() { + let nan = Histogram::new([1.0]); + nan.observe(f64::NAN); + let snapshot = nan.snapshot(); + assert!(snapshot.sum.is_nan()); + assert_eq!(snapshot.count, 1); + assert!(snapshot.buckets.iter().all(|(_, count)| *count == 0)); + + // Bounds are sorted ascending regardless of the order they were given in, + // so `0.5` lands in the `1.0` bucket rather than the leading `10.0` one. + let unsorted = Histogram::new([10.0, 1.0]); + unsorted.observe(0.5); + let snapshot = unsorted.snapshot(); + assert_eq!(snapshot.buckets, vec![(1.0, 1), (10.0, 0), (f64::MAX, 0)],); + } + + #[test] + fn encodes_one_protobuf_histogram_with_cumulative_buckets() { + let histogram = Histogram::new([1.0, 2.0]); + histogram.observe(0.5); + histogram.observe(1.5); + histogram.observe(3.0); + + let families = histogram.encode_metric_value(); + assert_eq!(families.len(), 1); + assert_eq!(families[0].name.as_deref(), Some("")); + assert_eq!(families[0].help, None); + assert_eq!(families[0].r#type, Some(MetricType::Histogram as i32)); + assert_eq!(families[0].unit, None); + + let encoded = families[0].metric[0] + .histogram + .as_ref() + .expect("encoded histogram is present"); + assert_eq!(encoded.sample_count, Some(3)); + assert_eq!(encoded.sample_sum, Some(5.0)); + assert_eq!( + encoded + .bucket + .iter() + .map(|bucket| (bucket.upper_bound, bucket.cumulative_count)) + .collect::>(), + vec![ + (Some(1.0), Some(1)), + (Some(2.0), Some(2)), + (Some(f64::MAX), Some(3)), + ] + ); + } + + #[test] + fn family_adds_labels_to_histogram_rows() { + #[derive(Clone, Eq, Hash, PartialEq, Serialize)] + struct Labels { + method: &'static str, + } + + let family = + Family::::new_with_constructor(HistogramBuilder { + buckets: &[0.1, 1.0], + }); + family.get_or_create(&Labels { method: "GET" }).observe(0.5); + family + .get_or_create(&Labels { method: "POST" }) + .observe(2.0); + + let families = family.encode_metric_value(); + assert_eq!(families.len(), 1); + assert_eq!(families[0].metric.len(), 2); + assert!(families[0].metric.iter().all(|metric| { + metric.histogram.is_some() + && metric + .label + .iter() + .any(|label| label.name.as_deref() == Some("method")) + })); + } + + #[test] + fn time_histogram_encodes_seconds_and_cumulative_buckets() { + let histogram = TimeHistogram::new([1.0, 2.0]); + histogram.observe(500_000_000); + histogram.observe(1_500_000_000); + histogram.observe(3_000_000_000); + + let families = histogram.encode_metric_value(); + assert_eq!(families.len(), 1); + assert_eq!(families[0].name.as_deref(), Some("")); + assert_eq!(families[0].r#type, Some(MetricType::Histogram as i32)); + + let encoded = families[0].metric[0] + .histogram + .as_ref() + .expect("encoded time histogram is present"); + assert_eq!(encoded.sample_count, Some(3)); + assert_eq!(encoded.sample_sum, Some(5.0)); + assert_eq!( + encoded + .bucket + .iter() + .map(|bucket| bucket.cumulative_count) + .collect::>(), + vec![Some(1), Some(2), Some(3)] + ); + } + + #[test] + fn time_histogram_sorts_unsorted_buckets() { + let histogram = TimeHistogram::new([2.0, 1.0]); + histogram.observe(500_000_000); + + assert_eq!( + histogram.snapshot().buckets(), + &[(1.0, 1), (2.0, 0), (f64::MAX, 0)] + ); + } + + #[test] + fn time_histogram_tracks_seconds_and_clones_share_storage() { + let histogram = TimeHistogram::new([1.0, 2.0, 4.0, 8.0, 16.0]); + let clone = histogram.clone(); + + for nanos in [ + 1_000_000_000, + 1_500_000_000, + 2_500_000_000, + 8_500_000_000, + 500_000_000, + ] { + clone.observe(nanos); + } + + let snapshot = histogram.snapshot(); + assert_eq!(snapshot.sum(), 14.0); + assert_eq!(snapshot.count(), 5); + assert_eq!( + snapshot.buckets(), + &[ + (1.0, 2), + (2.0, 1), + (4.0, 1), + (8.0, 0), + (16.0, 1), + (f64::MAX, 0), + ] + ); + } + + #[test] + fn histogram_builder_constructs_time_histograms() { + let histogram: TimeHistogram = HistogramBuilder { + buckets: &[0.5, 1.0], + } + .new_metric(); + + histogram.observe(750_000_000); + assert_eq!(histogram.snapshot().buckets()[1], (1.0, 1)); + } + + #[test] + fn timer_records_once_or_discards() { + let recorded = TimeHistogram::new([1.0]); + let _duration = recorded.start_timer().stop_and_record(); + assert_eq!(recorded.snapshot().count(), 1); + + let discarded = TimeHistogram::new([1.0]); + discarded.start_timer().stop_and_discard(); + assert_eq!(discarded.snapshot().count(), 0); + + let dropped = TimeHistogram::new([1.0]); + drop(dropped.start_timer()); + assert_eq!(dropped.snapshot().count(), 1); + } + + #[test] + fn timer_pause_and_resume_are_idempotent() { + let histogram = TimeHistogram::new([1.0]); + let mut timer = histogram.start_timer(); + + timer.pause(); + let paused_duration = timer.accumulated; + assert!(timer.start.is_none()); + timer.pause(); + assert_eq!(timer.accumulated, paused_duration); + + timer.resume(); + let resumed_at = timer.start; + assert!(resumed_at.is_some()); + timer.resume(); + assert_eq!(timer.start, resumed_at); + + timer.stop_and_discard(); + assert_eq!(histogram.snapshot().count(), 0); + } + + #[test] + fn family_adds_labels_to_time_histogram_rows() { + #[derive(Clone, Eq, Hash, PartialEq, Serialize)] + struct Labels { + operation: &'static str, + } + + let family = Family::::new_with_constructor( + HistogramBuilder { + buckets: &[0.1, 1.0], + }, + ); + family + .get_or_create(&Labels { operation: "read" }) + .observe(500_000_000); + family + .get_or_create(&Labels { operation: "write" }) + .observe(1_500_000_000); + + let families = family.encode_metric_value(); + assert_eq!(families.len(), 1); + assert_eq!(families[0].metric.len(), 2); + assert!(families[0].metric.iter().all(|metric| { + metric.histogram.is_some() + && metric + .label + .iter() + .any(|label| label.name.as_deref() == Some("operation")) + })); + } +} diff --git a/foundations-metrics/src/metrics/mod.rs b/foundations-metrics/src/metrics/mod.rs index 54ce79bd..d7eda4d8 100644 --- a/foundations-metrics/src/metrics/mod.rs +++ b/foundations-metrics/src/metrics/mod.rs @@ -37,10 +37,16 @@ use std::sync::atomic::{AtomicU64, Ordering}; mod counter; mod family; mod gauge; +mod histogram; +mod native_histogram; pub use counter::{Counter, CounterAtomic}; pub use family::{Family, FamilyMetricGuard, MetricConstructor}; pub use gauge::{Gauge, GaugeAtomic, GaugeGuard, RangeGauge}; +pub use histogram::{ + Histogram, HistogramBuilder, HistogramSnapshot, HistogramTimer, TimeHistogram, +}; +pub use native_histogram::{NativeHistogram, NativeHistogramBuilder}; fn update_f64(atomic: &AtomicU64, f: impl Fn(f64) -> f64) -> f64 { let mut old_bits = atomic.load(Ordering::Relaxed); diff --git a/foundations-metrics/src/metrics/native_histogram.rs b/foundations-metrics/src/metrics/native_histogram.rs new file mode 100644 index 00000000..812be320 --- /dev/null +++ b/foundations-metrics/src/metrics/native_histogram.rs @@ -0,0 +1,418 @@ +use foundations_metrics_registry::proto::{self, Bucket, BucketSpan, LabelPair, MetricType}; +use prometheus_client::encoding::prometheus_protobuf::{ + self, prometheus_data_model as prometheus_proto, +}; +use prometheus_client::metrics::histogram::{ + Histogram as PrometheusHistogram, NativeHistogramConfig, +}; +use prometheus_client::registry::Registry; + +use crate::diagnostics::report_collect_error; +use crate::{MetricFamily, value::EncodeMetricValue}; + +use super::MetricConstructor; + +/// A native (exponential-bucket) histogram. +/// +/// Unlike a classic [`Histogram`](super::Histogram), whose buckets are a fixed +/// list of upper bounds, a native histogram places observations into +/// exponentially sized buckets whose resolution is chosen by a growth factor. +/// Native histograms require the Prometheus protobuf exposition format. +/// +/// Clones share the same storage. +/// +/// # Examples +/// +/// ``` +/// use foundations_metrics::NativeHistogram; +/// +/// let request_latency = NativeHistogram::new(1.1); +/// request_latency.observe(0.25); +/// request_latency.observe(4.2); +/// ``` +#[derive(Clone, Debug)] +pub struct NativeHistogram { + inner: PrometheusHistogram, +} + +impl NativeHistogram { + /// Creates a native histogram with the given bucket growth `factor`. + /// + /// The factor bounds the ratio between adjacent bucket boundaries; a + /// smaller factor gives finer resolution. The zero bucket uses the + /// Prometheus-recommended default threshold and the number of buckets is + /// unbounded. Use [`NativeHistogramBuilder`] for full control. + /// + /// # Panics + /// + /// Panics if `factor` is not greater than `1.0`. + #[track_caller] + pub fn new(factor: f64) -> Self { + NativeHistogramBuilder::new(factor).new_metric() + } + + /// Records an observed value. + #[inline] + pub fn observe(&self, value: f64) { + self.inner.observe(value); + } +} + +/// Constructs [`NativeHistogram`]s with a fixed configuration. +/// +/// Use this with [`Family`](crate::Family) or a metric's `#[ctor = ...]` when a +/// native histogram needs bucket configuration at creation time. +/// +/// # Examples +/// +/// ``` +/// use foundations_metrics::{Family, NativeHistogram, NativeHistogramBuilder}; +/// use serde::Serialize; +/// +/// #[derive(Clone, Eq, Hash, PartialEq, Serialize)] +/// struct Labels { +/// method: &'static str, +/// } +/// +/// let builder = NativeHistogramBuilder::new(1.1).with_max_buckets(160); +/// let latencies = Family::::new_with_constructor(builder); +/// latencies.get_or_create(&Labels { method: "GET" }).observe(0.5); +/// ``` +#[derive(Clone, Copy, Debug)] +pub struct NativeHistogramBuilder { + /// Bucket growth factor; must be greater than `1.0`. Smaller factors give + /// finer resolution. + pub bucket_factor: f64, + + /// Width of the zero bucket, which absorbs observations close to zero. + /// + /// `0.0` keeps the Prometheus-recommended default threshold; a negative + /// value configures a zero-width zero bucket. + pub zero_threshold: f64, + + /// Best-effort upper bound on the number of populated buckets across both + /// the positive and negative ranges. + /// + /// `0` leaves the bucket count unbounded. + pub max_buckets: usize, +} + +impl NativeHistogramBuilder { + /// Creates a builder with the given bucket growth `factor`, the default zero + /// threshold, and an unbounded number of buckets. + pub fn new(factor: f64) -> Self { + Self { + bucket_factor: factor, + zero_threshold: 0.0, + max_buckets: 0, + } + } + + /// Sets the zero-bucket threshold. + /// + /// `0.0` keeps the default threshold; a negative value configures a + /// zero-width zero bucket. + pub fn with_zero_threshold(mut self, zero_threshold: f64) -> Self { + self.zero_threshold = zero_threshold; + self + } + + /// Sets a best-effort maximum number of populated buckets. + /// + /// `0` leaves the count unbounded. + pub fn with_max_buckets(mut self, max_buckets: usize) -> Self { + self.max_buckets = max_buckets; + self + } + + /// Translates this builder into the wrapped crate's configuration. + /// + /// # Panics + /// + /// Panics if `bucket_factor` is not greater than `1.0` or if + /// `zero_threshold` is not finite. + #[track_caller] + fn config(&self) -> NativeHistogramConfig { + // Validated here rather than left to `prometheus-client`: its assertions + // are not `#[track_caller]`, so they report a location inside that crate + // instead of the code that supplied the invalid configuration. + assert!( + self.bucket_factor > 1.0, + "native histogram bucket factor must be greater than 1.0, but was {}", + self.bucket_factor + ); + assert!( + self.zero_threshold.is_finite(), + "native histogram zero threshold must be finite, but was {}", + self.zero_threshold + ); + + NativeHistogramConfig::new(self.bucket_factor) + .zero_threshold(self.zero_threshold) + .max_buckets(self.max_buckets) + } +} + +impl MetricConstructor for NativeHistogramBuilder { + #[track_caller] + fn new_metric(&self) -> NativeHistogram { + NativeHistogram { + inner: PrometheusHistogram::new_native(self.config()), + } + } +} + +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(), + Err(error) => { + report_collect_error(format_args!( + "non-fatal error while collecting metrics: skipped a native histogram; protobuf encoding failed: {error}" + )); + Vec::new() + } + } + } +} + +fn convert_native_family(family: prometheus_proto::MetricFamily) -> MetricFamily { + MetricFamily { + name: Some(String::new()), + help: None, + r#type: Some(MetricType::Histogram as i32), + metric: family + .metric + .into_iter() + .map(convert_native_metric) + .collect(), + unit: (!family.unit.is_empty()).then_some(family.unit), + } +} + +fn convert_native_metric(metric: prometheus_proto::Metric) -> proto::Metric { + proto::Metric { + label: metric + .label + .into_iter() + .map(|label| LabelPair { + name: Some(label.name), + value: Some(label.value), + }) + .collect(), + histogram: metric.histogram.map(convert_native_histogram), + timestamp_ms: (metric.timestamp_ms != 0).then_some(metric.timestamp_ms), + ..Default::default() + } +} + +fn convert_native_histogram(histogram: prometheus_proto::Histogram) -> proto::Histogram { + proto::Histogram { + sample_count: Some(histogram.sample_count), + sample_count_float: (histogram.sample_count_float > 0.0) + .then_some(histogram.sample_count_float), + sample_sum: Some(histogram.sample_sum), + bucket: histogram + .bucket + .into_iter() + .map(|bucket| Bucket { + cumulative_count: Some(bucket.cumulative_count), + cumulative_count_float: (bucket.cumulative_count_float > 0.0) + .then_some(bucket.cumulative_count_float), + upper_bound: Some(bucket.upper_bound), + ..Default::default() + }) + .collect(), + created_timestamp: histogram.start_timestamp, + schema: Some(histogram.schema), + zero_threshold: Some(histogram.zero_threshold), + zero_count: Some(histogram.zero_count), + zero_count_float: (histogram.zero_count_float > 0.0).then_some(histogram.zero_count_float), + negative_span: histogram + .negative_span + .into_iter() + .map(convert_native_span) + .collect(), + negative_delta: histogram.negative_delta, + negative_count: histogram.negative_count, + positive_span: histogram + .positive_span + .into_iter() + .map(convert_native_span) + .collect(), + positive_delta: histogram.positive_delta, + positive_count: histogram.positive_count, + ..Default::default() + } +} + +fn convert_native_span(span: prometheus_proto::BucketSpan) -> BucketSpan { + BucketSpan { + offset: Some(span.offset), + length: Some(span.length), + } +} + +#[cfg(test)] +mod tests { + use foundations_metrics_registry::proto::{self, MetricType}; + use serde::Serialize; + + use super::*; + use crate::{EncodeMetric, Family, NamedMetric}; + + fn encoded_histogram(families: &[MetricFamily]) -> &proto::Histogram { + assert_eq!(families.len(), 1); + assert_eq!(families[0].r#type, Some(MetricType::Histogram as i32)); + assert_eq!(families[0].metric.len(), 1); + families[0].metric[0] + .histogram + .as_ref() + .expect("encoded native histogram is present") + } + + #[test] + fn clones_share_storage() { + let histogram = NativeHistogram::new(1.1); + let clone = histogram.clone(); + + histogram.observe(1.0); + clone.observe(3.5); + + let families = histogram.encode_metric_value(); + let encoded = encoded_histogram(&families); + assert_eq!(encoded.sample_count, Some(2)); + assert_eq!(encoded.sample_sum, Some(4.5)); + } + + #[test] + fn encodes_relative_name_and_native_fields() { + let histogram = NativeHistogram::new(2.0); + histogram.observe(1.0); + histogram.observe(2.0); + histogram.observe(4.0); + + let families = histogram.encode_metric_value(); + assert_eq!(families[0].name.as_deref(), Some("")); + assert_eq!(families[0].help, None); + + let encoded = encoded_histogram(&families); + assert_eq!(encoded.sample_count, Some(3)); + assert_eq!(encoded.sample_sum, Some(7.0)); + assert!(encoded.schema.is_some()); + assert!(encoded.zero_threshold.is_some()); + assert_eq!(encoded.zero_count, Some(0)); + assert!(!encoded.positive_span.is_empty()); + assert!(!encoded.positive_delta.is_empty()); + assert!(encoded.negative_span.is_empty()); + assert!(encoded.bucket.is_empty()); + } + + #[test] + fn empty_histogram_encodes_a_valid_family() { + let families = NativeHistogram::new(1.1).encode_metric_value(); + let encoded = encoded_histogram(&families); + + assert_eq!(encoded.sample_count, Some(0)); + assert_eq!(encoded.sample_sum, Some(0.0)); + assert!(encoded.schema.is_some()); + } + + #[test] + fn builder_applies_configuration() { + let histogram = NativeHistogramBuilder::new(1.5) + .with_zero_threshold(0.001) + .with_max_buckets(160) + .new_metric(); + histogram.observe(0.5); + + let families = histogram.encode_metric_value(); + let encoded = encoded_histogram(&families); + assert_eq!(encoded.sample_count, Some(1)); + assert_eq!(encoded.zero_threshold, Some(0.001)); + } + + #[test] + fn named_metric_rewrites_name_and_fills_help() { + let histogram = NativeHistogram::new(1.1); + histogram.observe(1.0); + + let named = NamedMetric::new("request_latency_seconds", "Latency of requests.", histogram); + + let families = named.encode(); + assert_eq!(families.len(), 1); + assert_eq!(families[0].name.as_deref(), Some("request_latency_seconds")); + assert_eq!(families[0].help.as_deref(), Some("Latency of requests.")); + assert_eq!(families[0].r#type, Some(MetricType::Histogram as i32)); + } + + #[test] + fn family_adds_labels_to_histogram_rows() { + #[derive(Clone, Eq, Hash, PartialEq, Serialize)] + struct Labels { + method: &'static str, + } + + let family = + Family::::new_with_constructor( + NativeHistogramBuilder::new(1.1), + ); + family.get_or_create(&Labels { method: "GET" }).observe(0.5); + family + .get_or_create(&Labels { method: "POST" }) + .observe(2.0); + + let families = family.encode_metric_value(); + assert_eq!(families.len(), 1); + assert_eq!(families[0].metric.len(), 2); + assert!(families[0].metric.iter().all(|metric| { + metric.histogram.is_some() + && metric + .label + .iter() + .any(|label| label.name.as_deref() == Some("method")) + })); + } + + // Each `#[track_caller]` between the caller and the assertion is + // load-bearing: dropping any one of them collapses the reported location + // onto that frame instead of the code that supplied the configuration. + #[test] + fn invalid_bucket_factor_panics_at_the_callers_location() { + // Relies on each test running in its own process, as `cargo nextest` + // does, since the panic hook is process-wide. + let previous = std::panic::take_hook(); + let location = std::sync::Arc::new(std::sync::Mutex::new(None)); + let captured = std::sync::Arc::clone(&location); + + std::panic::set_hook(Box::new(move |info| { + *captured.lock().unwrap() = info + .location() + .map(|location| (location.file().to_owned(), location.line())); + })); + + let expected_line = line!() + 1; + let result = std::panic::catch_unwind(|| NativeHistogram::new(1.0)); + + std::panic::set_hook(previous); + + assert!(result.is_err(), "a bucket factor of 1.0 must be rejected"); + + let (file, line) = location + .lock() + .unwrap() + .take() + .expect("the panic location was captured"); + + assert!( + file.ends_with("native_histogram.rs"), + "reported file: {file}" + ); + assert_eq!(line, expected_line); + } +}