diff --git a/src/config/src/config.rs b/src/config/src/config.rs
index 55ee83ce47..2193e56f5a 100644
--- a/src/config/src/config.rs
+++ b/src/config/src/config.rs
@@ -3246,9 +3246,11 @@ pub struct Prometheus {
pub ha_cluster_label: String,
#[env_config(name = "ZO_PROMETHEUS_HA_REPLICA", default = "__replica__")]
pub ha_replica_label: String,
- /// Max `le` labels (buckets + gap markers + inf) a native histogram sample may
- /// expand to; over-limit samples are downscaled (adjacent buckets merged).
- #[env_config(name = "ZO_PROMETHEUS_NATIVE_HISTOGRAM_MAX_BUCKETS", default = 16)]
+ /// Exponential histograms are stored at `min(producer schema, this)`, never count-driven.
+ #[env_config(name = "ZO_METRICS_EXP_HISTOGRAM_TARGET_SCHEMA", default = 1)]
+ pub exp_histogram_target_schema: i32,
+ /// Safety valve, not a layout knob: past this many `le` labels a sample is downscaled.
+ #[env_config(name = "ZO_PROMETHEUS_NATIVE_HISTOGRAM_MAX_BUCKETS", default = 512)]
pub native_histogram_max_buckets: usize,
}
@@ -3797,6 +3799,12 @@ fn check_common_config(cfg: &mut Config) -> Result<(), anyhow::Error> {
if cfg.limit.metrics_max_points_per_series == 0 {
cfg.limit.metrics_max_points_per_series = 30_000;
}
+ if !(-4..=8).contains(&cfg.prom.exp_histogram_target_schema) {
+ return Err(anyhow::anyhow!(
+ "ZO_METRICS_EXP_HISTOGRAM_TARGET_SCHEMA must be within -4..=8, got {}",
+ cfg.prom.exp_histogram_target_schema
+ ));
+ }
// check search job retention
if cfg.limit.search_job_retention == 0 {
@@ -5918,6 +5926,30 @@ mod tests {
assert_eq!(cfg.common.feature_bloom_filter_extra_fields, "trace_id");
}
+ #[test]
+ fn test_check_common_config_exp_histogram_target_schema_range() {
+ let cfg = Config::init().unwrap();
+ assert_eq!(cfg.prom.exp_histogram_target_schema, 1);
+ assert_eq!(cfg.prom.native_histogram_max_buckets, 512);
+ // check_common_config scales sizes in place, so each call gets a fresh config
+ for schema in [-4, 8] {
+ let mut cfg = Config::init().unwrap();
+ cfg.prom.exp_histogram_target_schema = schema;
+ check_common_config(&mut cfg).unwrap();
+ }
+ for schema in [-5, 9] {
+ let mut cfg = Config::init().unwrap();
+ cfg.prom.exp_histogram_target_schema = schema;
+ let err = check_common_config(&mut cfg).unwrap_err().to_string();
+ assert_eq!(
+ err,
+ format!(
+ "ZO_METRICS_EXP_HISTOGRAM_TARGET_SCHEMA must be within -4..=8, got {schema}"
+ )
+ );
+ }
+ }
+
#[test]
fn test_check_s3_config_bucket_prefix_trailing_slash() {
let mut cfg = Config::default();
diff --git a/src/core/src/metrics/native_histogram.rs b/src/core/src/metrics/native_histogram.rs
index d0496ecbaf..95547ed316 100644
--- a/src/core/src/metrics/native_histogram.rs
+++ b/src/core/src/metrics/native_histogram.rs
@@ -13,9 +13,9 @@
// You should have received a copy of the GNU Affero General Public License
// along with this program. If not, see .
-//! Degrades Prometheus native (sparse) histograms into their classic representation:
-//! `_count`, `_sum` and cumulative `le` `_bucket` records, so existing PromQL
-//! (`histogram_quantile` etc.) works on them unchanged.
+//! Degrades exponential-bucket histograms (Prometheus native, OTLP exponential) into
+//! their classic representation: `_count`, `_sum` and cumulative `le` `_bucket` records,
+//! so existing PromQL (`histogram_quantile` etc.) works on them unchanged.
//!
//! Known limitations, inherited from classic semantics:
//! - `sum by (le)` is only sound across series sharing a bucket layout. Each native series carries
@@ -36,6 +36,9 @@ pub const CLASSIC_HISTOGRAM_SUFFIXES: [&str; 3] = ["_bucket", "_count", "_sum"];
/// schema -53 -- cannot be converted: our wire type does not decode `custom_values`.
const SCHEMA_RANGE: std::ops::RangeInclusive = -4..=8;
+/// Finest schema ever emitted: label rounding keeps adjacent bounds distinct only up to here.
+const MAX_SCHEMA: i32 = *SCHEMA_RANGE.end();
+
/// Downscaling may merge below the native schema floor -- the emitted `le` bounds are
/// plain classic bounds, not required to form a valid native schema. At -10
/// (base 2^1024) one bucket spans all of f64, so the loop always terminates. A sample
@@ -49,13 +52,126 @@ pub type ClassicHistogramRecord = (&'static str, Option, f64);
/// Prometheus's stale-marker bit pattern in `sum`; an ordinary NaN is NOT stale.
const STALE_NAN_BITS: u64 = 0x7ff0_0000_0000_0002;
-/// Degrades one native histogram sample into classic records: cumulative `le` buckets
-/// closed by `le="inf"`. Empty for unsupported schemas and stale markers. A sample
-/// expanding to more than `max_buckets` `le` labels is downscaled (adjacent buckets
-/// merged) until it fits.
+/// Emitted schema is `min(producer schema, target_schema)`; `max_buckets` is only a safety valve.
+#[derive(Clone, Copy, Debug)]
+pub struct ExpansionLimits {
+ pub target_schema: i32,
+ pub max_buckets: usize,
+}
+
+impl ExpansionLimits {
+ pub fn from_config(cfg: &config::Config) -> Self {
+ Self {
+ target_schema: cfg.prom.exp_histogram_target_schema,
+ max_buckets: cfg.prom.native_histogram_max_buckets,
+ }
+ }
+}
+
+pub struct ExponentialHistogram {
+ pub schema: i32,
+ pub count: f64,
+ /// `None` emits no `_sum` record (OTLP marks `sum` optional).
+ pub sum: Option,
+ pub zero_count: f64,
+ pub zero_threshold: f64,
+ pub positive: Vec<(i64, f64)>,
+ pub negative: Vec<(i64, f64)>,
+}
+
+impl ExponentialHistogram {
+ pub fn expand(self, limits: ExpansionLimits) -> Vec {
+ let Self {
+ mut schema,
+ count,
+ sum,
+ mut zero_count,
+ zero_threshold,
+ mut positive,
+ mut negative,
+ } = self;
+ // a NaN or negative threshold reads as no zero region
+ let zero_threshold = zero_threshold.max(0.0);
+ // a NaN or negative zero count cannot cumulate, so it reads as an empty zero bucket
+ if zero_count.is_nan() || zero_count < 0.0 {
+ zero_count = 0.0;
+ }
+
+ // one deterministic step to the target, so the layout only follows the producer's schema
+ let target = limits.target_schema.clamp(MIN_DOWNSCALE_SCHEMA, MAX_SCHEMA);
+ if schema > target {
+ let steps = (i64::from(schema) - i64::from(target)).min(i64::from(u32::MAX)) as u32;
+ positive = downscale(positive, steps);
+ negative = downscale(negative, steps);
+ schema = target;
+ }
+ // merging past the target changes the series layout, so this valve must stay exceptional
+ let max_labels = limits.max_buckets.max(3);
+ if le_estimate(&positive, &negative, zero_threshold) > max_labels {
+ log::warn!(
+ "[METRICS] exponential histogram exceeds {max_labels} le labels at schema {schema}; downscaling"
+ );
+ }
+ while le_estimate(&positive, &negative, zero_threshold) > max_labels
+ && schema > MIN_DOWNSCALE_SCHEMA
+ {
+ schema -= 1;
+ positive = downscale(positive, 1);
+ negative = downscale(negative, 1);
+ }
+
+ // buckets inside [-zero_threshold, zero_threshold] fold into the zero bucket, overlaps clip
+ let mut buckets: Vec<(f64, f64, f64)> = Vec::new(); // (lower, upper, count)
+ for &(idx, c) in &positive {
+ let (lower, upper) = (bucket_bound(schema, idx - 1), bucket_bound(schema, idx));
+ if upper <= zero_threshold {
+ zero_count += c;
+ } else {
+ buckets.push((lower.max(zero_threshold), upper, c));
+ }
+ }
+ for &(idx, c) in &negative {
+ let (lower, upper) = (-bucket_bound(schema, idx), -bucket_bound(schema, idx - 1));
+ if lower >= -zero_threshold {
+ zero_count += c;
+ } else {
+ buckets.push((lower, upper.min(-zero_threshold), c));
+ }
+ }
+ // the zero bucket is always reported, so its `le` series must exist before it fills
+ buckets.push((-zero_threshold, zero_threshold, zero_count));
+
+ // classic buckets are cumulative in `le` order
+ buckets.sort_by(|a, b| a.1.total_cmp(&b.1));
+
+ let mut recs = Vec::with_capacity(2 * buckets.len() + 3);
+ recs.push(("_count", None, count));
+ if let Some(sum) = sum {
+ recs.push(("_sum", None, sum));
+ }
+
+ let mut points: Vec<(String, f64)> = Vec::with_capacity(2 * buckets.len() + 1);
+ let mut cumulative = 0.0;
+ let mut prev_upper = f64::NEG_INFINITY;
+ for (lower, upper, c) in buckets {
+ if lower > prev_upper && lower < upper {
+ push_le_point(&mut points, format_le(lower), cumulative);
+ }
+ cumulative += c;
+ push_le_point(&mut points, format_le(upper), cumulative);
+ prev_upper = upper;
+ }
+ // `le="inf"` must equal `_count`; `max` also repairs a short count field
+ push_le_point(&mut points, format_le(f64::INFINITY), count.max(cumulative));
+
+ recs.extend(points.into_iter().map(|(le, v)| ("_bucket", Some(le), v)));
+ recs
+ }
+}
+
pub fn expand_native_histogram(
hp: &prometheus_rpc::Histogram,
- max_buckets: usize,
+ limits: ExpansionLimits,
) -> Vec {
if !SCHEMA_RANGE.contains(&hp.schema) {
log::warn!(
@@ -82,67 +198,26 @@ pub fn expand_native_histogram(
None => 0.0,
};
- let mut pos = span_buckets(&hp.positive_spans, &hp.positive_deltas, &hp.positive_counts);
- let mut neg = span_buckets(&hp.negative_spans, &hp.negative_deltas, &hp.negative_counts);
-
- // every emitted `le` label becomes a series, so merge adjacent buckets (halving
- // resolution) until the sample's le count fits the cardinality budget
- let mut schema = hp.schema;
- while le_estimate(&pos, &neg, zero_count) > max_buckets.max(3) && schema > MIN_DOWNSCALE_SCHEMA
- {
- schema -= 1;
- pos = downscale(pos);
- neg = downscale(neg);
+ ExponentialHistogram {
+ schema: hp.schema,
+ count,
+ sum: Some(hp.sum),
+ zero_count,
+ zero_threshold: hp.zero_threshold,
+ positive: span_buckets(&hp.positive_spans, &hp.positive_deltas, &hp.positive_counts),
+ negative: span_buckets(&hp.negative_spans, &hp.negative_deltas, &hp.negative_counts),
}
-
- // bucket `idx` covers `(base^(idx-1), base^idx]`, mirrored on the negative side
- let mut buckets: Vec<(f64, f64, f64)> = Vec::new(); // (lower, upper, count)
- for &(idx, c) in &pos {
- buckets.push((bucket_bound(schema, idx - 1), bucket_bound(schema, idx), c));
- }
- for &(idx, c) in &neg {
- buckets.push((
- -bucket_bound(schema, idx),
- -bucket_bound(schema, idx - 1),
- c,
- ));
- }
- if zero_count > 0.0 {
- buckets.push((-hp.zero_threshold, hp.zero_threshold, zero_count));
- }
-
- // classic buckets are cumulative in `le` order
- buckets.sort_by(|a, b| a.1.total_cmp(&b.1));
-
- let mut recs = Vec::with_capacity(2 * buckets.len() + 3);
- recs.push(("_count", None, count));
- recs.push(("_sum", None, hp.sum));
-
- let mut points: Vec<(String, f64)> = Vec::with_capacity(2 * buckets.len() + 1);
- let mut cumulative = 0.0;
- let mut prev_upper = f64::NEG_INFINITY;
- for (lower, upper, c) in buckets {
- // a zero-increment record at the lower bound of each bucket run pins sparse
- // gaps, so quantile interpolation cannot smear counts across them.
- // `lower < upper` skips the marker when clamping collapsed both bounds.
- if lower > prev_upper && lower < upper {
- push_le_point(&mut points, format_le(lower), cumulative);
- }
- cumulative += c;
- push_le_point(&mut points, format_le(upper), cumulative);
- prev_upper = upper;
- }
- // `le="inf"` must equal `_count`; `max` also repairs a short count field
- push_le_point(&mut points, format_le(f64::INFINITY), count.max(cumulative));
-
- recs.extend(points.into_iter().map(|(le, v)| ("_bucket", Some(le), v)));
- recs
+ .expand(limits)
}
/// `le` label: 4 significant digits, shortest decimal ("0.5946", "8"). Cannot collide
/// adjacent buckets -- the finest schema spaces bounds 0.271% apart vs the 0.1%
/// worst-case label resolution.
fn format_le(v: f64) -> String {
+ // a negated underflowed bound is -0.0, which must share the zero bucket's label
+ if v == 0.0 {
+ return "0".to_string();
+ }
let rounded: f64 = format!("{v:.3e}").parse().unwrap();
if rounded.is_finite() || !v.is_finite() {
rounded.to_string()
@@ -162,19 +237,16 @@ fn push_le_point(points: &mut Vec<(String, f64)>, le: String, cumulative: f64) {
}
}
-/// Upper bound of bucket `idx`: `2^(idx * 2^-schema)` via `exp2`, so powers of two are
-/// exact and a boundary shared between schemas is the identical f64 (downscaling keeps
-/// surviving `le`s in the same series). Clamped finite so the last representable
-/// bucket cannot collide with `le="inf"`.
fn bucket_bound(schema: i32, idx: i64) -> f64 {
- let bound = ((idx as f64) * 2f64.powi(-schema)).exp2();
- bound.clamp(f64::MIN_POSITIVE, f64::MAX)
+ if idx == 0 {
+ return 1.0;
+ }
+ // widened so `-i32::MIN` cannot overflow; past ±1100 the factor is already 0 or inf
+ let factor = 2f64.powi((-(schema as i64)).clamp(-1100, 1100) as i32);
+ ((idx as f64) * factor).exp2().min(f64::MAX)
}
-/// Upper bound on the `le` labels a sample will emit: each populated bucket gets an
-/// upper record, each contiguous run a lower-bound gap marker, plus the zero bucket's
-/// two bounds and `inf`.
-fn le_estimate(pos: &[(i64, f64)], neg: &[(i64, f64)], zero_count: f64) -> usize {
+fn le_estimate(pos: &[(i64, f64)], neg: &[(i64, f64)], zero_threshold: f64) -> usize {
let side = |b: &[(i64, f64)]| {
let runs = b
.iter()
@@ -184,15 +256,15 @@ fn le_estimate(pos: &[(i64, f64)], neg: &[(i64, f64)], zero_count: f64) -> usize
+ usize::from(!b.is_empty());
b.len() + runs
};
- side(pos) + side(neg) + if zero_count > 0.0 { 2 } else { 0 } + 1
+ side(pos) + side(neg) + if zero_threshold > 0.0 { 2 } else { 1 } + 1
}
-/// Merges adjacent bucket pairs: `idx` at schema `s` maps to `ceil(idx / 2)` at
-/// schema `s - 1`.
-fn downscale(buckets: Vec<(i64, f64)>) -> Vec<(i64, f64)> {
+fn downscale(buckets: Vec<(i64, f64)>, steps: u32) -> Vec<(i64, f64)> {
+ // indices are i32 offsets plus a length, so 2^62 already maps every bucket to 0 or 1
+ let span = 1i64 << steps.min(62);
let mut out: Vec<(i64, f64)> = Vec::with_capacity(buckets.len() / 2 + 1);
for (idx, c) in buckets {
- let merged_idx = (idx + 1).div_euclid(2);
+ let merged_idx = (idx + span - 1).div_euclid(span);
match out.last_mut() {
Some((last_idx, last_c)) if *last_idx == merged_idx => *last_c += c,
_ => out.push((merged_idx, c)),
@@ -201,8 +273,6 @@ fn downscale(buckets: Vec<(i64, f64)>) -> Vec<(i64, f64)> {
out
}
-/// Decodes the sparse layout (spans + integer deltas or absolute float counts) into
-/// `(bucket index, absolute count)` pairs for every populated bucket.
fn span_buckets(
spans: &[prometheus_rpc::BucketSpan],
deltas: &[i64],
@@ -229,8 +299,7 @@ fn span_buckets(
cumulative_delta as f64
};
pos += 1;
- // skips zero-count buckets and NaN float counts
- if count > 0.0 {
+ if count >= 0.0 {
buckets.push((idx, count));
}
idx += 1;
@@ -243,6 +312,14 @@ fn span_buckets(
mod tests {
use super::*;
+ /// No target downscaling: the producer's schema is emitted as is, capped only by the valve.
+ fn lim(max_buckets: usize) -> ExpansionLimits {
+ ExpansionLimits {
+ target_schema: MAX_SCHEMA,
+ max_buckets,
+ }
+ }
+
fn native_histogram_base() -> prometheus_rpc::Histogram {
prometheus_rpc::Histogram {
schema: 0, // base = 2
@@ -296,7 +373,7 @@ mod tests {
..native_histogram_base()
};
- let recs = expand_native_histogram(&hp, 16);
+ let recs = expand_native_histogram(&hp, lim(16));
assert_eq!(scalar_record(&recs, "_count"), 12.0);
assert_eq!(scalar_record(&recs, "_sum"), 100.0);
// gap markers: le=4 pins the empty (2,4], le=-0.001/0.5 open each bucket run
@@ -337,15 +414,16 @@ mod tests {
..native_histogram_base()
};
- let recs = expand_native_histogram(&hp, 16);
+ let recs = expand_native_histogram(&hp, lim(16));
assert_eq!(
bucket_records(&recs),
vec![
(-2.0, 0.0), // lower-bound marker of the negative run
(-1.0, 3.0), // negative idx 1: [-2, -1)
(-0.5, 5.0), // negative idx 0: [-1, -0.5)
- (1.0, 5.0), // lower-bound marker: (-0.5, 1] carries nothing
- (2.0, 9.0), // positive idx 1: (1, 2]
+ (0.0, 5.0),
+ (1.0, 5.0), // lower-bound marker: (-0.5, 1] carries nothing
+ (2.0, 9.0), // positive idx 1: (1, 2]
(f64::INFINITY, 9.0),
]
);
@@ -359,7 +437,7 @@ mod tests {
count: Some(prometheus_rpc::histogram::Count::CountInt(5)),
..Default::default()
};
- assert!(expand_native_histogram(&hp, 16).is_empty());
+ assert!(expand_native_histogram(&hp, lim(16)).is_empty());
}
/// A stale marker drops the whole sample; an ordinary NaN sum keeps count/buckets
@@ -370,7 +448,7 @@ mod tests {
sum: f64::from_bits(0x7ff0_0000_0000_0002),
..native_histogram_base()
};
- assert!(expand_native_histogram(&stale, 16).is_empty());
+ assert!(expand_native_histogram(&stale, lim(16)).is_empty());
let plain_nan = prometheus_rpc::Histogram {
sum: f64::NAN,
@@ -382,12 +460,12 @@ mod tests {
positive_deltas: vec![7],
..native_histogram_base()
};
- let recs = expand_native_histogram(&plain_nan, 16);
+ let recs = expand_native_histogram(&plain_nan, lim(16));
assert_eq!(scalar_record(&recs, "_count"), 7.0);
assert!(scalar_record(&recs, "_sum").is_nan());
assert_eq!(
bucket_records(&recs),
- vec![(0.5, 0.0), (1.0, 7.0), (f64::INFINITY, 7.0)]
+ vec![(0.0, 0.0), (0.5, 0.0), (1.0, 7.0), (f64::INFINITY, 7.0)]
);
}
@@ -395,9 +473,12 @@ mod tests {
/// back to the bucket total so `le="inf"` stays monotonic.
#[test]
fn test_expand_native_histogram_empty_and_short_count() {
- let empty = expand_native_histogram(&native_histogram_base(), 16);
+ let empty = expand_native_histogram(&native_histogram_base(), lim(16));
assert_eq!(scalar_record(&empty, "_count"), 0.0);
- assert_eq!(bucket_records(&empty), vec![(f64::INFINITY, 0.0)]);
+ assert_eq!(
+ bucket_records(&empty),
+ vec![(0.0, 0.0), (f64::INFINITY, 0.0)]
+ );
let hp = prometheus_rpc::Histogram {
positive_spans: vec![prometheus_rpc::BucketSpan {
@@ -407,10 +488,10 @@ mod tests {
positive_deltas: vec![7],
..native_histogram_base()
};
- let recs = expand_native_histogram(&hp, 16);
+ let recs = expand_native_histogram(&hp, lim(16));
assert_eq!(
bucket_records(&recs),
- vec![(0.5, 0.0), (1.0, 7.0), (f64::INFINITY, 7.0)]
+ vec![(0.0, 0.0), (0.5, 0.0), (1.0, 7.0), (f64::INFINITY, 7.0)]
);
}
@@ -436,7 +517,7 @@ mod tests {
positive_deltas: vec![2, -1, 2, -1, 1],
..native_histogram_base()
};
- let recs = expand_native_histogram(&hp, 16);
+ let recs = expand_native_histogram(&hp, lim(16));
assert_bucket_invariants(&recs);
assert_eq!(scalar_record(&recs, "_count"), 13.0);
assert_eq!(scalar_record(&recs, "_sum"), 175.5);
@@ -471,17 +552,17 @@ mod tests {
positive_deltas: vec![5, -2],
..native_histogram_base()
};
- let recs = expand_native_histogram(&hp, 16);
+ let recs = expand_native_histogram(&hp, lim(16));
assert_bucket_invariants(&recs);
let buckets = bucket_records(&recs);
- // lower marker at idx 249's bound, uppers for idx 250 and 251, then inf
- assert_eq!(buckets.len(), 4);
+ // zero bucket, lower marker at idx 249's bound, uppers for idx 250 and 251, then inf
+ assert_eq!(buckets.len(), 5);
assert_eq!(
buckets.iter().map(|(_, c)| *c).collect::>(),
- vec![0.0, 5.0, 8.0, 8.0]
+ vec![0.0, 0.0, 5.0, 8.0, 8.0]
);
- assert_eq!(buckets[1].0, 1.968);
- assert_eq!(buckets[2].0, 1.973);
+ assert_eq!(buckets[2].0, 1.968);
+ assert_eq!(buckets[3].0, 1.973);
}
/// Power-of-two bounds are exact and a boundary shared between schemas is the
@@ -519,7 +600,7 @@ mod tests {
#[test]
fn test_expand_native_histogram_merges_le_rounding_collisions() {
let hp = prometheus_rpc::Histogram {
- zero_threshold: 1.0001, // rounds to le="1", same as bucket idx 0's bound
+ zero_threshold: 0.99995, // rounds to le="1", same as bucket idx 0's bound
count: Some(prometheus_rpc::histogram::Count::CountInt(5)),
zero_count: Some(prometheus_rpc::histogram::ZeroCount::ZeroCountInt(2)),
positive_spans: vec![prometheus_rpc::BucketSpan {
@@ -529,12 +610,12 @@ mod tests {
positive_deltas: vec![3],
..native_histogram_base()
};
- let recs = expand_native_histogram(&hp, 16);
+ let recs = expand_native_histogram(&hp, lim(16));
assert_bucket_invariants(&recs);
// both uppers format to "1" and merge, keeping the full cumulative 5
assert_eq!(
bucket_records(&recs),
- vec![(0.5, 0.0), (1.0, 5.0), (f64::INFINITY, 5.0)]
+ vec![(-1.0, 0.0), (1.0, 5.0), (f64::INFINITY, 5.0)]
);
}
@@ -553,15 +634,16 @@ mod tests {
};
// no limit pressure: 8 buckets stay at schema 2
- let full = expand_native_histogram(&hp, 64);
- assert_eq!(bucket_records(&full).len(), 10); // 8 uppers + 1 lower marker + inf
+ let full = expand_native_histogram(&hp, lim(64));
+ assert_eq!(bucket_records(&full).len(), 11); // zero bucket + 8 uppers + 1 lower marker + inf
- // limit 4 le labels: two halvings land on schema 0
- let scaled = expand_native_histogram(&hp, 4);
+ // limit 5 le labels: two halvings land on schema 0
+ let scaled = expand_native_histogram(&hp, lim(5));
assert_bucket_invariants(&scaled);
assert_eq!(
bucket_records(&scaled),
vec![
+ (0.0, 0.0),
(1.0, 0.0), // lower marker
(2.0, 10.0), // (1, 2]: 1+2+3+4
(4.0, 36.0), // (2, 4]: 5+6+7+8
@@ -589,11 +671,10 @@ mod tests {
..native_histogram_base()
};
- // 4 buckets + 4 markers + inf = 9 labels > 8: one halving makes them
- // contiguous (idx 0..=3 at schema 2), 6 labels
- let recs = expand_native_histogram(&hp, 8);
+ // 4 buckets + 4 markers + zero bucket + inf = 10 > 8: one halving makes 7 contiguous labels
+ let recs = expand_native_histogram(&hp, lim(8));
assert_bucket_invariants(&recs);
- assert_eq!(bucket_records(&recs).len(), 6);
+ assert_eq!(bucket_records(&recs).len(), 7);
assert_eq!(scalar_record(&recs, "_count"), 4.0);
}
@@ -613,7 +694,7 @@ mod tests {
positive_deltas: vec![1, 0, 0, 0, 0, 0], // idx 0,2,4,6,8,10
..native_histogram_base()
};
- let recs = expand_native_histogram(&hp, 5);
+ let recs = expand_native_histogram(&hp, lim(5));
assert_bucket_invariants(&recs);
assert!(bucket_records(&recs).len() <= 5);
assert_eq!(scalar_record(&recs, "_count"), 6.0);
@@ -636,7 +717,7 @@ mod tests {
positive_deltas: vec![3],
..native_histogram_base()
};
- let recs = expand_native_histogram(&hp, 16);
+ let recs = expand_native_histogram(&hp, lim(16));
assert_bucket_invariants(&recs);
assert_eq!(
bucket_records(&recs),
@@ -644,6 +725,168 @@ mod tests {
);
}
+ #[test]
+ fn test_expand_native_histogram_zero_threshold_folds_and_clips_both_sides() {
+ let hp = prometheus_rpc::Histogram {
+ zero_threshold: 1.5, // (0.5, 1] lies inside, (1, 2] overlaps
+ count: Some(prometheus_rpc::histogram::Count::CountInt(9)),
+ zero_count: Some(prometheus_rpc::histogram::ZeroCount::ZeroCountInt(1)),
+ positive_spans: vec![prometheus_rpc::BucketSpan {
+ offset: 0,
+ length: 2,
+ }],
+ positive_deltas: vec![2, 1], // idx 0: 2, idx 1: 3
+ negative_spans: vec![prometheus_rpc::BucketSpan {
+ offset: 0,
+ length: 2,
+ }],
+ negative_deltas: vec![1, 1], // idx 0: 1, idx 1: 2
+ ..native_histogram_base()
+ };
+ let recs = expand_native_histogram(&hp, lim(16));
+ assert_bucket_invariants(&recs);
+ assert_eq!(
+ bucket_records(&recs),
+ vec![
+ (-2.0, 0.0),
+ (-1.5, 2.0),
+ (1.5, 6.0), // zero 1 + inside buckets 2 + 1
+ (2.0, 9.0),
+ (f64::INFINITY, 9.0),
+ ]
+ );
+ }
+
+ #[test]
+ fn test_expand_extreme_schema_is_bounded() {
+ let recs = ExponentialHistogram {
+ schema: i32::MAX,
+ count: 3.0,
+ sum: None,
+ zero_count: 0.0,
+ zero_threshold: 0.0,
+ positive: vec![(1, 1.0), (5, 1.0)],
+ negative: vec![(1, 1.0)],
+ }
+ .expand(lim(16));
+ assert_bucket_invariants(&recs);
+ // every index collapses onto schema-8 bucket 1, (1, 2^(1/256)], on both sides
+ assert_eq!(
+ bucket_records(&recs),
+ vec![
+ (-1.003, 0.0),
+ (-1.0, 1.0),
+ (0.0, 1.0),
+ (1.0, 1.0),
+ (1.003, 3.0),
+ (f64::INFINITY, 3.0),
+ ]
+ );
+ }
+
+ #[test]
+ fn test_layout_follows_target_schema_not_counts() {
+ // schema 3, one contiguous run: 13 then 14 then 28 populated buckets
+ let sample = |n: i64, per: f64| ExponentialHistogram {
+ schema: 3,
+ count: n as f64 * per,
+ sum: None,
+ zero_count: 0.0,
+ zero_threshold: 0.0,
+ positive: (1..=n).map(|i| (i, per)).collect(),
+ negative: vec![],
+ };
+ let limits = ExpansionLimits {
+ target_schema: 2,
+ max_buckets: 512,
+ };
+ let les = |recs: &[ClassicHistogramRecord]| -> Vec {
+ recs.iter().filter_map(|(_, le, _)| le.clone()).collect()
+ };
+ let a = les(&sample(13, 1.0).expand(limits));
+ let b = les(&sample(14, 2.0).expand(limits));
+ let c = les(&sample(28, 3.0).expand(limits));
+ // every emitted schema-2 bound survives into the next sample; the old valve switched here
+ assert!(a.iter().all(|le| b.contains(le)), "{a:?} vs {b:?}");
+ assert!(b.iter().all(|le| c.contains(le)), "{b:?} vs {c:?}");
+ assert_eq!(a.len(), 7 + 1 + 1 + 1); // ceil(13/2)=7 uppers + marker + zero + inf
+ }
+
+ #[test]
+ fn test_layout_identical_across_producer_schemas() {
+ // the same distribution at schema 5 and schema 3 lands on identical schema-2 records
+ let coarse = ExponentialHistogram {
+ schema: 3,
+ count: 10.0,
+ sum: Some(1.0),
+ zero_count: 0.0,
+ zero_threshold: 0.0,
+ positive: vec![(1, 3.0), (2, 0.0), (3, 7.0)],
+ negative: vec![],
+ };
+ // schema 5 splits each schema-3 bucket k into 4k-3..=4k
+ let fine = ExponentialHistogram {
+ schema: 5,
+ count: 10.0,
+ sum: Some(1.0),
+ zero_count: 0.0,
+ zero_threshold: 0.0,
+ positive: vec![(2, 3.0), (5, 0.0), (11, 7.0)],
+ negative: vec![],
+ };
+ let limits = ExpansionLimits {
+ target_schema: 2,
+ max_buckets: 512,
+ };
+ assert_eq!(coarse.expand(limits), fine.expand(limits));
+ }
+
+ #[test]
+ fn test_valve_only_past_max_buckets() {
+ let hp = prometheus_rpc::Histogram {
+ count: Some(prometheus_rpc::histogram::Count::CountInt(40)),
+ positive_spans: vec![prometheus_rpc::BucketSpan {
+ offset: 0,
+ length: 40,
+ }],
+ positive_deltas: std::iter::once(1)
+ .chain(std::iter::repeat_n(0, 39))
+ .collect(),
+ ..native_histogram_base()
+ };
+ let stable = expand_native_histogram(&hp, lim(512));
+ assert_eq!(bucket_records(&stable).len(), 40 + 1 + 1 + 1);
+ let valve = expand_native_histogram(&hp, lim(16));
+ assert!(bucket_records(&valve).len() <= 16);
+ assert_bucket_invariants(&valve);
+ }
+
+ #[test]
+ fn test_bucket_bound_extreme_negative_schema() {
+ assert_eq!(bucket_bound(i32::MIN, 0), 1.0);
+ assert_eq!(bucket_bound(i32::MIN, 1), f64::MAX);
+ assert_eq!(bucket_bound(i32::MIN, -1), 0.0);
+ let recs = ExponentialHistogram {
+ schema: i32::MIN,
+ count: 2.0,
+ sum: None,
+ zero_count: 0.0,
+ zero_threshold: 0.0,
+ positive: vec![(1, 1.0)],
+ negative: vec![(1, 1.0)],
+ }
+ .expand(lim(16));
+ assert_bucket_invariants(&recs);
+ }
+
+ #[test]
+ fn test_bucket_bound_keeps_subnormals() {
+ assert_eq!(bucket_bound(0, -1023), 2f64.powi(-1023));
+ assert!(bucket_bound(0, -1023) < f64::MIN_POSITIVE);
+ assert_eq!(bucket_bound(0, -1080), 0.0);
+ assert_eq!(format_le(-0.0), "0");
+ }
+
/// The gauge reset hint changes rate/reset semantics at query time, not the bucket
/// layout: expansion output is identical with and without it.
#[test]
@@ -663,8 +906,8 @@ mod tests {
..counter.clone()
};
assert_eq!(
- expand_native_histogram(&counter, 16),
- expand_native_histogram(&gauge, 16)
+ expand_native_histogram(&counter, lim(16)),
+ expand_native_histogram(&gauge, lim(16))
);
}
@@ -678,7 +921,7 @@ mod tests {
zero_count: Some(prometheus_rpc::histogram::ZeroCount::ZeroCountInt(5)),
..native_histogram_base()
};
- let recs = expand_native_histogram(&hp, 16);
+ let recs = expand_native_histogram(&hp, lim(16));
assert_bucket_invariants(&recs);
assert_eq!(
bucket_records(&recs),
@@ -699,11 +942,11 @@ mod tests {
positive_deltas: vec![9], // carries 1
..native_histogram_base()
};
- let recs = expand_native_histogram(&hp, 16);
+ let recs = expand_native_histogram(&hp, lim(16));
assert_bucket_invariants(&recs);
assert_eq!(
bucket_records(&recs),
- vec![(0.5, 0.0), (1.0, 9.0), (f64::INFINITY, 9.0)]
+ vec![(0.0, 0.0), (0.5, 0.0), (1.0, 9.0), (f64::INFINITY, 9.0)]
);
}
@@ -719,12 +962,12 @@ mod tests {
positive_deltas: vec![7],
..native_histogram_base()
};
- let recs = expand_native_histogram(&hp, 16);
+ let recs = expand_native_histogram(&hp, lim(16));
assert_bucket_invariants(&recs);
assert_eq!(scalar_record(&recs, "_count"), 10.0);
assert_eq!(
bucket_records(&recs),
- vec![(0.5, 0.0), (1.0, 7.0), (f64::INFINITY, 10.0)]
+ vec![(0.0, 0.0), (0.5, 0.0), (1.0, 7.0), (f64::INFINITY, 10.0)]
);
}
@@ -747,11 +990,12 @@ mod tests {
positive_deltas: vec![3, -2],
..native_histogram_base()
};
- let recs = expand_native_histogram(&hp, 16);
+ let recs = expand_native_histogram(&hp, lim(16));
assert_bucket_invariants(&recs);
assert_eq!(
bucket_records(&recs),
vec![
+ (0.0, 0.0),
(0.5, 0.0),
(1.0, 3.0),
// rounding f64::MAX to 4 digits would overflow to infinity and collide
@@ -762,51 +1006,175 @@ mod tests {
);
}
- /// NaN float counts are skipped; the remaining buckets keep their positions.
+ /// NaN and negative float counts are skipped while an explicit zero count is kept.
#[test]
fn test_expand_native_histogram_nan_float_count_skipped() {
let hp = prometheus_rpc::Histogram {
count: Some(prometheus_rpc::histogram::Count::CountFloat(5.0)),
positive_spans: vec![prometheus_rpc::BucketSpan {
offset: 1,
- length: 2,
+ length: 4,
}],
- positive_counts: vec![f64::NAN, 5.0],
+ positive_counts: vec![f64::NAN, 0.0, -1.0, 5.0],
..native_histogram_base()
};
- let recs = expand_native_histogram(&hp, 16);
+ let recs = expand_native_histogram(&hp, lim(16));
assert_bucket_invariants(&recs);
- // idx 1 (NaN) skipped; idx 2 -> (2, 4] with its lower-bound marker
+ // idx 1 (NaN) skipped, so le=2 is the gap marker; idx 2 empty but kept; idx 3 (-1) skipped
assert_eq!(
bucket_records(&recs),
- vec![(2.0, 0.0), (4.0, 5.0), (f64::INFINITY, 5.0)]
+ vec![
+ (0.0, 0.0),
+ (2.0, 0.0),
+ (4.0, 0.0),
+ (8.0, 0.0),
+ (16.0, 5.0),
+ (f64::INFINITY, 5.0)
+ ]
);
}
- /// An interior empty bucket is skipped; the gap marker pins the next bucket's
- /// lower bound.
+ /// A negative integer delta run is skipped, so cumulative counts never decrease.
#[test]
- fn test_expand_native_histogram_interior_zero_bucket_gap() {
+ fn test_expand_native_histogram_negative_integer_count_skipped() {
let hp = prometheus_rpc::Histogram {
- count: Some(prometheus_rpc::histogram::Count::CountInt(5)),
+ count: Some(prometheus_rpc::histogram::Count::CountInt(7)),
positive_spans: vec![prometheus_rpc::BucketSpan {
offset: 0,
length: 3,
}],
- positive_deltas: vec![3, -3, 2], // buckets 3, 0, 2
+ positive_deltas: vec![3, -4, 5], // buckets 3, -1, 4
..native_histogram_base()
};
- let recs = expand_native_histogram(&hp, 16);
+ let recs = expand_native_histogram(&hp, lim(16));
assert_bucket_invariants(&recs);
assert_eq!(
bucket_records(&recs),
vec![
+ (0.0, 0.0),
(0.5, 0.0),
(1.0, 3.0),
- (2.0, 3.0), // idx 1 empty: le=2 is the gap marker for (2, 4]
- (4.0, 5.0),
- (f64::INFINITY, 5.0),
+ (2.0, 3.0), // idx 1 (-1) skipped: le=2 is the gap marker for (2, 4]
+ (4.0, 7.0),
+ (f64::INFINITY, 7.0),
]
);
}
+
+ #[test]
+ fn test_expand_native_histogram_interior_zero_bucket_kept() {
+ let hp = prometheus_rpc::Histogram {
+ count: Some(prometheus_rpc::histogram::Count::CountInt(7)),
+ positive_spans: vec![prometheus_rpc::BucketSpan {
+ offset: 0,
+ length: 3,
+ }],
+ positive_deltas: vec![3, -3, 4], // buckets 3, 0, 4
+ ..native_histogram_base()
+ };
+ let recs = expand_native_histogram(&hp, lim(16));
+ assert_bucket_invariants(&recs);
+ assert_eq!(
+ bucket_records(&recs),
+ vec![
+ (0.0, 0.0),
+ (0.5, 0.0),
+ (1.0, 3.0),
+ (2.0, 3.0), // idx 1 empty: its own upper record, not a gap marker
+ (4.0, 7.0),
+ (f64::INFINITY, 7.0),
+ ]
+ );
+ }
+
+ /// An explicitly empty bucket inside the zero region adds nothing and emits no record.
+ #[test]
+ fn test_expand_native_histogram_empty_bucket_in_zero_region() {
+ let hp = prometheus_rpc::Histogram {
+ count: Some(prometheus_rpc::histogram::Count::CountInt(2)),
+ zero_threshold: 4.0,
+ positive_spans: vec![prometheus_rpc::BucketSpan {
+ offset: 1,
+ length: 3,
+ }],
+ positive_deltas: vec![0, 0, 2], // buckets (1,2]=0, (2,4]=0, (4,8]=2
+ ..native_histogram_base()
+ };
+ let recs = expand_native_histogram(&hp, lim(16));
+ assert_bucket_invariants(&recs);
+ assert_eq!(
+ bucket_records(&recs),
+ vec![(-4.0, 0.0), (4.0, 0.0), (8.0, 2.0), (f64::INFINITY, 2.0)]
+ );
+ }
+
+ /// The zero bucket's `le` series exists before its first observation.
+ #[test]
+ fn test_expand_native_histogram_zero_bucket_series_stable_across_samples() {
+ let sample = |zero_count: u64| {
+ let hp = prometheus_rpc::Histogram {
+ count: Some(prometheus_rpc::histogram::Count::CountInt(3 + zero_count)),
+ zero_count: Some(prometheus_rpc::histogram::ZeroCount::ZeroCountInt(
+ zero_count,
+ )),
+ positive_spans: vec![prometheus_rpc::BucketSpan {
+ offset: 0,
+ length: 1,
+ }],
+ positive_deltas: vec![3],
+ ..native_histogram_base()
+ };
+ let recs = expand_native_histogram(&hp, lim(16));
+ assert_bucket_invariants(&recs);
+ bucket_records(&recs)
+ };
+ assert_eq!(
+ sample(0),
+ vec![(0.0, 0.0), (0.5, 0.0), (1.0, 3.0), (f64::INFINITY, 3.0)]
+ );
+ assert_eq!(
+ sample(1),
+ vec![(0.0, 1.0), (0.5, 1.0), (1.0, 4.0), (f64::INFINITY, 4.0)]
+ );
+ }
+
+ /// A negative or NaN float zero count reads as an empty zero bucket; neighbours are unaffected.
+ #[test]
+ fn test_expand_native_histogram_invalid_zero_count_reads_as_empty() {
+ for zero_count in [-1.0, f64::NAN] {
+ let hp = prometheus_rpc::Histogram {
+ count: Some(prometheus_rpc::histogram::Count::CountFloat(7.0)),
+ zero_count: Some(prometheus_rpc::histogram::ZeroCount::ZeroCountFloat(
+ zero_count,
+ )),
+ zero_threshold: 0.1,
+ positive_spans: vec![prometheus_rpc::BucketSpan {
+ offset: 0,
+ length: 1,
+ }],
+ positive_counts: vec![4.0],
+ negative_spans: vec![prometheus_rpc::BucketSpan {
+ offset: 0,
+ length: 1,
+ }],
+ negative_counts: vec![3.0],
+ ..native_histogram_base()
+ };
+ let recs = expand_native_histogram(&hp, lim(16));
+ assert_bucket_invariants(&recs);
+ assert_eq!(
+ bucket_records(&recs),
+ vec![
+ (-1.0, 0.0),
+ (-0.5, 3.0),
+ (-0.1, 3.0),
+ (0.1, 3.0),
+ (0.5, 3.0),
+ (1.0, 7.0),
+ (f64::INFINITY, 7.0),
+ ],
+ "zero_count {zero_count}"
+ );
+ }
+ }
}
diff --git a/src/core/src/metrics/otlp.rs b/src/core/src/metrics/otlp.rs
index 0761929711..fd7f416d32 100644
--- a/src/core/src/metrics/otlp.rs
+++ b/src/core/src/metrics/otlp.rs
@@ -57,6 +57,7 @@ use schema::stream_schema_exists;
use super::{
columnar::{self, ColumnarStream},
ingest::{self, PipelineFailure, PipelineInputs, RecordsByStream},
+ native_histogram,
};
use crate::{
common::meta::{http::HttpResponse as MetaHttpResponse, stream::SchemaRecords},
@@ -873,9 +874,10 @@ fn process_exponential_histogram(
);
let mut records = vec![];
process_aggregation_temporality(rec, hist.aggregation_temporality);
+ let limits = native_histogram::ExpansionLimits::from_config(&config::get_config());
for data_point in &hist.data_points {
let mut dp_rec = rec.clone();
- for mut bucket_rec in process_exp_hist_data_point(&mut dp_rec, data_point) {
+ for mut bucket_rec in process_exp_hist_data_point(&mut dp_rec, data_point, limits) {
let val_map = bucket_rec.as_object_mut().unwrap();
let hash = super::signature_without_labels(val_map, METRICS_HASH_EXCLUDED_LABELS);
val_map.insert(HASH_LABEL.to_string(), json::Value::Number(hash.into()));
@@ -995,6 +997,7 @@ fn process_hist_data_point(
fn process_exp_hist_data_point(
rec: &mut json::Value,
data_point: &ExponentialHistogramDataPoint,
+ limits: native_histogram::ExpansionLimits,
) -> Vec {
if no_recorded_value(data_point.flags) {
return vec![];
@@ -1006,46 +1009,53 @@ fn process_exp_hist_data_point(
rec["start_time"] = data_point.start_time_unix_nano.to_string().into();
rec["flag"] = data_point_flag(data_point.flags).into();
process_exemplars(rec, &data_point.exemplars);
- // add count record
- let mut count_rec = rec.clone();
- count_rec[VALUE_LABEL] = (data_point.count as f64).into();
- count_rec[NAME_LABEL] = format!("{}_count", count_rec[NAME_LABEL].as_str().unwrap()).into();
- bucket_recs.push(count_rec);
- // add sum record -- OTLP marks `sum` optional, so an absent (or NaN) sum emits no record
- if let Some(sum) = data_point.sum.and_then(super::metric_value) {
- let mut sum_rec = rec.clone();
- sum_rec[VALUE_LABEL] = sum;
- sum_rec[NAME_LABEL] = format!("{}_sum", sum_rec[NAME_LABEL].as_str().unwrap()).into();
- bucket_recs.push(sum_rec);
+ // OTLP bucket `i` covers `(base^i, base^(i+1)]`, the native layout's bucket `i+1`
+ let otlp_buckets = |b: &Option| -> Vec<(i64, f64)> {
+ let Some(b) = b else { return vec![] };
+ // explicitly reported empty buckets are kept so their `le` series exists before they fill
+ b.bucket_counts
+ .iter()
+ .enumerate()
+ .map(|(i, c)| (b.offset as i64 + i as i64 + 1, *c as f64))
+ .collect()
+ };
+ let expanded = native_histogram::ExponentialHistogram {
+ schema: data_point.scale,
+ count: data_point.count as f64,
+ // OTLP marks `sum` optional, so an absent sum emits no record
+ sum: data_point.sum,
+ zero_count: data_point.zero_count as f64,
+ zero_threshold: data_point.zero_threshold,
+ positive: otlp_buckets(&data_point.positive),
+ negative: otlp_buckets(&data_point.negative),
+ }
+ .expand(limits);
+ for (suffix, le, value) in expanded {
+ let Some(value) = super::metric_value(value) else {
+ continue;
+ };
+ let mut hist_rec = rec.clone();
+ hist_rec[VALUE_LABEL] = value;
+ hist_rec[NAME_LABEL] = format!("{}{suffix}", rec[NAME_LABEL].as_str().unwrap()).into();
+ if let Some(le) = le {
+ hist_rec["le"] = le.into();
+ }
+ bucket_recs.push(hist_rec);
}
- // OTLP exponential histogram bucket boundaries are powers of `base`, where
- // `base = 2^(2^-scale)`. Bucket index `idx` (offset + array position) covers the
- // range (base^idx, base^(idx+1)], so its upper bound is `base^(idx+1)`.
- // NOTE: `^` is bitwise XOR in Rust, not exponentiation -- use `powf`/`powi`.
- let base = 2f64.powf(2f64.powi(-data_point.scale));
- // add negative bucket records (negative values, so the boundary is negated)
- if let Some(buckets) = &data_point.negative {
- let offset = buckets.offset;
- for (i, val) in buckets.bucket_counts.iter().enumerate() {
- let mut bucket_rec = rec.clone();
- bucket_rec[NAME_LABEL] = format!("{}_bucket", rec[NAME_LABEL].as_str().unwrap()).into();
- bucket_rec[VALUE_LABEL] = (*val as f64).into();
- bucket_rec["le"] = (-base.powi(offset + (i as i32) + 1)).to_string().into();
- bucket_recs.push(bucket_rec);
- }
+ if let Some(min) = data_point.min.and_then(super::metric_value) {
+ let mut min_rec = rec.clone();
+ min_rec[VALUE_LABEL] = min;
+ min_rec[NAME_LABEL] = format!("{}_min", min_rec[NAME_LABEL].as_str().unwrap()).into();
+ bucket_recs.push(min_rec);
}
- // add positive bucket records
- if let Some(buckets) = &data_point.positive {
- let offset = buckets.offset;
- for (i, val) in buckets.bucket_counts.iter().enumerate() {
- let mut bucket_rec = rec.clone();
- bucket_rec[NAME_LABEL] = format!("{}_bucket", rec[NAME_LABEL].as_str().unwrap()).into();
- bucket_rec[VALUE_LABEL] = (*val as f64).into();
- bucket_rec["le"] = base.powi(offset + (i as i32) + 1).to_string().into();
- bucket_recs.push(bucket_rec);
- }
+
+ if let Some(max) = data_point.max.and_then(super::metric_value) {
+ let mut max_rec = rec.clone();
+ max_rec[VALUE_LABEL] = max;
+ max_rec[NAME_LABEL] = format!("{}_max", max_rec[NAME_LABEL].as_str().unwrap()).into();
+ bucket_recs.push(max_rec);
}
bucket_recs
@@ -1228,6 +1238,14 @@ mod tests {
use super::*;
+ /// No target downscaling: the producer's scale is emitted as is, capped only by the valve.
+ fn lim(max_buckets: usize) -> native_histogram::ExpansionLimits {
+ native_histogram::ExpansionLimits {
+ target_schema: 8,
+ max_buckets,
+ }
+ }
+
fn process_gauge(
rec: &json::Value,
gauge: &Gauge,
@@ -1677,17 +1695,31 @@ mod tests {
}),
};
- let result = process_exp_hist_data_point(&mut rec, &data_point);
+ let result = process_exp_hist_data_point(&mut rec, &data_point, lim(16));
- // Verify the processed data
- assert!(!result.is_empty());
- // Should have count, sum, and bucket records
- assert!(result.len() >= 2);
-
- // Check count record
let count_rec = &result[0];
assert!(count_rec["__name__"].as_str().unwrap().ends_with("_count"));
assert_eq!(count_rec["value"], json!(100.0));
+
+ // the explicitly empty negative bucket (-2, -1] keeps its `le` series
+ let buckets: Vec<(&str, f64)> = result
+ .iter()
+ .filter(|r| r["__name__"].as_str().unwrap().ends_with("_bucket"))
+ .map(|r| (r["le"].as_str().unwrap(), r["value"].as_f64().unwrap()))
+ .collect();
+ assert_eq!(
+ buckets,
+ vec![
+ ("-2", 0.0),
+ ("-1", 0.0),
+ ("0", 10.0),
+ ("1", 10.0),
+ ("2", 60.0),
+ ("4", 90.0),
+ ("8", 110.0),
+ ("inf", 110.0),
+ ]
+ );
}
#[test]
@@ -2608,6 +2640,19 @@ mod tests {
assert!(le_values.contains(&"10".to_string()));
}
+ fn exp_hist_buckets(result: &[serde_json::Value]) -> Vec<(String, f64)> {
+ result
+ .iter()
+ .filter(|r| r["__name__"].as_str().unwrap_or("").ends_with("_bucket"))
+ .map(|r| {
+ (
+ r["le"].as_str().unwrap().to_string(),
+ r["value"].as_f64().unwrap(),
+ )
+ })
+ .collect()
+ }
+
#[test]
fn test_exponential_histogram_buckets() {
let mut rec = json!({"__name__": "test_exp_histogram"});
@@ -2634,50 +2679,344 @@ mod tests {
}),
};
- let result = process_exp_hist_data_point(&mut rec, &data_point);
+ let result = process_exp_hist_data_point(&mut rec, &data_point, lim(16));
- // Should have count, sum, positive buckets, and negative buckets
- assert!(result.len() >= 2); // At least count and sum
+ assert_eq!(result[0]["__name__"], "test_exp_histogram_count");
+ assert_eq!(result[0]["value"], 200.0);
+ assert_eq!(result[1]["__name__"], "test_exp_histogram_sum");
+ assert_eq!(result[1]["value"], 500.0);
- // Check count record
- let count_record = result
+ let expected = [
+ ("-1", 0.0),
+ ("-0.7071", 10.0),
+ ("-0.5", 15.0),
+ ("-0.001", 15.0),
+ ("0.001", 20.0),
+ ("1", 20.0),
+ ("1.414", 30.0),
+ ("2", 50.0),
+ ("2.828", 80.0),
+ // `count` wins over a short bucket total so `le="inf"` equals `_count`
+ ("inf", 200.0),
+ ];
+ let expected: Vec<(String, f64)> = expected
.iter()
- .find(|r| r["__name__"].as_str().unwrap_or("").ends_with("_count"));
- assert!(count_record.is_some());
- assert_eq!(count_record.unwrap()["value"], 200.0);
-
- // Check sum record
- let sum_record = result
- .iter()
- .find(|r| r["__name__"].as_str().unwrap_or("").ends_with("_sum"));
- assert!(sum_record.is_some());
- assert_eq!(sum_record.unwrap()["value"], 500.0);
-
- // Bucket boundaries: base = 2^(2^-scale). scale=1 => base = 2^0.5 = sqrt(2).
- let base = 2f64.powf(2f64.powi(-1));
-
- // Positive buckets (offset 0): le = base^(idx+1) for idx = 0,1,2.
- let positive_les: Vec = result
- .iter()
- .filter(|r| r["__name__"].as_str().unwrap_or("").ends_with("_bucket"))
- .filter_map(|r| r["le"].as_str().and_then(|s| s.parse::().ok()))
- .filter(|le| *le > 0.0)
+ .map(|(le, v)| (le.to_string(), *v))
.collect();
- assert_eq!(positive_les.len(), 3);
- for (i, le) in positive_les.iter().enumerate() {
- assert!((le - base.powi(i as i32 + 1)).abs() < 1e-9);
+ assert_eq!(exp_hist_buckets(&result), expected);
+ }
+
+ #[test]
+ fn test_exponential_histogram_buckets_issue_14634() {
+ let mut rec = json!({"__name__": "h"});
+ let data_point = opentelemetry_proto::tonic::metrics::v1::ExponentialHistogramDataPoint {
+ attributes: vec![],
+ start_time_unix_nano: 0,
+ time_unix_nano: 1640995200000000000,
+ exemplars: vec![],
+ flags: 0,
+ count: 7,
+ sum: Some(21.0),
+ min: None,
+ max: None,
+ scale: 2,
+ zero_count: 1,
+ zero_threshold: 0.0,
+ positive: Some(opentelemetry_proto::tonic::metrics::v1::exponential_histogram_data_point::Buckets {
+ offset: 0,
+ bucket_counts: vec![2, 3],
+ }),
+ negative: Some(opentelemetry_proto::tonic::metrics::v1::exponential_histogram_data_point::Buckets {
+ offset: 0,
+ bucket_counts: vec![1],
+ }),
+ };
+
+ let result = process_exp_hist_data_point(&mut rec, &data_point, lim(16));
+
+ let expected = [
+ ("-1.189", 0.0),
+ ("-1", 1.0),
+ ("0", 2.0),
+ ("1", 2.0),
+ ("1.189", 4.0),
+ ("1.414", 7.0),
+ ("inf", 7.0),
+ ];
+ let expected: Vec<(String, f64)> = expected
+ .iter()
+ .map(|(le, v)| (le.to_string(), *v))
+ .collect();
+ assert_eq!(exp_hist_buckets(&result), expected);
+ }
+
+ #[test]
+ fn test_exponential_histogram_extreme_scales_still_expand() {
+ for scale in [i32::MAX, 21, -11, i32::MIN] {
+ let mut rec = json!({"__name__": "h"});
+ let data_point = opentelemetry_proto::tonic::metrics::v1::ExponentialHistogramDataPoint {
+ attributes: vec![],
+ start_time_unix_nano: 0,
+ time_unix_nano: 1640995200000000000,
+ exemplars: vec![],
+ flags: 0,
+ count: 1,
+ sum: Some(1.0),
+ min: None,
+ max: None,
+ scale,
+ zero_count: 0,
+ zero_threshold: 0.0,
+ positive: Some(opentelemetry_proto::tonic::metrics::v1::exponential_histogram_data_point::Buckets {
+ offset: 0,
+ bucket_counts: vec![1],
+ }),
+ negative: None,
+ };
+ let result = process_exp_hist_data_point(&mut rec, &data_point, lim(16));
+ assert_eq!(result[0]["__name__"], "h_count", "scale {scale}");
+ assert_eq!(result[1]["__name__"], "h_sum", "scale {scale}");
+ let buckets = exp_hist_buckets(&result);
+ assert_eq!(
+ buckets.last().unwrap(),
+ &("inf".to_string(), 1.0),
+ "scale {scale}"
+ );
+ for w in buckets.windows(2) {
+ let (a, b) = (
+ w[0].0.parse::().unwrap(),
+ w[1].0.parse::().unwrap(),
+ );
+ assert!(a < b && w[0].1 <= w[1].1, "scale {scale}: {buckets:?}");
+ }
}
+ }
- // Negative buckets (offset -2): le = -base^(idx+1) for idx = -2,-1.
- let negative_les: Vec = result
+ #[test]
+ fn test_exponential_histogram_clips_bucket_to_zero_threshold() {
+ let mut rec = json!({"__name__": "h"});
+ let data_point = opentelemetry_proto::tonic::metrics::v1::ExponentialHistogramDataPoint {
+ attributes: vec![],
+ start_time_unix_nano: 0,
+ time_unix_nano: 1640995200000000000,
+ exemplars: vec![],
+ flags: 0,
+ count: 10,
+ sum: Some(17.5),
+ min: None,
+ max: None,
+ scale: 0,
+ zero_count: 0,
+ zero_threshold: 1.5,
+ positive: Some(opentelemetry_proto::tonic::metrics::v1::exponential_histogram_data_point::Buckets {
+ offset: 0,
+ bucket_counts: vec![10],
+ }),
+ negative: None,
+ };
+
+ let result = process_exp_hist_data_point(&mut rec, &data_point, lim(16));
+
+ let expected = [("-1.5", 0.0), ("1.5", 0.0), ("2", 10.0), ("inf", 10.0)];
+ let expected: Vec<(String, f64)> = expected
.iter()
- .filter(|r| r["__name__"].as_str().unwrap_or("").ends_with("_bucket"))
- .filter_map(|r| r["le"].as_str().and_then(|s| s.parse::().ok()))
- .filter(|le| *le < 0.0)
+ .map(|(le, v)| (le.to_string(), *v))
.collect();
- assert_eq!(negative_les.len(), 2);
- assert!((negative_les[0] - (-base.powi(-1))).abs() < 1e-9);
- assert!((negative_les[1] - (-base.powi(0))).abs() < 1e-9);
+ assert_eq!(exp_hist_buckets(&result), expected);
+ }
+
+ #[test]
+ fn test_exponential_histogram_layout_stable_as_buckets_grow() {
+ // scale 0, 13 then 14 buckets: the old 16-label default downscaled the second sample
+ let sample = |counts: Vec| {
+ let mut rec = json!({"__name__": "h"});
+ let dp = opentelemetry_proto::tonic::metrics::v1::ExponentialHistogramDataPoint {
+ attributes: vec![],
+ start_time_unix_nano: 0,
+ time_unix_nano: 1640995200000000000,
+ exemplars: vec![],
+ flags: 0,
+ count: counts.iter().sum(),
+ sum: None,
+ min: None,
+ max: None,
+ scale: 0,
+ zero_count: 0,
+ zero_threshold: 0.0,
+ positive: Some(opentelemetry_proto::tonic::metrics::v1::exponential_histogram_data_point::Buckets {
+ offset: 0,
+ bucket_counts: counts,
+ }),
+ negative: None,
+ };
+ let limits = native_histogram::ExpansionLimits {
+ target_schema: 2,
+ max_buckets: 512,
+ };
+ exp_hist_buckets(&process_exp_hist_data_point(&mut rec, &dp, limits))
+ .into_iter()
+ .map(|(le, _)| le)
+ .collect::>()
+ };
+ let a = sample(vec![1; 13]);
+ let b = sample(vec![1; 14]);
+ assert_eq!(a.len(), 13 + 1 + 1 + 1);
+ assert_eq!(b.len(), 14 + 1 + 1 + 1);
+ assert!(a.iter().all(|le| b.contains(le)), "{a:?} vs {b:?}");
+ }
+
+ #[test]
+ fn test_exponential_histogram_downscales_fine_scale() {
+ let mut rec = json!({"__name__": "h"});
+ let data_point = opentelemetry_proto::tonic::metrics::v1::ExponentialHistogramDataPoint {
+ attributes: vec![],
+ start_time_unix_nano: 0,
+ time_unix_nano: 1640995200000000000,
+ exemplars: vec![],
+ flags: 0,
+ count: 3,
+ sum: Some(3.0),
+ min: None,
+ max: None,
+ scale: 10,
+ zero_count: 0,
+ zero_threshold: 0.0,
+ positive: Some(opentelemetry_proto::tonic::metrics::v1::exponential_histogram_data_point::Buckets {
+ offset: 0,
+ bucket_counts: vec![1, 1, 1],
+ }),
+ negative: None,
+ };
+
+ let result = process_exp_hist_data_point(&mut rec, &data_point, lim(16));
+
+ // scale 10 buckets 1..=3 merge into schema-8 bucket 1: (1, 2^(1/256)]
+ let expected = [("0", 0.0), ("1", 0.0), ("1.003", 3.0), ("inf", 3.0)];
+ let expected: Vec<(String, f64)> = expected
+ .iter()
+ .map(|(le, v)| (le.to_string(), *v))
+ .collect();
+ assert_eq!(exp_hist_buckets(&result), expected);
+ }
+
+ /// Explicitly reported empty buckets keep their `le` series before they fill.
+ #[test]
+ fn test_exponential_histogram_keeps_explicit_empty_buckets() {
+ let sample = |counts: Vec| {
+ let mut rec = json!({"__name__": "h"});
+ let data_point = opentelemetry_proto::tonic::metrics::v1::ExponentialHistogramDataPoint {
+ attributes: vec![],
+ start_time_unix_nano: 0,
+ time_unix_nano: 1640995200000000000,
+ exemplars: vec![],
+ flags: 0,
+ count: counts.iter().sum(),
+ sum: None,
+ min: None,
+ max: None,
+ scale: 0,
+ zero_count: 0,
+ zero_threshold: 0.0,
+ positive: Some(opentelemetry_proto::tonic::metrics::v1::exponential_histogram_data_point::Buckets {
+ offset: 0,
+ bucket_counts: counts,
+ }),
+ negative: None,
+ };
+ exp_hist_buckets(&process_exp_hist_data_point(&mut rec, &data_point, lim(16)))
+ };
+ let expect = |pairs: &[(&str, f64)]| -> Vec<(String, f64)> {
+ pairs.iter().map(|(le, v)| (le.to_string(), *v)).collect()
+ };
+
+ // (2,4] and (4,8] are empty but reported: no gap marker, cumulative unchanged
+ assert_eq!(
+ sample(vec![10, 0, 0, 10]),
+ expect(&[
+ ("0", 0.0),
+ ("1", 0.0),
+ ("2", 10.0),
+ ("4", 10.0),
+ ("8", 10.0),
+ ("16", 20.0),
+ ("inf", 20.0),
+ ])
+ );
+ assert_eq!(
+ sample(vec![10, 1, 0, 10]),
+ expect(&[
+ ("0", 0.0),
+ ("1", 0.0),
+ ("2", 10.0),
+ ("4", 11.0),
+ ("8", 11.0),
+ ("16", 21.0),
+ ("inf", 21.0),
+ ])
+ );
+ // leading and trailing zeros are kept as well
+ assert_eq!(
+ sample(vec![0, 5, 0]),
+ expect(&[
+ ("0", 0.0),
+ ("1", 0.0),
+ ("2", 0.0),
+ ("4", 5.0),
+ ("8", 5.0),
+ ("inf", 5.0),
+ ])
+ );
+ }
+
+ /// The zero bucket's `le` series exists before its first observation.
+ #[test]
+ fn test_exponential_histogram_zero_bucket_series_stable_across_samples() {
+ let sample = |zero_count: u64| {
+ let mut rec = json!({"__name__": "h"});
+ let data_point = opentelemetry_proto::tonic::metrics::v1::ExponentialHistogramDataPoint {
+ attributes: vec![],
+ start_time_unix_nano: 0,
+ time_unix_nano: 1640995200000000000,
+ exemplars: vec![],
+ flags: 0,
+ count: 10 + zero_count,
+ sum: None,
+ min: None,
+ max: None,
+ scale: 0,
+ zero_count,
+ zero_threshold: 0.001,
+ positive: Some(opentelemetry_proto::tonic::metrics::v1::exponential_histogram_data_point::Buckets {
+ offset: 0,
+ bucket_counts: vec![10],
+ }),
+ negative: None,
+ };
+ exp_hist_buckets(&process_exp_hist_data_point(&mut rec, &data_point, lim(16)))
+ };
+ let expect = |pairs: &[(&str, f64)]| -> Vec<(String, f64)> {
+ pairs.iter().map(|(le, v)| (le.to_string(), *v)).collect()
+ };
+ assert_eq!(
+ sample(0),
+ expect(&[
+ ("-0.001", 0.0),
+ ("0.001", 0.0),
+ ("1", 0.0),
+ ("2", 10.0),
+ ("inf", 10.0),
+ ])
+ );
+ assert_eq!(
+ sample(1),
+ expect(&[
+ ("-0.001", 0.0),
+ ("0.001", 1.0),
+ ("1", 1.0),
+ ("2", 11.0),
+ ("inf", 11.0),
+ ])
+ );
}
}
@@ -3022,7 +3361,7 @@ mod tests {
negative: None,
};
- let result = process_exp_hist_data_point(&mut rec, &data_point);
+ let result = process_exp_hist_data_point(&mut rec, &data_point, lim(16));
// Should have at least count and sum records
let metric_names: Vec<&str> = result
@@ -3181,6 +3520,7 @@ mod tests {
}),
negative: None,
},
+ lim(16),
)
}
@@ -3395,7 +3735,7 @@ mod tests {
negative: None,
};
- assert!(process_exp_hist_data_point(&mut rec, &stale).is_empty());
+ assert!(process_exp_hist_data_point(&mut rec, &stale, lim(16)).is_empty());
}
#[test]
diff --git a/src/core/src/metrics/prom.rs b/src/core/src/metrics/prom.rs
index 80a21afdf5..cf1fbb9611 100644
--- a/src/core/src/metrics/prom.rs
+++ b/src/core/src/metrics/prom.rs
@@ -53,7 +53,7 @@ use search_service;
use super::{
columnar, ingest,
- native_histogram::{CLASSIC_HISTOGRAM_SUFFIXES, expand_native_histogram},
+ native_histogram::{CLASSIC_HISTOGRAM_SUFFIXES, ExpansionLimits, expand_native_histogram},
prom_decode,
};
use crate::{
@@ -459,7 +459,7 @@ pub async fn remote_write(
&event.histograms,
&labels,
&metric_name,
- cfg.prom.native_histogram_max_buckets,
+ ExpansionLimits::from_config(&cfg),
&mut gate,
&mut sink,
)
@@ -1112,7 +1112,7 @@ async fn buffer_native_histograms(
histograms: &[prometheus_rpc::Histogram],
labels: &json::Map,
metric_name: &str,
- max_buckets: usize,
+ limits: ExpansionLimits,
gate: &mut HaGate<'_>,
sink: &mut RecordSink<'_>,
) -> Option {
@@ -1127,7 +1127,7 @@ async fn buffer_native_histograms(
let mut counted = 0;
for hp in histograms {
counted += 1;
- let records = expand_native_histogram(hp, max_buckets);
+ let records = expand_native_histogram(hp, limits);
if records.is_empty() {
// unsupported schema or stale marker: nothing will be written
continue;