Reject metric interval merges that overflow

MetricData::update added the timestamp delta and an interval with
unchecked u32 arithmetic. Intervals near u32::MAX panicked in debug
builds and wrapped in release builds. Return false instead, as the
function already does when the timestamp delta does not fit.

Round the Datadog whole-second interval with integer arithmetic so it
needs no float cast, and fix two pedantic lints in the new tests.

Author: gremlinops

changelog.d/26597_datadog_metrics_fractional_interval.fix.md

@@ -3,6 +3,7 @@ division. Counters with an interval under one second were sent as rates with inf
 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.
+combines their intervals, so the rate is not overstated. Merging two metric intervals whose
+combined window does not fit in `interval_ms` no longer panics or wraps.
 
 authors: gremlinops

lib/vector-core/src/event/metric/data.rs

@@ -108,16 +108,19 @@ impl MetricData {
                     return false;
                 };
 
-                if t1 > t2 {
+                let (start, end_interval) = if t1 > t2 {
                     // The interval window starts from the beginning of `other` (aka `t2`)
                     // and goes to the end of `self` (which is `t1 + i1`).
-                    (Some(t2), NonZeroU32::new(delta_t + i1.get()))
+                    (t2, i1)
                 } else {
                     // The interval window starts from the beginning of `self` (aka `t1`)
                     // and goes to the end of `other` (which is `t2 + i2`).
-
-                    (Some(t1), NonZeroU32::new(delta_t + i2.get()))
-                }
+                    (t1, i2)
+                };
+                let Some(interval) = delta_t.checked_add(end_interval.get()) else {
+                    return false;
+                };
+                (Some(start), NonZeroU32::new(interval))
             }
             (Some(t), _, None, _) | (None, _, Some(t), _) => (Some(t), None),
             (Some(t1), _, Some(t2), _) => (Some(t1.max(t2)), None),

lib/vector-core/src/event/metric/mod.rs

@@ -1203,4 +1203,23 @@ mod test {
         assert!(gauge.data.add(&delta.data));
         assert_eq!(gauge, expected);
     }
+
+    #[test]
+    fn merge_interval_overflow_is_rejected() {
+        let mut counter = Metric::new(
+            "counter",
+            MetricKind::Incremental,
+            MetricValue::Counter { value: 1.0 },
+        )
+        .with_timestamp(Some(ts()))
+        .with_interval_ms(std::num::NonZeroU32::new(u32::MAX));
+
+        let delta = counter
+            .clone()
+            .with_timestamp(Some(ts() + chrono::Duration::milliseconds(1)));
+
+        let before = counter.clone();
+        assert!(!counter.data.add(&delta.data));
+        assert_eq!(counter, before);
+    }
 }

src/sinks/datadog/metrics/encoder.rs

@@ -570,10 +570,11 @@ pub(super) struct SeriesInterval {
 /// 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;
+        let ms = interval_ms.get();
+        let rounded_secs = ms / 1000 + u32::from(ms % 1000 >= 500);
         SeriesInterval {
-            whole_secs: (exact_secs.round() as u32).max(1),
-            exact_secs,
+            whole_secs: rounded_secs.max(1),
+            exact_secs: f64::from(ms) / 1000.0,
         }
     })
 }

src/transforms/aggregate/tests.rs

@@ -1711,7 +1711,10 @@ fn with_set_interval_ms(config: AggregateConfig) -> AggregateConfig {
 }
 
 fn interval_ms_of(event: &Event) -> Option<u32> {
-    event.as_metric().interval_ms().map(|i| i.get())
+    event
+        .as_metric()
+        .interval_ms()
+        .map(std::num::NonZeroU32::get)
 }
 
 #[test]
@@ -1728,10 +1731,10 @@ fn set_interval_ms_parses_and_defaults_to_false() {
     assert!(!cfg.set_interval_ms);
 
     let cfg: AggregateConfig = toml::from_str(
-        r#"
+        r"
         interval_ms = 5000
         set_interval_ms = true
-    "#,
+    ",
     )
     .unwrap();
     assert!(cfg.set_interval_ms);