fix(metrics): calculate ddsketch averages from histogram sum & count fields

.github/actions/spelling/expect.txt

@@ -522,6 +522,7 @@ SIGHUPs
 sinkbuilder
 sinknetworkbytessent
 slideover
+SLOs
 smartphone
 Smol
 smtracking

changelog.d/dd_sketch_averages_from_histograms.fix.md

@@ -0,0 +1,5 @@
+Modified behavior of histograms (ddsketch) submitted by the Datadog metrics sink.  Previously the ddsketch sum, count, and average fields were interpolated from bucket boundaries, which could cause slight inaccuracies.  Now those fields will be calculated directly from the corresponding fields on the incoming histogram.  Min and max fields remain interpolated.
+
+Users with alerts, dashboards, or SLOs based on histogram averages should review and adjust thresholds after upgrading, as values will settle to their correct levels.
+
+authors: tony-resendiz

lib/vector-core/src/metrics/ddsketch.rs

@@ -7,6 +7,7 @@ use ordered_float::OrderedFloat;
 use serde::{Deserialize, Deserializer, Serialize, Serializer, de::Error};
 use serde_with::{DeserializeAs, SerializeAs, serde_as};
 use snafu::Snafu;
+use tracing::warn;
 use vector_common::byte_size_of::ByteSizeOf;
 use vector_config::configurable_component;
 
@@ -788,10 +789,55 @@ impl AgentDDSketch {
                 }
                 Some(sketch)
             }
-            MetricValue::AggregatedHistogram { buckets, .. } => {
+            MetricValue::AggregatedHistogram {
+                buckets,
+                sum,
+                count,
+            } => {
                 let delta_buckets = mem::take(buckets);
                 let mut sketch = AgentDDSketch::with_agent_defaults();
                 sketch.insert_interpolate_buckets(delta_buckets)?;
+
+                // Prefer exact avg from sum/count when available.
+                //
+                // NOTE:
+                // AggregatedHistogram -> DDSketch conversion historically relies entirely
+                // on bucket interpolation, which is approximate.
+                //
+                // Some upstream sources (e.g. OTLP) encode missing sums as 0.0, so we
+                // cannot distinguish between "unknown sum" and "actual zero sum" here.
+                // As a result, we only override when sum != 0.0 and fall back to
+                // interpolation otherwise.
+                //
+                // This preserves existing behavior for ambiguous cases while improving
+                // accuracy when a non-zero sum is available.
+                //
+                // Do not override sketch.count here.
+                // The interpolated bins define the sketch's sample mass, and count must
+                // remain consistent with those bins to preserve DDSketch invariants.
+                //
+                // We do override sum/avg to improve accuracy when upstream provides them.
+                // Note that this can result in avg != sum / sketch.count if upstream
+                // count differs from bucket totals (e.g. Prometheus +Inf bucket drop).
+                // This tradeoff favors more accurate summary stats while preserving
+                // correct quantile behavior.
+                match u32::try_from(*count) {
+                    Ok(c) => {
+                        let sc = sketch.count();
+
+                        if c > 0 && *sum != 0.0 && sc > 0 {
+                            sketch.sum = *sum;
+                            sketch.avg = *sum / sc as f64;
+                        }
+                    }
+                    Err(_) => {
+                        warn!(
+                            message = "Aggregated histogram count too large to convert to u32; \
+                            falling back to interpolated average"
+                        );
+                    }
+                }
+
                 Some(sketch)
             }
             // We can't convert from any other metric value.
@@ -1097,7 +1143,8 @@ fn round_to_even(v: f64) -> f64 {
 #[cfg(test)]
 mod tests {
     use super::{AGENT_DEFAULT_EPS, AgentDDSketch, Config, MAX_KEY, round_to_even};
-    use crate::event::metric::Bucket;
+    use crate::event::metric::{Bucket, MetricSketch};
+    use crate::event::{Metric, MetricKind, MetricValue};
 
     const FLOATING_POINT_ACCEPTABLE_ERROR: f64 = 1.0e-10;
 
@@ -1206,6 +1253,40 @@ mod tests {
         assert!((end - max).abs() < FLOATING_POINT_ACCEPTABLE_ERROR);
     }
 
+    #[test]
+    fn test_transform_to_sketch_uses_histogram_sum_count() {
+        let buckets = vec![
+            Bucket {
+                upper_limit: 1.0,
+                count: 1,
+            },
+            Bucket {
+                upper_limit: 2.0,
+                count: 1,
+            },
+        ];
+        let histogram = Metric::new(
+            "histogram",
+            MetricKind::Absolute,
+            MetricValue::AggregatedHistogram {
+                count: 2,
+                sum: 3.0,
+                buckets,
+            },
+        );
+        let metric = AgentDDSketch::transform_to_sketch(histogram).unwrap();
+        match metric.data().value() {
+            MetricValue::Sketch {
+                sketch: MetricSketch::AgentDDSketch(dd),
+            } => {
+                assert_eq!(dd.count, 2);
+                assert_eq!(dd.sum, 3.0);
+                assert_eq!(dd.avg, 1.5);
+            }
+            other => panic!("expected AgentDDSketch, got: {other:?}"),
+        }
+    }
+
     #[test]
     fn test_out_of_range_buckets_error() {
         let mut sketch = AgentDDSketch::with_agent_defaults();