feat(aggregate transform): Add set_interval_ms option

changelog.d/15659_aggregate_set_interval_ms.enhancement.md

@@ -0,0 +1,6 @@
+The `aggregate` transform now has a `set_interval_ms` option. When enabled, each flushed
+incremental metric gets `interval_ms` set to the configured flush interval, so sinks such as
+`datadog_metrics` send aggregated counters as rates instead of counts. With `event_time`, these
+metrics also get the start of their bucket as their timestamp.
+
+authors: gremlinops

changelog.d/26597_datadog_metrics_fractional_interval.fix.md

@@ -0,0 +1,8 @@
+The `datadog_metrics` sink now converts a counter's `interval_ms` to seconds without integer
+division. Counters with an interval under one second were sent as rates with infinite or `NaN`
+values, and intervals that are not whole seconds overstated the rate. The rate value now uses
+the exact interval, and the interval field is rounded to the nearest whole second (at least 1).
+When the sink combines counters with an interval that share a series and second, it now also
+combines their intervals, so the rate is not overstated.
+
+authors: gremlinops

src/sinks/datadog/metrics/encoder.rs

@@ -558,6 +558,26 @@ fn sketch_to_proto_message(
     })
 }
 
+/// A metric's interval, in the two forms the series wire formats need.
+#[derive(Clone, Copy, Debug, PartialEq)]
+pub(super) struct SeriesInterval {
+    /// Whole seconds for the wire `interval` field, rounded to the nearest second and at least 1.
+    pub(super) whole_secs: u32,
+    /// Exact seconds, used to scale a rate's value to per-second.
+    pub(super) exact_secs: f64,
+}
+
+/// Converts a metric's millisecond `interval_ms`, if any, into a [`SeriesInterval`].
+pub(super) fn series_interval(metric: &Metric) -> Option<SeriesInterval> {
+    metric.interval_ms().map(|interval_ms| {
+        let exact_secs = f64::from(interval_ms.get()) / 1000.0;
+        SeriesInterval {
+            whole_secs: (exact_secs.round() as u32).max(1),
+            exact_secs,
+        }
+    })
+}
+
 /// A metric's tags, split into the three pieces the series wire formats send separately.
 pub(super) struct SeriesTags {
     /// Remaining tags, encoded as sorted `key:value` (or bare `key`) strings.
@@ -646,16 +666,15 @@ fn series_to_proto_message(
 
     let timestamp = encode_timestamp(metric.timestamp());
 
-    // our internal representation is in milliseconds but the expected output is in seconds
-    let maybe_interval = metric.interval_ms().map(|i| i.get() / 1000);
+    let maybe_interval = series_interval(metric);
 
     let (points, metric_type) = match metric.value() {
         MetricValue::Counter { value } => {
             if let Some(interval) = maybe_interval {
                 // When an interval is defined, it implies the value should be in a per-second form,
                 // so we need to get back to seconds from our milliseconds-based interval, and then
                 // divide our value by that amount as well.
-                let value = *value / (interval as f64);
+                let value = *value / interval.exact_secs;
                 (
                     vec![ddmetric_proto::metric_payload::MetricPoint { value, timestamp }],
                     ddmetric_proto::metric_payload::MetricType::Rate,
@@ -703,7 +722,7 @@ fn series_to_proto_message(
         // unit is omitted
         unit: "".to_string(),
         source_type_name,
-        interval: maybe_interval.unwrap_or(0) as i64,
+        interval: maybe_interval.map_or(0, |interval| i64::from(interval.whole_secs)),
         metadata,
     })
 }
@@ -873,9 +892,10 @@ mod tests {
     };
 
     use super::{
-        DatadogMetricsEncoder, EncoderError, ddmetric_proto, encode_proto_key_and_message,
-        encode_tags, encode_timestamp, get_sketch_payload_sketches_field_number,
-        max_compressed_size, series_to_proto_message, sketch_to_proto_message,
+        DatadogMetricsEncoder, EncoderError, SeriesInterval, ddmetric_proto,
+        encode_proto_key_and_message, encode_tags, encode_timestamp,
+        get_sketch_payload_sketches_field_number, max_compressed_size, series_interval,
+        series_to_proto_message, sketch_to_proto_message,
     };
     use crate::sinks::{
         datadog::metrics::{
@@ -1207,6 +1227,56 @@ mod tests {
         }
     }
 
+    #[test]
+    fn series_interval_rounds_wire_field_and_keeps_exact_seconds() {
+        let cases = [
+            (1, 1, 0.001),
+            (500, 1, 0.5),
+            (1_000, 1, 1.0),
+            (1_499, 1, 1.499),
+            (1_500, 2, 1.5),
+            (10_000, 10, 10.0),
+            (u32::MAX, 4_294_967, 4_294_967.295),
+        ];
+        for (interval_ms, whole_secs, exact_secs) in cases {
+            let metric = get_simple_rate_counter(1.0, interval_ms);
+            assert_eq!(
+                series_interval(&metric),
+                Some(SeriesInterval {
+                    whole_secs,
+                    exact_secs
+                }),
+                "interval_ms = {interval_ms}"
+            );
+        }
+        assert_eq!(series_interval(&get_simple_counter()), None);
+    }
+
+    #[test]
+    fn encode_counter_with_fractional_second_interval_as_rate() {
+        for (value, interval_ms, expected_value, expected_interval) in
+            [(10.0, 500, 20.0, 1), (3.0, 1_500, 2.0, 2)]
+        {
+            let series_proto = series_to_proto_message(
+                &get_simple_rate_counter(value, interval_ms),
+                &None,
+                log_schema(),
+                DEFAULT_DD_ORIGIN_PRODUCT_VALUE,
+            )
+            .unwrap();
+            assert_eq!(series_proto.r#type, 2, "interval_ms = {interval_ms}");
+            assert_eq!(
+                series_proto.interval, expected_interval,
+                "interval_ms = {interval_ms}"
+            );
+            assert_eq!(series_proto.points.len(), 1);
+            assert_eq!(
+                series_proto.points[0].value, expected_value,
+                "interval_ms = {interval_ms}"
+            );
+        }
+    }
+
     #[test]
     fn encode_non_rate_metric_with_interval() {
         // It is possible that the Agent sends Gauges with an interval set. This

src/sinks/datadog/metrics/encoder_v3.rs

@@ -24,7 +24,7 @@ use super::{
     config::DatadogMetricsEndpoint,
     encoder::{
         EncoderError, FinishError, ORIGIN_CATEGORY_VALUE, ORIGIN_PRODUCT_VALUE, SeriesTags,
-        generate_origin_metadata, split_series_tags,
+        generate_origin_metadata, series_interval, split_series_tags,
     },
 };
 use crate::sinks::util::{
@@ -180,7 +180,7 @@ fn encode_metric_to_v3(
 ) -> Result<(), EncoderError> {
     // Mirrors V2's `series_to_proto_message`: a Counter with an interval is sent as a
     // per-second-scaled Rate, not a raw Count.
-    let maybe_interval = metric.interval_ms().map(|i| i.get() / 1000);
+    let maybe_interval = series_interval(metric);
 
     let metric_type = match metric.value() {
         MetricValue::Counter { .. } if maybe_interval.is_some() => V3MetricType::Rate,
@@ -261,9 +261,9 @@ fn encode_metric_to_v3(
     }
 
     // Interval — matches V2, which always stamps the interval field on the message
-    // (`interval: maybe_interval.unwrap_or(0)`), even though only Rate uses it to scale the value.
+    // when an interval is present, even though only Rate uses it to scale the value.
     if let Some(interval) = maybe_interval {
-        builder.set_interval(interval.into());
+        builder.set_interval(interval.whole_secs.into());
     }
 
     // Note: `unit` is intentionally never set — V2 always sends it empty (see
@@ -275,7 +275,7 @@ fn encode_metric_to_v3(
     match metric.value() {
         MetricValue::Counter { value } => {
             let value = match maybe_interval {
-                Some(interval) => *value / (interval as f64),
+                Some(interval) => *value / interval.exact_secs,
                 None => *value,
             };
             builder.add_point(timestamp, value);
@@ -520,6 +520,25 @@ mod tests {
         );
     }
 
+    #[test]
+    fn v3_counter_with_sub_second_interval_scales_by_exact_seconds() {
+        let encode = |value: f64, interval_ms: u32| {
+            let mut enc = DatadogMetricsV3Encoder::new(
+                DatadogMetricsEndpoint::Series(SeriesApiVersion::V3),
+                None,
+            );
+            let metric = counter("rate.counter", value)
+                .with_timestamp(DateTime::from_timestamp(1_700_000_000, 0))
+                .with_interval_ms(NonZeroU32::new(interval_ms));
+            enc.try_encode(metric).unwrap();
+            enc.finish().unwrap().0.into_payload()
+        };
+
+        // Both are a rate of 1/s with a wire interval of 1 second.
+        assert_eq!(encode(0.5, 500), encode(1.0, 1_000));
+        assert_ne!(encode(1.0, 500), encode(1.0, 1_000));
+    }
+
     #[test]
     fn v3_non_series_metric_values_are_rejected() {
         // `AggregatedSummary` is split into counters/gauges, and `Distribution`/

src/sinks/datadog/metrics/sink.rs

@@ -239,30 +239,47 @@ fn sort_and_collapse_counters_by_series_and_timestamp(mut metrics: Vec<Metric>)
         }
 
         // Only aggregate counters. All other types can be skipped.
-        if let (
-            MetricValue::Counter { value: left_value },
-            MetricValue::Counter { value: right_value },
-        ) = (left.value(), right.value_mut())
-        {
-            // NOTE: The docs for `dedup_by` specify that if `left`/`right` are equal, then
-            // `left` is the element that gets removed.
-            *right_value += left_value;
-            right
-                .metadata_mut()
-                .merge_finalizers(left.metadata_mut().take_finalizers());
-
-            true
-        } else {
-            false
+        let (MetricValue::Counter { value: left_value }, MetricValue::Counter { .. }) =
+            (left.value(), right.value())
+        else {
+            return false;
+        };
+        let left_value = *left_value;
+
+        // NOTE: The docs for `dedup_by` specify that if `left`/`right` are equal, then
+        // `left` is the element that gets removed.
+        match (left.interval_ms(), right.interval_ms()) {
+            // Rates are scaled by their interval, so the collapsed counter must also cover both
+            // windows. `update` merges the windows from the timestamps and sums the values.
+            (Some(_), Some(_)) if left.timestamp().is_some() && right.timestamp().is_some() => {
+                if !right.data_mut().update(left.data()) {
+                    return false;
+                }
+            }
+            (left_interval, right_interval) => {
+                if let MetricValue::Counter { value } = right.value_mut() {
+                    *value += left_value;
+                }
+                if let (Some(left_interval), Some(right_interval)) = (left_interval, right_interval)
+                {
+                    right.data_mut().time.interval_ms =
+                        Some(right_interval.saturating_add(left_interval.get()));
+                }
+            }
         }
+        right
+            .metadata_mut()
+            .merge_finalizers(left.metadata_mut().take_finalizers());
+
+        true
     });
 
     metrics
 }
 
 #[cfg(test)]
 mod tests {
-    use std::{collections::HashSet, time::Duration};
+    use std::{collections::HashSet, num::NonZeroU32, time::Duration};
 
     use chrono::{DateTime, Utc};
     use proptest::prelude::*;
@@ -366,6 +383,45 @@ mod tests {
         assert_eq!(expected, actual);
     }
 
+    fn create_rate_counter(
+        value: f64,
+        interval_ms: u32,
+        timestamp: Option<DateTime<Utc>>,
+    ) -> Metric {
+        create_counter("rate", value)
+            .with_interval_ms(NonZeroU32::new(interval_ms))
+            .with_timestamp(timestamp)
+    }
+
+    #[test]
+    fn collapse_rate_counters_merges_their_windows() {
+        let start = DateTime::from_timestamp_millis(1_700_000_000_000);
+        let half_second_later = DateTime::from_timestamp_millis(1_700_000_000_500);
+        for input in [
+            vec![
+                create_rate_counter(1.0, 500, start),
+                create_rate_counter(3.0, 500, half_second_later),
+            ],
+            vec![
+                create_rate_counter(3.0, 500, half_second_later),
+                create_rate_counter(1.0, 500, start),
+            ],
+        ] {
+            let actual = sort_and_collapse_counters_by_series_and_timestamp(input);
+            assert_eq!(vec![create_rate_counter(4.0, 1_000, start)], actual);
+        }
+    }
+
+    #[test]
+    fn collapse_rate_counters_without_timestamps_sums_their_intervals() {
+        let input = vec![
+            create_rate_counter(1.0, 500, None),
+            create_rate_counter(3.0, 500, None),
+        ];
+        let actual = sort_and_collapse_counters_by_series_and_timestamp(input);
+        assert_eq!(vec![create_rate_counter(4.0, 1_000, None)], actual);
+    }
+
     #[test]
     fn collapse_identical_metrics_counter_unsorted() {
         let gauge_value = 1.0;

src/transforms/aggregate/config.rs

@@ -33,6 +33,19 @@ pub struct AggregateConfig {
     /// they are processed. Omit this block to keep the default system-time behavior.
     #[serde(default, skip_serializing_if = "Option::is_none")]
     pub event_time: Option<EventTimeConfig>,
+
+    /// Set `interval_ms` on each aggregated incremental metric to the flush interval.
+    ///
+    /// Downstream components can then treat an aggregated counter as a rate over the flush
+    /// interval. For example, the `datadog_metrics` sink sends counters that have an interval
+    /// as rates. Absolute metrics and metrics that pass through unchanged are not modified.
+    ///
+    /// With `event_time`, the timestamp of these metrics is also set to the start of their
+    /// bucket, so the timestamp and the interval describe the same window.
+    ///
+    /// When enabled, `interval_ms` must not be greater than 4294967295 (about 49.7 days).
+    #[serde(default)]
+    pub set_interval_ms: bool,
 }
 
 /// Settings for event-time aggregation windows.
@@ -150,6 +163,7 @@ impl Default for AggregateConfig {
             interval_ms: default_interval_ms(),
             mode: default_mode(),
             event_time: None,
+            set_interval_ms: false,
         }
     }
 }

src/transforms/aggregate/event_time.rs

@@ -412,6 +412,7 @@ impl Aggregate {
             .collect();
 
         for bucket_key in buckets_to_flush {
+            let bucket_output_start = output.len();
             if let Some(bucket_map) = self.event_time_buckets.remove(&bucket_key) {
                 // Diff mode must retain `bucket_map` to subtract against the
                 // next flush, so it iterates by reference and per-entry clones
@@ -507,6 +508,7 @@ impl Aggregate {
                     }
                 }
             }
+            self.set_bucket_start_timestamps(&mut output[bucket_output_start..], bucket_key);
 
             // Advance the watermark to the *exclusive end* of the highest
             // flushed bucket so subsequent events for that window (or any
@@ -521,4 +523,21 @@ impl Aggregate {
             emit!(AggregateFlushed);
         }
     }
+
+    /// With `set_interval_ms`, an incremental bucket metric describes the window
+    /// `[bucket_key, bucket_key + interval_ms)`, so its timestamp must be the window start.
+    fn set_bucket_start_timestamps(&self, flushed: &mut [Event], bucket_key: BucketKey) {
+        if self.output_interval_ms.is_none() {
+            return;
+        }
+        let Some(bucket_start) = DateTime::<Utc>::from_timestamp_millis(bucket_key) else {
+            return;
+        };
+        for event in flushed {
+            let metric = event.as_mut_metric();
+            if metric.kind() == MetricKind::Incremental {
+                metric.data_mut().time.timestamp = Some(bucket_start);
+            }
+        }
+    }
 }