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 this option,
+`interval_ms` must be a whole number of seconds.
+
+authors: gremlinops

changelog.d/26597_datadog_metrics_fractional_interval.fix.md

@@ -0,0 +1,6 @@
+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).
+
+authors: gremlinops

src/sinks/datadog/metrics/encoder.rs

@@ -558,6 +558,27 @@ 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 ms = interval_ms.get();
+        let rounded_secs = ms / 1000 + u32::from(ms % 1000 >= 500);
+        SeriesInterval {
+            whole_secs: rounded_secs.max(1),
+            exact_secs: f64::from(ms) / 1000.0,
+        }
+    })
+}
+
 /// 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 +667,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 +723,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 +893,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 +1228,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/transforms/aggregate/config.rs

@@ -33,6 +33,17 @@ 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.
+    ///
+    /// When enabled, `interval_ms` must be a whole number of seconds (a multiple of 1000) and
+    /// 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 +161,7 @@ impl Default for AggregateConfig {
             interval_ms: default_interval_ms(),
             mode: default_mode(),
             event_time: None,
+            set_interval_ms: false,
         }
     }
 }