fix(otlp): preserve cumulative explicit histogram state across payload generations (#1919)

* feat(otlp): enforce count strictly across amongst other invariants

* feat(otlp): add changelog and replace falsely removed comments

Author: lucastemb

CHANGELOG.md

@@ -12,7 +12,8 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
   metric kind to simplify intake debugging.
 - OpenTelemetry cumulative metric generation now updates cumulative sums using
   aggregation temporality, preserves cumulative histogram min/max bounds, and
-  supports configurable histogram count limits.
+  supports configurable histogram count limits. Cumulative explicit histograms
+  now retain their state across payload generations.
 - HTTP blackhole now supports an `openmetrics` body variant for generated
   Prometheus/OpenMetrics scrape responses.
 - Updated to rand 0.10.x

ci/fingerprints/otel_metrics/fingerprint.txt

@@ -1 +1 @@
-otel_metrics_grpc: dfd6a48683a2731748fcb9c40a636ac93b7fffc679f73aa6e660f11b3060d4aa entropy=6.7678
+otel_metrics_grpc: 3cd72bf5b323d0482a620db2383d4f81e257a91eeb58090888bc5b2a27502b57 entropy=6.8202

lading_payload/src/opentelemetry/common/templates.rs

@@ -58,38 +58,56 @@ where
         G: crate::SizedGenerator<'a, Output = T>,
         G::Error: 'a,
     {
-        // If we are at context cap, search by_size for templates <= budget and
-        // return a random choice. If we are not at context cap, call
-        // generator with the budget and then store the result
-        // for future use in `by_size`.
+        self.fetch_mut(rng, budget).map(|template| &*template)
+    }
+
+    /// Return a mutable reference to an item from the pool.
+    ///
+    /// Callers that model cumulative protocol state update this stored template
+    /// so later selections of the same template retain their prior values.
+    pub(crate) fn fetch_mut<'a, R>(
+        &'a mut self,
+        rng: &mut R,
+        budget: &mut usize,
+    ) -> Result<&'a mut T, PoolError<G::Error>>
+    where
+        R: Rng + ?Sized,
+        G: crate::SizedGenerator<'a, Output = T>,
+        G::Error: 'a,
+    {
+        // If we are at context cap, search by_size for templates within the
+        // budget. Otherwise, generate and store one additional template before
+        // selecting a random eligible template.
         //
         // Size search is in the interval (0, budget].
-
         let upper = *budget;
 
         // Generate new instances until either context_cap is hit or the
-        // remaining space drops below our lookup interval.
+        // remaining storage budget is exhausted.
         if self.len < self.context_cap && self.consumed_bytes < self.max_available_bytes {
             let mut limit = *budget;
             if let Ok(item) = self.generator.generate(rng, &mut limit) {
-                let sz = item.encoded_len();
-                self.by_size.entry(sz).or_default().push(item);
+                let size = item.encoded_len();
+                self.by_size.entry(size).or_default().push(item);
                 self.len += 1;
-                self.consumed_bytes = self.consumed_bytes.saturating_add(sz);
-            } else {
-                // Generation failed. It's possible there's an existing
-                // template that fits the budget.
+                self.consumed_bytes = self.consumed_bytes.saturating_add(size);
             }
         }
+        // A generation failure does not prevent selecting an existing template
+        // that fits the requested budget.
 
-        let (choice_sz, choices) = self
+        let choice_size = *self
             .by_size
             .range(..=upper)
+            .map(|(size, _)| size)
             .choose(rng)
             .ok_or(PoolError::EmptyChoice)?;
-
-        let choice = choices.choose(rng).ok_or(PoolError::EmptyChoice)?;
-        *budget = budget.saturating_sub(*choice_sz);
+        let choices = self
+            .by_size
+            .get_mut(&choice_size)
+            .ok_or(PoolError::EmptyChoice)?;
+        let choice = choices.choose_mut(rng).ok_or(PoolError::EmptyChoice)?;
+        *budget = budget.saturating_sub(choice_size);
 
         Ok(choice)
     }

lading_payload/src/opentelemetry/metric.rs

@@ -618,8 +618,8 @@ impl<'a> SizedGenerator<'a> for OpentelemetryMetrics {
         self.incr_f += rng.random_range(1.0..=100.0);
         self.incr_i += rng.random_range(1_i64..=100_i64);
 
-        let mut tpl: ResourceMetrics = match self.pool.fetch(rng, budget) {
-            Ok(t) => t.to_owned(),
+        let tpl: &mut ResourceMetrics = match self.pool.fetch_mut(rng, budget) {
+            Ok(template) => template,
             Err(PoolError::EmptyChoice) => {
                 debug!("Pool was unable to satify request for {budget} size");
                 Err(PoolError::EmptyChoice)?
@@ -882,7 +882,7 @@ impl<'a> SizedGenerator<'a> for OpentelemetryMetrics {
         *budget = original_budget - required_bytes;
         self.data_points_per_resource = data_points_count;
 
-        Ok(tpl)
+        Ok(tpl.clone())
     }
 }
 
@@ -1814,7 +1814,10 @@ mod test {
                 let previous = first_histogram_point(&first);
                 let current = first_histogram_point(&second);
                 assert_non_decreasing_buckets(&previous.bucket_counts, &current.bucket_counts);
+                assert_eq!(current.count, current.bucket_counts.iter().sum::<u64>());
                 assert!(current.count >= previous.count);
+                assert!(current.time_unix_nano > previous.time_unix_nano);
+                assert_eq!(current.start_time_unix_nano, previous.start_time_unix_nano);
                 assert!(current.sum >= previous.sum);
                 assert!(current.min <= previous.min);
                 assert!(current.max >= previous.max);

lading_payload/src/opentelemetry/metric/templates.rs

@@ -194,8 +194,15 @@ impl<'a> crate::SizedGenerator<'a> for MetricTemplateGenerator {
         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);
+        // Cumulative histograms keep one state per metric identity. Other
+        // metric kinds use a weighted distribution, heavily favoring one or
+        // two points while allowing up to 60 for payload variety.
+        let total_data_points = match kind {
+            Kind::Histogram {
+                aggregation_temporality: 2,
+            } => 1,
+            _ => exponential_weighted_range(rng, 1, 60),
+        };
         let data = match kind {
             Kind::Gauge => {
                 let data_points = (0..total_data_points)