feat(aggregate transform): Add set_interval_ms option

changelog.d/15659_aggregate_set_interval_ms.enhancement.md

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

src/transforms/aggregate/config.rs

@@ -33,6 +33,16 @@ 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 not be greater than 4294967295 (about 49.7 days).
+    #[serde(default)]
+    pub set_interval_ms: bool,
 }
 
 /// Settings for event-time aggregation windows.
@@ -150,6 +160,7 @@ impl Default for AggregateConfig {
             interval_ms: default_interval_ms(),
             mode: default_mode(),
             event_time: None,
+            set_interval_ms: false,
         }
     }
 }

src/transforms/aggregate/tests.rs

@@ -191,6 +191,7 @@ fn system_time_config(mode: AggregationMode) -> AggregateConfig {
         interval_ms: 1000,
         mode,
         event_time: None,
+        set_interval_ms: false,
     }
 }
 
@@ -199,6 +200,7 @@ fn event_time_config(interval_ms: u64, mode: AggregationMode) -> AggregateConfig
         interval_ms,
         mode,
         event_time: Some(EventTimeConfig::default()),
+        set_interval_ms: false,
     }
 }
 
@@ -213,6 +215,7 @@ fn event_time_config_with(
         interval_ms,
         mode,
         event_time: Some(event_time),
+        set_interval_ms: false,
     }
 }
 
@@ -1699,3 +1702,200 @@ fn event_time_count_counts_mixed_kinds_in_same_bucket() {
     assert_eq!(out.len(), 1);
     assert_counter(&out[0], 2.0);
 }
