fix(metrics): prevent negative utilization on late messages (#24073)

* fix(metrics): prevent negative utilization on late messages

This prevents late messages from breaking expected utilization bounds, by clamping all late messages
to the start of reporting period. Messages arrive in order, so that should not break the expected
state (waiting or not waiting). Utilization metric can still be slightly inaccurate on late
messages, but it was never meant to be completely accurate.

Fixes: #24060

* Add changelog entry

* Add more details to changelog

* Update src/utilization.rs

Co-authored-by: Pavlos Rontidis <[email protected]>

* Use `assert_approx_eq` in utilization tests

---------

Co-authored-by: Pavlos Rontidis <[email protected]>

Author: esensar

changelog.d/24073_utilization_bounds.fix.md

@@ -0,0 +1,3 @@
+Fixed an issue where utilization could report negative values. This could happen if messages from components were processed too late and were accounted for wrong utilization measurement period. These messages are now moved to the current utilization period, meaning there might be some inaccuracy in the resulting utilization metric, but it was never meant to be precise.
+
+authors: esensar Quad9DNS

src/utilization.rs

@@ -110,15 +110,17 @@ impl Timer {
     /// Begin a new span representing time spent waiting
     pub(crate) fn start_wait(&mut self, at: Instant) {
         if !self.waiting {
-            self.end_span(at);
+            // Clamp start time in case of a late message
+            self.end_span(at.max(self.overall_start));
             self.waiting = true;
         }
     }
 
     /// Complete the current waiting span and begin a non-waiting span
     pub(crate) fn stop_wait(&mut self, at: Instant) {
         if self.waiting {
-            self.end_span(at);
+            // Clamp stop time in case of a late message
+            self.end_span(at.max(self.overall_start));
             self.waiting = false;
         }
     }
@@ -143,7 +145,7 @@ impl Timer {
         self.gauge.set(avg_rounded);
 
         // Reset overall statistics for the next reporting period.
-        self.overall_start = self.span_start;
+        self.overall_start = now;
         self.total_wait = Duration::new(0, 0);
 
         #[cfg(debug_assertions)]
@@ -162,7 +164,9 @@ impl Timer {
 
     fn end_span(&mut self, at: Instant) {
         if self.waiting {
-            self.total_wait += at - self.span_start;
+            // `at` can be before span start here, the result will be clamped to 0
+            // because `duration_since` returns zero if `at` is before span start
+            self.total_wait += at.duration_since(self.span_start);
         }
         self.span_start = at;
     }
@@ -322,6 +326,49 @@ mod tests {
         );
     }
 
+    #[test]
+    #[serial]
+    fn test_utilization_in_bounds_on_late_start() {
+        let mut timer = setup_timer();
+
+        MockClock::advance(Duration::from_secs(5));
+
+        timer.update_utilization();
+
+        let avg = timer.ewma.average().unwrap();
+        assert_approx_eq(avg, 1.0, "near 1.0 (never waiting)");
+
+        // Late message for start wait
+        timer.start_wait(Instant::now() - Duration::from_secs(1));
+        MockClock::advance(Duration::from_secs(5));
+
+        timer.update_utilization();
+        let avg = timer.ewma.average().unwrap();
+        assert_approx_eq(avg, 0.1, "~0.1");
+    }
+
+    #[test]
+    #[serial]
+    fn test_utilization_in_bounds_on_late_stop() {
+        let mut timer = setup_timer();
+
+        MockClock::advance(Duration::from_secs(5));
+
+        timer.waiting = true;
+        timer.update_utilization();
+
+        let avg = timer.ewma.average().unwrap();
+        assert_approx_eq(avg, 0.0, "near 0 (always waiting)");
+
+        // Late message for stop wait
+        timer.stop_wait(Instant::now() - Duration::from_secs(4));
+        MockClock::advance(Duration::from_secs(5));
+
+        timer.update_utilization();
+        let avg = timer.ewma.average().unwrap();
+        assert_approx_eq(avg, 0.9, "~0.9");
+    }
+
     #[test]
     #[serial]
     fn test_normal_utilization_within_bounds() {