enhancement(aggregate transform): Add Distribution as an aggregation mode

changelog.d/23994_aggregate_transform.enhancement.md

@@ -0,0 +1,3 @@
+The aggregate transform now supports aggregating metrics into Distribution metric types, enabling statistical analysis of metric values over time windows.
+
+authors: jlambatl

src/transforms/aggregate/config.rs

@@ -138,6 +138,14 @@ pub enum AggregationMode {
 
     /// Stdev value of absolute metric; incremental metrics pass through unchanged.
     Stdev,
+
+    /// Collects the values of absolute metrics into a single sorted distribution; incremental
+    /// metrics pass through unchanged.
+    ///
+    /// Gauges and sets (numeric members only) contribute one sample per value, distributions
+    /// contribute their existing samples, and aggregated histograms contribute one sample at each
+    /// bucket's upper limit per count in that bucket. Other absolute metric types are ignored.
+    Distribution,
 }
 
 const fn default_mode() -> AggregationMode {

src/transforms/aggregate/event_time.rs

@@ -217,8 +217,9 @@ impl Aggregate {
                 let bucket = self.event_time_buckets.entry(bucket_key).or_default();
                 Self::record_comparison_in_map(bucket, series, data, metadata, mode);
             }
-            AggregationMode::Mean | AggregationMode::Stdev => {
-                // `will_be_stored` has already guaranteed Absolute + Gauge.
+            AggregationMode::Mean | AggregationMode::Stdev | AggregationMode::Distribution => {
+                // `will_be_stored` has already guaranteed Absolute (+ Gauge for Mean/Stdev, or a
+                // convertible value for Distribution).
                 //
                 // Mean/Stdev write samples into `event_time_multi_buckets`;
                 // an empty entry is also placed in `event_time_buckets` so
@@ -466,6 +467,17 @@ impl Aggregate {
                             continue;
                         }
 
+                        if matches!(self.config.mode, AggregationMode::Distribution) {
+                            if let Some((data, metadata)) =
+                                Self::distribution_from_entries(&entries)
+                            {
+                                output.push(Event::Metric(Metric::from_parts(
+                                    series, data, metadata,
+                                )));
+                            }
+                            continue;
+                        }
+
                         let (mut final_sum, mut final_metadata) = entries.first().unwrap().clone();
                         for (data, metadata) in entries.iter().skip(1) {
                             if !final_sum.update(data) {