+
+fn with_set_interval_ms(config: AggregateConfig) -> AggregateConfig {
+    AggregateConfig {
+        set_interval_ms: true,
+        ..config
+    }
+}
+
+fn interval_ms_of(event: &Event) -> Option<u32> {
+    event.as_metric().interval_ms().map(|i| i.get())
+}
+
+#[test]
+fn set_interval_ms_parses_and_defaults_to_false() {
+    use crate::config::GenerateConfig;
+
+    let generated = AggregateConfig::generate_config();
+    assert_eq!(
+        generated.get("set_interval_ms"),
+        Some(&serde_json::Value::Bool(false))
+    );
+
+    let cfg: AggregateConfig = toml::from_str("interval_ms = 5000").unwrap();
+    assert!(!cfg.set_interval_ms);
+
+    let cfg: AggregateConfig = toml::from_str(
+        r#"
+        interval_ms = 5000
+        set_interval_ms = true
+    "#,
+    )
+    .unwrap();
+    assert!(cfg.set_interval_ms);
+}
+
+#[test]
+fn set_interval_ms_rejects_interval_above_u32() {
+    let too_large = u64::from(u32::MAX) + 1;
+
+    let err = Aggregate::new(&with_set_interval_ms(event_time_config(
+        too_large,
+        AggregationMode::Auto,
+    )))
+    .expect_err("interval_ms above u32::MAX must be rejected with set_interval_ms");
+    assert!(
+        err.to_string().contains("set_interval_ms"),
+        "error should mention set_interval_ms, got: {err}"
+    );
+
+    Aggregate::new(&event_time_config(too_large, AggregationMode::Auto))
+        .expect("the u32 limit only applies when set_interval_ms is enabled");
+    Aggregate::new(&with_set_interval_ms(event_time_config(
+        u64::from(u32::MAX),
+        AggregationMode::Auto,
+    )))
+    .expect("u32::MAX is the largest accepted interval with set_interval_ms");
+}
+
+#[test]
+fn set_interval_ms_disabled_leaves_interval_unset() {
+    let mut agg = Aggregate::new(&system_time_config(AggregationMode::Auto)).unwrap();
+    agg.record(make_metric(
+        "counter_a",
+        MetricKind::Incremental,
+        MetricValue::Counter { value: 1.0 },
+    ));
+
+    let mut out = vec![];
+    agg.flush_into(&mut out);
+    assert_eq!(1, out.len());
+    assert_eq!(interval_ms_of(&out[0]), None);
+}
+
+#[test]
+fn set_interval_ms_sets_interval_on_flushed_incremental_metrics() {
+    let mut agg = Aggregate::new(&with_set_interval_ms(system_time_config(
+        AggregationMode::Auto,
+    )))
+    .unwrap();
+    agg.record(make_metric(
+        "counter_a",
+        MetricKind::Incremental,
+        MetricValue::Counter { value: 42.0 },
+    ));
+    agg.record(make_metric(
+        "counter_a",
+        MetricKind::Incremental,
+        MetricValue::Counter { value: 43.0 },
+    ));
+    agg.record(make_metric(
+        "gauge_a",
+        MetricKind::Absolute,
+        MetricValue::Gauge { value: 7.0 },
+    ));
+
+    let mut out = vec![];
+    agg.flush_into(&mut out);
+    assert_eq!(2, out.len());
+    for event in &out {
+        match event.as_metric().series().name.name.as_str() {
+            "counter_a" => {
+                assert_counter(event, 85.0);
+                assert_eq!(interval_ms_of(event), Some(1000));
+            }
+            "gauge_a" => {
+                assert_gauge(event, 7.0);
+                assert_eq!(
+                    interval_ms_of(event),
+                    None,
+                    "absolute metrics keep no interval"
+                );
+            }
+            other => panic!("Unexpected metric name in aggregate output: {other}"),
+        }
+    }
+
+    // The final flush on shutdown sets the interval too.
+    agg.record(make_metric(
+        "counter_a",
+        MetricKind::Incremental,
+        MetricValue::Counter { value: 1.0 },
+    ));
+    let out = flush_final(&mut agg);
+    assert_eq!(1, out.len());
+    assert_eq!(interval_ms_of(&out[0]), Some(1000));
+}
+
+#[test]
+fn set_interval_ms_overwrites_input_interval() {
+    let mut agg = Aggregate::new(&with_set_interval_ms(system_time_config(
+        AggregationMode::Sum,
+    )))
+    .unwrap();
+    let mut counter = make_metric(
+        "counter_a",
+        MetricKind::Incremental,
+        MetricValue::Counter { value: 1.0 },
+    );
+    counter.as_mut_metric().data_mut().time.interval_ms = std::num::NonZeroU32::new(250);
+    agg.record(counter);
+
+    let mut out = vec![];
+    agg.flush_into(&mut out);
+    assert_eq!(1, out.len());
+    assert_eq!(interval_ms_of(&out[0]), Some(1000));
+}
+
+#[test]
+fn set_interval_ms_does_not_touch_passthrough_or_existing_output() {
+    let mut agg = Aggregate::new(&with_set_interval_ms(system_time_config(
+        AggregationMode::Latest,
+    )))
+    .unwrap();
+    let passthrough = make_metric(
+        "counter_a",
+        MetricKind::Incremental,
+        MetricValue::Counter { value: 1.0 },
+    );
+    let passed = agg
+        .record(passthrough.clone())
+        .expect("Latest passes incremental metrics through");
+    assert_eq!(interval_ms_of(&passed), None);
+
+    // Events already in the output buffer before a flush are left alone.
+    let mut out = vec![passed];
+    agg.flush_into(&mut out);
+    assert_eq!(1, out.len());
+    assert_eq!(out[0], passthrough);
+}
+
+#[test]
+fn set_interval_ms_sets_interval_on_event_time_buckets() {
+    let interval_ms = 10_000;
+    let mut agg = Aggregate::new(&with_set_interval_ms(event_time_config(
+        interval_ms,
+        AggregationMode::Auto,
+    )))
+    .unwrap();
+    let base_time = open_bucket_timestamp(interval_ms);
+    agg.record(make_metric_with_timestamp(
+        "counter_a",
+        MetricKind::Incremental,
+        MetricValue::Counter { value: 2.0 },
+        base_time,
+    ));
+    agg.record(make_metric_with_timestamp(
+        "counter_a",
+        MetricKind::Incremental,
+        MetricValue::Counter { value: 3.0 },
+        base_time + chrono::Duration::milliseconds(100),
+    ));
+
+    let out = flush_final(&mut agg);
+    assert_eq!(1, out.len());
+    assert_counter(&out[0], 5.0);
+    assert_eq!(interval_ms_of(&out[0]), Some(10_000));
+}

