diff --git a/Cargo.lock b/Cargo.lock index 046900dba..14392c506 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -74,9 +74,9 @@ checksum = "940b3a0ca603d1eade50a4846a2afffd5ef57a9feac2c0e2ec2e14f9ead76000" [[package]] name = "anyhow" -version = "1.0.102" +version = "1.0.103" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "7f202df86484c868dbad7eaa557ef785d5c66295e41b460ef922eca0723b842c" +checksum = "2a4385e2e34eb35d6b3efe798b9eb88096925d87726c0798709bf56d9ed84af3" [[package]] name = "approx" @@ -835,9 +835,9 @@ dependencies = [ [[package]] name = "crossbeam-epoch" -version = "0.9.18" +version = "0.9.20" source = "registry+https://github.com/rust-lang/crates.io-index" -checksum = "5b82ac4a3c2ca9c3460964f020e1402edd5753411d7737aa39c3714ad1b5420e" +checksum = "2d6914041f254d6e9176c01941b21115dcfb7089e55135a35411081bd106ef3f" dependencies = [ "crossbeam-utils", ] diff --git a/ci/fingerprints/otel_metrics/fingerprint.txt b/ci/fingerprints/otel_metrics/fingerprint.txt index 061d84011..7e5b84173 100644 --- a/ci/fingerprints/otel_metrics/fingerprint.txt +++ b/ci/fingerprints/otel_metrics/fingerprint.txt @@ -1 +1 @@ -otel_metrics_grpc: 1a99c0f4baf2470e547d0038dcaa6e7f134a1a31ae9dea50460e21f5e4e3a562 entropy=6.7343 +otel_metrics_grpc: ae0fe4374599f61afe94c09b3a92d9689b30ea49dc9c0f1be229abf8544153a6 entropy=6.7752 diff --git a/lading_payload/benches/opentelemetry_metric.rs b/lading_payload/benches/opentelemetry_metric.rs index 587ebeb3a..963ea5a78 100644 --- a/lading_payload/benches/opentelemetry_metric.rs +++ b/lading_payload/benches/opentelemetry_metric.rs @@ -20,6 +20,7 @@ fn opentelemetry_metric_setup(c: &mut Criterion) { gauge: 50, sum_delta: 25, sum_cumulative: 25, + ..Default::default() }, contexts: Contexts { total_contexts: ConfRange::Constant(100), @@ -49,6 +50,7 @@ fn opentelemetry_metric_throughput(c: &mut Criterion) { gauge: 50, sum_delta: 25, sum_cumulative: 25, + ..Default::default() }, contexts: Contexts { total_contexts: ConfRange::Constant(100), diff --git a/lading_payload/src/opentelemetry/common/templates.rs b/lading_payload/src/opentelemetry/common/templates.rs index 5ddbdbe6a..27625b187 100644 --- a/lading_payload/src/opentelemetry/common/templates.rs +++ b/lading_payload/src/opentelemetry/common/templates.rs @@ -96,6 +96,7 @@ where /// Check if any template in the pool can fit within the given budget. /// Returns true if at least one template fits, false otherwise. + #[cfg(test)] pub(crate) fn template_fits(&self, budget: usize) -> bool { // Check if any existing template in the pool fits self.by_size.range(..=budget).next().is_some() diff --git a/lading_payload/src/opentelemetry/metric.rs b/lading_payload/src/opentelemetry/metric.rs index 609aa5767..cbfffc858 100644 --- a/lading_payload/src/opentelemetry/metric.rs +++ b/lading_payload/src/opentelemetry/metric.rs @@ -28,9 +28,14 @@ // // * Gauge -- a collection of `NumberDataPoint`, values sampled at specific times // * Sum -- `Gauge` with the addition of monotonic flag, aggregation metadata -// -// We omit Histogram / ExponentialHistogram / Summary in this current version -// but will introduce them in the near-term. The `NumberDataPoint` is a +// * Histogram -- explicit-bucket histogram: each data point carries bucket counts +// aligned to a fixed set of explicit boundaries, plus count/sum. +// * ExponentialHistogram -- base-2 exponential bucket histogram: each data point +// carries positive/negative Buckets structs (offset + counts), zero_count, +// and a scale factor that governs bucket width. +// * Summary -- quantile summaries: each data point carries count/sum plus a +// sorted list of (quantile, value) pairs. +// The`NumberDataPoint` is a // // * attributes: Vec -- tags // * start_time_unix_nano: u64 -- represents the first possible moment a measurement could be recorded, optional @@ -41,6 +46,7 @@ pub(crate) mod templates; pub(crate) mod unit; +use opentelemetry_proto::tonic::metrics::v1::AggregationTemporality; use rand::RngExt; use std::rc::Rc; use std::{cell::RefCell, io::Write}; @@ -50,7 +56,10 @@ use crate::opentelemetry::common::templates::PoolError; use crate::{Error, common::config::ConfRange, common::strings}; use bytes::BytesMut; use opentelemetry_proto::tonic::collector::metrics::v1::ExportMetricsServiceRequest; -use opentelemetry_proto::tonic::metrics::v1::{ResourceMetrics, metric::Data, number_data_point}; +use opentelemetry_proto::tonic::metrics::v1::{ + ExponentialHistogramDataPoint, HistogramDataPoint, ResourceMetrics, + exponential_histogram_data_point, metric::Data, number_data_point, +}; use prost::Message; use serde::Deserialize; use templates::{Pool, ResourceTemplateGenerator}; @@ -64,6 +73,113 @@ pub const SMALLEST_PROTOBUF: usize = 31; /// Increment timestamps by 100 milliseconds (in nanoseconds) per tick const TIME_INCREMENT_NANOS: u64 = 1_000_000; +fn increment_exp_bucket_counts( + buckets: &mut exponential_histogram_data_point::Buckets, + increment: u64, +) -> u64 { + for count in &mut buckets.bucket_counts { + *count = count.saturating_add(increment); + } + buckets.bucket_counts.iter().sum() +} + +fn randomize_exp_bucket_counts( + buckets: &mut exponential_histogram_data_point::Buckets, + rng: &mut R, +) -> u64 +where + R: rand::Rng + ?Sized, +{ + for count in &mut buckets.bucket_counts { + *count = rng.random_range(0_u64..=10); + } + buckets.bucket_counts.iter().sum() +} + +fn populated_bucket_range(bucket_counts: &[u64]) -> Option<(usize, usize)> { + let first = bucket_counts.iter().position(|count| *count > 0)?; + let last = bucket_counts.iter().rposition(|count| *count > 0)?; + Some((first, last)) +} + +fn histogram_bucket_bounds(point: &HistogramDataPoint, bucket: usize) -> (f64, f64) { + let lower = if bucket == 0 { + 0.0 + } else { + point.explicit_bounds[bucket - 1] + }; + let upper = point + .explicit_bounds + .get(bucket) + .copied() + .unwrap_or_else(|| point.explicit_bounds.last().copied().unwrap_or(1.0) * 2.0); + (lower, upper) +} + +fn set_histogram_min_max(point: &mut HistogramDataPoint, rng: &mut R) +where + R: rand::Rng + ?Sized, +{ + let Some((first, last)) = populated_bucket_range(&point.bucket_counts) else { + point.min = Some(0.0); + point.max = Some(0.0); + return; + }; + let (min_lower, min_upper) = histogram_bucket_bounds(point, first); + let (max_lower, max_upper) = histogram_bucket_bounds(point, last); + let min = rng.random_range(min_lower..=min_upper); + let max = rng.random_range(max_lower..=max_upper).max(min); + point.min = Some(min); + point.max = Some(max); +} + +fn exp_histogram_bucket_bounds(scale: i32, index: i32) -> (f64, f64) { + let base = 2.0_f64.powf(2.0_f64.powi(-scale)); + (base.powi(index), base.powi(index + 1)) +} + +fn exp_histogram_populated_index_range( + buckets: &exponential_histogram_data_point::Buckets, +) -> Option<(i32, i32)> { + let (first, last) = populated_bucket_range(&buckets.bucket_counts)?; + let first = i32::try_from(first).ok()?; + let last = i32::try_from(last).ok()?; + Some(( + buckets.offset.checked_add(first)?, + buckets.offset.checked_add(last)?, + )) +} + +fn set_exp_histogram_min_max(point: &mut ExponentialHistogramDataPoint, rng: &mut R) +where + R: rand::Rng + ?Sized, +{ + let positive_range = point + .positive + .as_ref() + .and_then(exp_histogram_populated_index_range); + let negative_range = point + .negative + .as_ref() + .and_then(exp_histogram_populated_index_range); + + let min = if let Some((_, last)) = negative_range { + let (lower, upper) = exp_histogram_bucket_bounds(point.scale, last); + rng.random_range(-upper..=-lower) + } else { + 0.0 + }; + let max = if let Some((_, last)) = positive_range { + let (lower, upper) = exp_histogram_bucket_bounds(point.scale, last); + rng.random_range(lower..=upper) + } else { + 0.0 + }; + + point.min = Some(min.min(max)); + point.max = Some(max.max(min)); +} + /// Configure the OpenTelemetry metric payload. #[derive(Debug, Deserialize, serde::Serialize, Clone, PartialEq, Copy)] #[cfg_attr(feature = "arbitrary", derive(arbitrary::Arbitrary))] @@ -110,14 +226,45 @@ pub struct MetricWeights { pub sum_delta: u8, /// The relative probability of generating a sum cumulative metric. pub sum_cumulative: u8, + /// The relative probability of generating a delta histogram metric. + pub histogram_delta: u8, + /// The relative probability of generating a cumulative histogram metric. + pub histogram_cumulative: u8, + /// The relative probability of generating a delta exponential histogram metric. + pub exp_histogram_delta: u8, + /// The relative probability of generating a cumulative exponential histogram metric. + pub exp_histogram_cumulative: u8, + /// The relative probability of generating a summary metric. + pub summary: u8, +} + +impl MetricWeights { + fn has_zero_weight(self) -> bool { + [ + self.gauge, + self.sum_delta, + self.sum_cumulative, + self.histogram_delta, + self.histogram_cumulative, + self.exp_histogram_delta, + self.exp_histogram_cumulative, + self.summary, + ] + .contains(&0) + } } impl Default for MetricWeights { fn default() -> Self { Self { - gauge: 50, // 50% - sum_delta: 25, // 25% - sum_cumulative: 25, // 25% + gauge: 30, + sum_delta: 15, + sum_cumulative: 15, + histogram_delta: 10, + histogram_cumulative: 10, + exp_histogram_delta: 5, + exp_histogram_cumulative: 5, + summary: 10, } } } @@ -140,12 +287,10 @@ impl Config { /// Function will error if the configuration is invalid #[expect(clippy::too_many_lines)] pub fn valid(&self) -> Result<(), String> { - // Validate metric weights - both types must have non-zero probability to ensure - // we can generate a diverse set of metrics - if self.metric_weights.gauge == 0 - || self.metric_weights.sum_delta == 0 - || self.metric_weights.sum_cumulative == 0 - { + // Validate metric weights: every supported metric type must have + // non-zero probability so generated payloads cover the full OTLP metric + // surface area. + if self.metric_weights.has_zero_weight() { return Err("Metric weights cannot be 0".to_string()); } @@ -332,14 +477,16 @@ impl OpentelemetryMetrics { } } +#[expect(clippy::too_many_lines)] impl<'a> SizedGenerator<'a> for OpentelemetryMetrics { type Output = ResourceMetrics; type Error = Error; /// Generate OTLP metrics with the following enhancements: /// - /// * Monotonic sums are truly monotonic, incrementing by a random amount - /// each tick + /// * Cumulative sums evolve from their prior value each tick; monotonic + /// cumulative sums only increase while non-monotonic cumulative sums may + /// move in either direction /// * Timestamps advance monotonically based on internal tick counter /// starting at epoch /// * Each call advances the tick counter by a random amount (1-60) @@ -347,6 +494,8 @@ impl<'a> SizedGenerator<'a> for OpentelemetryMetrics { where R: rand::Rng + ?Sized, { + let original_budget = *budget; + self.tick += rng.random_range(1..=60); self.incr_f += rng.random_range(1.0..=100.0); self.incr_i += rng.random_range(1_i64..=100_i64); @@ -387,32 +536,34 @@ impl<'a> SizedGenerator<'a> for OpentelemetryMetrics { } Data::Sum(sum) => { data_points_count += sum.data_points.len() as u64; - let is_accumulating = sum.is_monotonic; + let is_cumulative = sum.aggregation_temporality + == AggregationTemporality::Cumulative as i32; for point in &mut sum.data_points { point.time_unix_nano = self.tick * TIME_INCREMENT_NANOS; - if is_accumulating { - // For accumulating sums, monotonically - // increase by some factor of `tick_diff` - if let Some(value) = &mut point.value { - match value { - number_data_point::Value::AsDouble(v) => { - *v += self.incr_f; - } - #[allow(clippy::cast_possible_wrap)] - number_data_point::Value::AsInt(v) => { - *v += self.incr_i; - } - } - } - } else { - // For non-accumulating sums, use random - // values - if let Some(value) = &mut point.value { - match value { - number_data_point::Value::AsDouble(v) => { + if let Some(value) = &mut point.value { + match value { + number_data_point::Value::AsDouble(v) => { + if is_cumulative { + if sum.is_monotonic { + *v += self.incr_f; + } else { + *v += rng + .random_range(-self.incr_f..=self.incr_f); + } + } else { *v = rng.random(); } - number_data_point::Value::AsInt(v) => { + } + #[allow(clippy::cast_possible_wrap)] + number_data_point::Value::AsInt(v) => { + if is_cumulative { + if sum.is_monotonic { + *v += self.incr_i; + } else { + *v += rng + .random_range(-self.incr_i..=self.incr_i); + } + } else { *v = rng.random(); } } @@ -420,12 +571,129 @@ impl<'a> SizedGenerator<'a> for OpentelemetryMetrics { } } } - _ => unimplemented!(), + Data::Histogram(hist) => { + data_points_count += hist.data_points.len() as u64; + let is_cumulative = hist.aggregation_temporality + == AggregationTemporality::Cumulative as i32; + for point in &mut hist.data_points { + point.time_unix_nano = self.tick * TIME_INCREMENT_NANOS; + if is_cumulative { + let previous_min = point.min; + let previous_max = point.max; + // Cumulative: bucket_counts and sum only + // grow, mirroring monotonic sum behavior. + // incr_i always positive so cast is safe. + #[allow(clippy::cast_sign_loss)] + let increment = self.incr_i as u64; + for bc in &mut point.bucket_counts { + *bc = bc.saturating_add(increment); + } + point.count = point.bucket_counts.iter().sum(); + set_histogram_min_max(point, rng); + if let (Some(previous), Some(current)) = + (previous_min, point.min.as_mut()) + { + *current = current.min(previous); + } + if let (Some(previous), Some(current)) = + (previous_max, point.max.as_mut()) + { + *current = current.max(previous); + } + if let Some(sum) = &mut point.sum { + *sum += self.incr_f; + } + } else { + // Delta: fresh observations each window. + // Use 1..=10 (not 0..=10) so count stays + // non-zero and the encoded size is stable. + for bc in &mut point.bucket_counts { + *bc = rng.random_range(1_u64..=10); + } + point.count = point.bucket_counts.iter().sum(); + set_histogram_min_max(point, rng); + if let Some(sum) = &mut point.sum { + *sum = rng.random_range(1.0_f64..=1000.0); + } + } + } + } + Data::ExponentialHistogram(exp_hist) => { + data_points_count += exp_hist.data_points.len() as u64; + let is_cumulative = exp_hist.aggregation_temporality + == AggregationTemporality::Cumulative as i32; + for point in &mut exp_hist.data_points { + point.time_unix_nano = self.tick * TIME_INCREMENT_NANOS; + if is_cumulative { + let previous_min = point.min; + let previous_max = point.max; + // Cumulative: zero, positive, and negative + // bucket counts only grow. + #[allow(clippy::cast_sign_loss)] + let increment = self.incr_i as u64; + point.zero_count = point.zero_count.saturating_add(increment); + let mut count = point.zero_count; + if let Some(positive) = &mut point.positive { + count += increment_exp_bucket_counts(positive, increment); + } + if let Some(negative) = &mut point.negative { + count += increment_exp_bucket_counts(negative, increment); + } + point.count = count; + set_exp_histogram_min_max(point, rng); + if let (Some(previous), Some(current)) = + (previous_min, point.min.as_mut()) + { + *current = current.min(previous); + } + if let (Some(previous), Some(current)) = + (previous_max, point.max.as_mut()) + { + *current = current.max(previous); + } + if let Some(sum) = &mut point.sum { + *sum += self.incr_f; + } + } else { + // Delta: fresh observations each window. + point.zero_count = rng.random_range(1_u64..=10); + let mut count = point.zero_count; + if let Some(positive) = &mut point.positive { + count += randomize_exp_bucket_counts(positive, rng); + } + if let Some(negative) = &mut point.negative { + count += randomize_exp_bucket_counts(negative, rng); + } + point.count = count; + set_exp_histogram_min_max(point, rng); + if let Some(sum) = &mut point.sum { + *sum = rng.random_range(1.0_f64..=1000.0); + } + } + } + } + Data::Summary(summary) => { + data_points_count += summary.data_points.len() as u64; + for point in &mut summary.data_points { + point.time_unix_nano = self.tick * TIME_INCREMENT_NANOS; + } + } } } } } + let required_bytes = tpl.encoded_len(); + if required_bytes > original_budget { + *budget = original_budget; + debug!( + ?required_bytes, + ?original_budget, + "Generated metric exceeded request budget" + ); + Err(PoolError::EmptyChoice)?; + } + *budget = original_budget - required_bytes; self.data_points_per_resource = data_points_count; Ok(tpl) @@ -495,13 +763,9 @@ impl crate::Serialize for OpentelemetryMetrics { } bytes_remaining = max_bytes.saturating_sub(required_bytes); } else { - // Belt with suspenders time: verify no templates could possibly - // fit. If we pass this assertion, break as no template will - // ever fit the requested max_bytes. - assert!( - !self.pool.template_fits(bytes_remaining), - "Pool claims template fits {bytes_remaining} bytes but generate() failed, indicative of a logic error", - ); + // A template may fit its cached size but exceed the remaining + // budget after live fields are updated. Stop rather than + // asserting on the pre-update pool size. break; } } @@ -1007,7 +1271,30 @@ mod test { .push(point.time_unix_nano); } }, - _ => {}, + Data::Histogram(histogram) => { + for point in &histogram.data_points { + timestamps_by_metric + .entry(id) + .or_default() + .push(point.time_unix_nano); + } + }, + Data::ExponentialHistogram(exp_histogram) => { + for point in &exp_histogram.data_points { + timestamps_by_metric + .entry(id) + .or_default() + .push(point.time_unix_nano); + } + }, + Data::Summary(summary) => { + for point in &summary.data_points { + timestamps_by_metric + .entry(id) + .or_default() + .push(point.time_unix_nano); + } + }, } } } @@ -1033,6 +1320,196 @@ mod test { } } + #[test] + #[expect(clippy::too_many_lines)] + fn data_points_include_attributes() -> Result<(), crate::Error> { + let configs = [ + super::MetricWeights { + gauge: 1, + sum_delta: 0, + sum_cumulative: 0, + histogram_delta: 0, + histogram_cumulative: 0, + exp_histogram_delta: 0, + exp_histogram_cumulative: 0, + summary: 0, + }, + super::MetricWeights { + gauge: 0, + sum_delta: 1, + sum_cumulative: 1, + histogram_delta: 0, + histogram_cumulative: 0, + exp_histogram_delta: 0, + exp_histogram_cumulative: 0, + summary: 0, + }, + super::MetricWeights { + gauge: 0, + sum_delta: 0, + sum_cumulative: 0, + histogram_delta: 1, + histogram_cumulative: 1, + exp_histogram_delta: 0, + exp_histogram_cumulative: 0, + summary: 0, + }, + super::MetricWeights { + gauge: 0, + sum_delta: 0, + sum_cumulative: 0, + histogram_delta: 0, + histogram_cumulative: 0, + exp_histogram_delta: 1, + exp_histogram_cumulative: 1, + summary: 0, + }, + super::MetricWeights { + gauge: 0, + sum_delta: 0, + sum_cumulative: 0, + histogram_delta: 0, + histogram_cumulative: 0, + exp_histogram_delta: 0, + exp_histogram_cumulative: 0, + summary: 1, + }, + ]; + + for metric_weights in configs { + let config = Config { + contexts: Contexts { + total_contexts: ConfRange::Constant(10), + attributes_per_resource: ConfRange::Constant(1), + scopes_per_resource: ConfRange::Constant(1), + attributes_per_scope: ConfRange::Constant(0), + metrics_per_scope: ConfRange::Constant(4), + attributes_per_metric: ConfRange::Constant(1), + }, + metric_weights, + }; + let mut budget = 1_000_000; + let mut rng = SmallRng::seed_from_u64(42); + let mut otel_metrics = OpentelemetryMetrics::new(config, budget, &mut rng)?; + let resource_metric = otel_metrics.generate(&mut rng, &mut budget)?; + + for scope_metric in &resource_metric.scope_metrics { + for metric in &scope_metric.metrics { + match &metric.data { + Some(Data::Gauge(gauge)) => { + for point in &gauge.data_points { + assert!(!point.attributes.is_empty()); + } + } + Some(Data::Sum(sum)) => { + for point in &sum.data_points { + assert!(!point.attributes.is_empty()); + } + } + Some(Data::Histogram(histogram)) => { + for point in &histogram.data_points { + assert!(!point.attributes.is_empty()); + } + } + Some(Data::ExponentialHistogram(exp_histogram)) => { + for point in &exp_histogram.data_points { + assert!(!point.attributes.is_empty()); + } + } + Some(Data::Summary(summary)) => { + for point in &summary.data_points { + assert!(!point.attributes.is_empty()); + } + } + None => {} + } + } + } + } + + Ok(()) + } + + #[test] + fn histograms_populate_min_max() -> Result<(), crate::Error> { + let configs = [ + super::MetricWeights { + gauge: 0, + sum_delta: 0, + sum_cumulative: 0, + histogram_delta: 1, + histogram_cumulative: 1, + exp_histogram_delta: 0, + exp_histogram_cumulative: 0, + summary: 0, + }, + super::MetricWeights { + gauge: 0, + sum_delta: 0, + sum_cumulative: 0, + histogram_delta: 0, + histogram_cumulative: 0, + exp_histogram_delta: 1, + exp_histogram_cumulative: 1, + summary: 0, + }, + ]; + + for metric_weights in configs { + let config = Config { + contexts: Contexts { + total_contexts: ConfRange::Constant(10), + attributes_per_resource: ConfRange::Constant(1), + scopes_per_resource: ConfRange::Constant(1), + attributes_per_scope: ConfRange::Constant(0), + metrics_per_scope: ConfRange::Constant(8), + attributes_per_metric: ConfRange::Constant(0), + }, + metric_weights, + }; + let mut budget = 1_000_000; + let mut rng = SmallRng::seed_from_u64(42); + let mut otel_metrics = OpentelemetryMetrics::new(config, budget, &mut rng)?; + let resource_metric = otel_metrics.generate(&mut rng, &mut budget)?; + + for scope_metric in &resource_metric.scope_metrics { + for metric in &scope_metric.metrics { + match &metric.data { + Some(Data::Histogram(histogram)) => { + for point in &histogram.data_points { + assert!(point.min.is_some()); + assert!(point.max.is_some()); + assert!(point.min <= point.max); + } + } + Some(Data::ExponentialHistogram(exp_histogram)) => { + for point in &exp_histogram.data_points { + let positive_count = point + .positive + .as_ref() + .map_or(0, |buckets| buckets.bucket_counts.iter().sum()); + let negative_count = point + .negative + .as_ref() + .map_or(0, |buckets| buckets.bucket_counts.iter().sum()); + assert_eq!( + point.count, + point.zero_count + positive_count + negative_count + ); + assert!(point.min.is_some()); + assert!(point.max.is_some()); + assert!(point.min <= point.max); + } + } + _ => {} + } + } + } + } + + Ok(()) + } + // Property: tick tally in OpentelemetryMetrics increase with calls to // `generate`. proptest! { @@ -1060,6 +1537,11 @@ mod test { gauge: 0, // Only generate sum metrics sum_delta: 50, sum_cumulative: 50, + histogram_delta: 0, + histogram_cumulative: 0, + exp_histogram_delta: 0, + exp_histogram_cumulative: 0, + summary: 0, }, }; @@ -1100,6 +1582,11 @@ mod test { gauge: 0, // Only generate sum metrics sum_delta: 50, sum_cumulative: 50, + histogram_delta: 0, + histogram_cumulative: 0, + exp_histogram_delta: 0, + exp_histogram_cumulative: 0, + summary: 0, }, }; @@ -1345,6 +1832,7 @@ mod test { gauge: 0, sum_delta: 25, sum_cumulative: 25, + ..Default::default() }, ..valid_config }; @@ -1355,16 +1843,47 @@ mod test { gauge: 50, sum_delta: 0, sum_cumulative: 0, + ..Default::default() }, ..valid_config }; assert!(zero_sum_weight.valid().is_err()); + let zero_histogram_weight = Config { + metric_weights: super::MetricWeights { + histogram_delta: 0, + histogram_cumulative: 0, + ..Default::default() + }, + ..valid_config + }; + assert!(zero_histogram_weight.valid().is_err()); + + let zero_exp_histogram_weight = Config { + metric_weights: super::MetricWeights { + exp_histogram_delta: 0, + exp_histogram_cumulative: 0, + ..Default::default() + }, + ..valid_config + }; + assert!(zero_exp_histogram_weight.valid().is_err()); + + let zero_summary_weight = Config { + metric_weights: super::MetricWeights { + summary: 0, + ..Default::default() + }, + ..valid_config + }; + assert!(zero_summary_weight.valid().is_err()); + let zero_weights = Config { metric_weights: super::MetricWeights { gauge: 0, sum_delta: 0, sum_cumulative: 0, + ..Default::default() }, ..valid_config }; diff --git a/lading_payload/src/opentelemetry/metric/templates.rs b/lading_payload/src/opentelemetry/metric/templates.rs index 997ef9213..daaf2779a 100644 --- a/lading_payload/src/opentelemetry/metric/templates.rs +++ b/lading_payload/src/opentelemetry/metric/templates.rs @@ -1,11 +1,16 @@ -use std::{cmp, rc::Rc}; +use std::{ + cmp::{self, Ordering}, + rc::Rc, +}; use opentelemetry_proto::tonic::{ - common::v1::InstrumentationScope, + common::v1::{InstrumentationScope, KeyValue}, metrics::{ self, v1::{ - Metric, NumberDataPoint, ResourceMetrics, ScopeMetrics, metric::Data, number_data_point, + ExponentialHistogramDataPoint, HistogramDataPoint, Metric, NumberDataPoint, + ResourceMetrics, ScopeMetrics, SummaryDataPoint, exponential_histogram_data_point, + metric::Data, number_data_point, summary_data_point, }, }, resource, @@ -13,7 +18,7 @@ use opentelemetry_proto::tonic::{ use prost::Message; use rand::{ Rng, RngExt, - distr::{Distribution, StandardUniform, weighted::WeightedIndex}, + distr::{Distribution, weighted::WeightedIndex}, }; use tracing::debug; @@ -51,33 +56,23 @@ fn exponential_weighted_range(rng: &mut R, min: u32, max: u32) max } -struct Ndp(NumberDataPoint); -impl Distribution for StandardUniform { - fn sample(&self, rng: &mut R) -> Ndp - where - R: Rng + ?Sized, - { - let value = match rng.random_range(0..=1) { - 0 => number_data_point::Value::AsDouble(0.0), - 1 => number_data_point::Value::AsInt(0), - _ => unreachable!(), - }; +fn random_number_data_point(rng: &mut R, attributes: &[KeyValue]) -> NumberDataPoint +where + R: Rng + ?Sized, +{ + let value = match rng.random_range(0..=1) { + 0 => number_data_point::Value::AsDouble(0.0), + 1 => number_data_point::Value::AsInt(0), + _ => unreachable!(), + }; - Ndp(NumberDataPoint { - // NOTE absent a reason to set attributes to not-empty, it's unclear - // that we should. - attributes: Vec::new(), - start_time_unix_nano: 0, // epoch instant - time_unix_nano: rng.random(), - // Unclear that this needs to be set. - exemplars: Vec::new(), - // Equivalent to DoNotUse, the flag is ignored. This is discussed in - // the upstream OTLP protobuf definition, which we inherit from the - // SDK. If we ever set `value` to None this must be set to - // DATA_POINT_FLAGS_NO_RECORDED_VALUE_MASK - flags: 0, - value: Some(value), - }) + NumberDataPoint { + attributes: attributes.to_vec(), + start_time_unix_nano: 0, + time_unix_nano: rng.random(), + exemplars: Vec::new(), + flags: 0, + value: Some(value), } } @@ -112,6 +107,11 @@ impl MetricTemplateGenerator { u16::from(config.metric_weights.gauge), u16::from(config.metric_weights.sum_delta), u16::from(config.metric_weights.sum_cumulative), + u16::from(config.metric_weights.histogram_delta), + u16::from(config.metric_weights.histogram_cumulative), + u16::from(config.metric_weights.exp_histogram_delta), + u16::from(config.metric_weights.exp_histogram_cumulative), + u16::from(config.metric_weights.summary), ])?, unit_gen: UnitGenerator::new(), str_pool: Rc::clone(str_pool), @@ -120,6 +120,7 @@ impl MetricTemplateGenerator { } } +#[expect(clippy::too_many_lines)] impl<'a> crate::SizedGenerator<'a> for MetricTemplateGenerator { type Output = Metric; type Error = GeneratorError; @@ -146,7 +147,7 @@ impl<'a> crate::SizedGenerator<'a> for MetricTemplateGenerator { Err(e) => Err(e)?, }; - let name = self + let name_suffix = self .str_pool .of_size_range(rng, 1_u8..16) .ok_or(Self::Error::StringGenerate)? @@ -175,24 +176,74 @@ impl<'a> crate::SizedGenerator<'a> for MetricTemplateGenerator { aggregation_temporality: 2, is_monotonic: rng.random_bool(0.5), }, + 3 => Kind::Histogram { + aggregation_temporality: 1, + }, + 4 => Kind::Histogram { + aggregation_temporality: 2, + }, + 5 => Kind::ExponentialHistogram { + aggregation_temporality: 1, + }, + 6 => Kind::ExponentialHistogram { + aggregation_temporality: 2, + }, + 7 => Kind::Summary, _ => unreachable!(), }; + let prefix = kind.name_prefix(); + let name = format!("{prefix}_{name_suffix}"); // Use weighted distribution: heavily favors small numbers (1-2) but can go up to 60 let total_data_points = exponential_weighted_range(rng, 1, 60); - let data_points = (0..total_data_points) - .map(|_| rng.random::().0) - .collect(); let data = match kind { - Kind::Gauge => Data::Gauge(metrics::v1::Gauge { data_points }), + Kind::Gauge => { + let data_points = (0..total_data_points) + .map(|_| random_number_data_point(rng, &metadata)) + .collect(); + Data::Gauge(metrics::v1::Gauge { data_points }) + } Kind::Sum { aggregation_temporality, is_monotonic, - } => Data::Sum(metrics::v1::Sum { - data_points, + } => { + let data_points = (0..total_data_points) + .map(|_| random_number_data_point(rng, &metadata)) + .collect(); + Data::Sum(metrics::v1::Sum { + data_points, + aggregation_temporality, + is_monotonic, + }) + } + Kind::Histogram { aggregation_temporality, - is_monotonic, - }), + } => { + let data_points = (0..total_data_points) + .map(|_| random_histogram_data_point(rng, &metadata)) + .collect(); + Data::Histogram(metrics::v1::Histogram { + data_points, + aggregation_temporality, + }) + } + Kind::ExponentialHistogram { + aggregation_temporality, + } => { + let data_points = (0..total_data_points) + .map(|_| random_exp_histogram_data_point(rng, &metadata)) + .collect(); + Data::ExponentialHistogram(metrics::v1::ExponentialHistogram { + data_points, + aggregation_temporality, + }) + } + Kind::Summary => { + let data_points = (0..total_data_points) + .map(|_| random_summary_data_point(rng, &metadata)) + .collect(); + Data::Summary(metrics::v1::Summary { data_points }) + } }; let mut metric = Metric { name, @@ -224,14 +275,17 @@ impl<'a> crate::SizedGenerator<'a> for MetricTemplateGenerator { } fn data_points_total(metric: &Metric) -> usize { - let data = &metric.data; - match data { + match &metric.data { Some( Data::Gauge(metrics::v1::Gauge { data_points }) | Data::Sum(metrics::v1::Sum { data_points, .. }), ) => data_points.len(), + Some(Data::Histogram(metrics::v1::Histogram { data_points, .. })) => data_points.len(), + Some(Data::ExponentialHistogram(metrics::v1::ExponentialHistogram { + data_points, .. + })) => data_points.len(), + Some(Data::Summary(metrics::v1::Summary { data_points })) => data_points.len(), None => 0, - _ => unimplemented!("only gauge/sum metrics supported"), } } @@ -261,8 +315,36 @@ fn cut_data_points(metric: Metric) -> Metric { is_monotonic, })) } + Some(Data::Histogram(metrics::v1::Histogram { + mut data_points, + aggregation_temporality, + })) => { + let new_len = data_points.len() / 2; + data_points.truncate(new_len); + Some(Data::Histogram(metrics::v1::Histogram { + data_points, + aggregation_temporality, + })) + } + Some(Data::ExponentialHistogram(metrics::v1::ExponentialHistogram { + mut data_points, + aggregation_temporality, + })) => { + let new_len = data_points.len() / 2; + data_points.truncate(new_len); + Some(Data::ExponentialHistogram( + metrics::v1::ExponentialHistogram { + data_points, + aggregation_temporality, + }, + )) + } + Some(Data::Summary(metrics::v1::Summary { mut data_points })) => { + let new_len = data_points.len() / 2; + data_points.truncate(new_len); + Some(Data::Summary(metrics::v1::Summary { data_points })) + } None => None, - _ => unimplemented!("only gauge/sum metrics supported"), }; Metric { @@ -281,6 +363,150 @@ pub(crate) enum Kind { aggregation_temporality: i32, is_monotonic: bool, }, + Histogram { + aggregation_temporality: i32, + }, + ExponentialHistogram { + aggregation_temporality: i32, + }, + Summary, +} + +impl Kind { + fn name_prefix(self) -> &'static str { + match self { + Self::Gauge => "gauge", + Self::Sum { + aggregation_temporality: 1, + .. + } => "sum_delta", + Self::Sum { + aggregation_temporality: 2, + .. + } => "sum_cumulative", + Self::Sum { .. } => "sum", + Self::Histogram { + aggregation_temporality: 1, + } => "histogram_delta", + Self::Histogram { + aggregation_temporality: 2, + } => "histogram_cumulative", + Self::Histogram { .. } => "histogram", + Self::ExponentialHistogram { + aggregation_temporality: 1, + } => "exp_histogram_delta", + Self::ExponentialHistogram { + aggregation_temporality: 2, + } => "exp_histogram_cumulative", + Self::ExponentialHistogram { .. } => "exp_histogram", + Self::Summary => "summary", + } + } +} + +/// Construct an explicit-bucket `HistogramDataPoint` scaffold. +/// +/// The bucket count is fixed so generated payload size stays predictable, but +/// the explicit bounds are randomized to avoid repeating the same histogram +/// shape across templates. The live state is updated each tick in `generate()`. +fn random_histogram_data_point( + rng: &mut R, + attributes: &[KeyValue], +) -> HistogramDataPoint { + let num_of_bounds = rng.random_range(3_usize..=8); + let mut explicit_bounds = Vec::with_capacity(num_of_bounds); + let mut bound = rng.random_range(1.0_f64..=1000.0); + for _ in 0..num_of_bounds { + explicit_bounds.push(bound); + bound += rng.random_range(1.0_f64..=1000.0); + } + let n_buckets = explicit_bounds.len() + 1; + HistogramDataPoint { + attributes: attributes.to_vec(), + start_time_unix_nano: 1, + time_unix_nano: 1, + count: n_buckets as u64, + sum: Some(0.0), + bucket_counts: vec![1; n_buckets], + explicit_bounds, + exemplars: Vec::new(), + flags: 0, + min: Some(0.0), + max: Some(0.0), + } +} + +/// Construct an `ExponentialHistogramDataPoint` scaffold. +/// +/// The template uses fixed bucket counts for positive and negative ranges so +/// generated payload size stays predictable. `generate()` later adjusts the +/// per-bucket counts while preserving the same bucket layout. +fn random_exp_histogram_data_point( + rng: &mut R, + attributes: &[KeyValue], +) -> ExponentialHistogramDataPoint { + let scale: i32 = rng.random_range(-3_i32..=3); + + let positive_bucket_count = rng.random_range(1_usize..=100); + let negative_bucket_count = rng.random_range(1_usize..=100); + let zero_data_count = rng.random_range(1_usize..=6); + + let positive_offset = rng.random_range(0_i32..=10); + let negative_offset = rng.random_range(0_i32..=10); + let bucket_count = positive_bucket_count + negative_bucket_count + zero_data_count; + ExponentialHistogramDataPoint { + attributes: attributes.to_vec(), + start_time_unix_nano: 1, + time_unix_nano: 1, + count: bucket_count as u64, + sum: Some(0.0), + scale, + zero_count: zero_data_count as u64, + positive: Some(exponential_histogram_data_point::Buckets { + offset: positive_offset, + bucket_counts: vec![1; positive_bucket_count], + }), + negative: Some(exponential_histogram_data_point::Buckets { + offset: negative_offset, + bucket_counts: vec![1; negative_bucket_count], + }), + flags: 0, + exemplars: Vec::new(), + min: Some(0.0), + max: Some(0.0), + zero_threshold: 0.0, + } +} + +/// Generate a random `SummaryDataPoint`. +/// +/// Produces five standard quantiles (0.0, 0.5, 0.9, 0.99, 1.0). Values are +/// drawn randomly and sorted so they are monotonically non-decreasing, matching +/// the convention that higher quantiles carry equal or greater values. +fn random_summary_data_point( + rng: &mut R, + attributes: &[KeyValue], +) -> SummaryDataPoint { + let count: u64 = rng.random_range(1_u64..=1_000_000); + let sum: f64 = rng.random_range(0.0_f64..=1_000_000.0); + let mut raw: Vec = (0..5) + .map(|_| rng.random_range(0.0_f64..=1_000_000.0)) + .collect(); + raw.sort_by(|a, b| a.partial_cmp(b).unwrap_or(Ordering::Equal)); + let quantile_values = [0.0_f64, 0.5, 0.9, 0.99, 1.0] + .iter() + .zip(raw.iter()) + .map(|(&quantile, &value)| summary_data_point::ValueAtQuantile { quantile, value }) + .collect(); + SummaryDataPoint { + attributes: attributes.to_vec(), + start_time_unix_nano: 0, + time_unix_nano: rng.random(), + count, + sum, + quantile_values, + flags: 0, + } } #[derive(Clone, Debug)] @@ -563,7 +789,13 @@ mod test { _ => panic!("invalid aggregation temporality"), } } - _ => panic!("invalid metric data"), + Data::Histogram(_) => assert!(config.metric_weights.histogram_delta >= 1 + || config.metric_weights.histogram_cumulative >= 1), + Data::ExponentialHistogram(_) => { + assert!(config.metric_weights.exp_histogram_delta >= 1 + || config.metric_weights.exp_histogram_cumulative >= 1); + } + Data::Summary(_) => assert!(config.metric_weights.summary >= 1), } } }