src/transforms/aggregate/transform.rs

@@ -2,6 +2,7 @@ use super::{AggregateConfig, AggregationMode};
 
 use std::{
     collections::{BTreeMap, HashMap, hash_map::Entry},
+    num::NonZeroU32,
     pin::Pin,
     time::Duration,
 };
@@ -95,6 +96,8 @@ pub struct Aggregate {
         BTreeMap<BucketKey, HashMap<MetricSeries, Vec<MetricEntry>>>,
     pub(crate) watermark: Option<BucketKey>,
     pub(crate) config: AggregateConfig,
+    /// Interval to set on flushed incremental metrics; `Some` only when `set_interval_ms` is enabled.
+    output_interval_ms: Option<NonZeroU32>,
 }
 
 /// Upper bound for any millisecond-valued duration field that is later cast
@@ -129,6 +132,19 @@ impl Aggregate {
                 .into());
             }
         }
+        let output_interval_ms = if config.set_interval_ms {
+            let interval_ms = u32::try_from(config.interval_ms).map_err(|_| {
+                format!(
+                    "`interval_ms` ({}) exceeds the maximum supported value of {} ms when `set_interval_ms` is enabled",
+                    config.interval_ms,
+                    u32::MAX
+                )
+            })?;
+            // Non-zero was validated above.
+            NonZeroU32::new(interval_ms)
+        } else {
+            None
+        };
 
         Ok(Self {
             interval: Duration::from_millis(config.interval_ms),
@@ -139,6 +155,7 @@ impl Aggregate {
             event_time_multi_buckets: Default::default(),
             watermark: None,
             config: *config,
+            output_interval_ms,
         })
     }
 
@@ -324,11 +341,13 @@ impl Aggregate {
         }
     }
     pub fn flush_into(&mut self, output: &mut Vec<Event>) {
+        let start = output.len();
         if self.config.is_event_time() {
             self.flush_event_time_buckets(output, false);
         } else {
             self.flush_system_time(output);
         }
+        self.set_output_interval(&mut output[start..]);
     }
 
     /// Final flush invoked when the input stream closes. In event-time mode
@@ -337,11 +356,27 @@ impl Aggregate {
     /// shutdown or topology reload, matching system-time semantics where
     /// `flush_system_time` always empties `self.map`.
     pub fn flush_final(&mut self, output: &mut Vec<Event>) {
+        let start = output.len();
         if self.config.is_event_time() {
             self.flush_event_time_buckets(output, true);
         } else {
             self.flush_system_time(output);
         }
+        self.set_output_interval(&mut output[start..]);
+    }
+
+    /// Overwrites `interval_ms` on flushed incremental metrics with the flush interval, so a
+    /// counter aggregated over the window can be read as a rate over that window.
+    fn set_output_interval(&self, flushed: &mut [Event]) {
+        let Some(interval_ms) = self.output_interval_ms else {
+            return;
+        };
+        for event in flushed {
+            let metric = event.as_mut_metric();
+            if metric.kind() == MetricKind::Incremental {
+                metric.data_mut().time.interval_ms = Some(interval_ms);
+            }
+        }
     }
 
     fn flush_system_time(&mut self, output: &mut Vec<Event>) {

website/cue/reference/components/transforms/generated/aggregate.cue

@@ -97,4 +97,17 @@ generated: components: transforms: aggregate: configuration: {
 			}
 		}
 	}
+	set_interval_ms: {
+		description: """
+			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 not be greater than 4294967295 (about 49.7 days).
+			"""
+		required: false
+		type: bool: default: false
+	}
 }