fix(buffers): Prevent negative buffer size and event gauges (#23453)

* fix(buffers): Prevent negative buffer size and event gauges

Signed-off-by: Vitalii Parfonov <vparfono@redhat.com>

* add changelog

Signed-off-by: Vitalii Parfonov <vparfono@redhat.com>

* fix formatting

Signed-off-by: Vitalii Parfonov <vparfonov@redhat.com>

* Update changelog.d/23453_prevent_negative_buffer_size_and_event_gauges.fix.md

Co-authored-by: Pavlos Rontidis <pavlos.rontidis@gmail.com>

* Handle value capping. Add recorder-based unit tests

Signed-off-by: Vitalii Parfonov <vparfono@redhat.com>

* Improve cast operation

* rename function with better name

Signed-off-by: Vitalii Parfonov <vparfono@redhat.com>

---------

Signed-off-by: Vitalii Parfonov <vparfono@redhat.com>
Signed-off-by: Vitalii Parfonov <vparfonov@redhat.com>
Co-authored-by: Pavlos Rontidis <pavlos.rontidis@gmail.com>

Author: vparfonov

Cargo.lock

@@ -0,0 +1,3 @@
+Fixed race condition bug causing negative values to be reported by the `vector_buffer_byte_size` and `vector_buffer_events` gauges.
+
+authors: vparfonov

changelog.d/23453_prevent_negative_buffer_size_and_event_gauges.fix.md

@@ -32,6 +32,8 @@ tokio = { version = "1.45.1", default-features = false, features = ["rt", "macro
 tracing = { version = "0.1.34", default-features = false, features = ["attributes"] }
 vector-config = { path = "../vector-config", default-features = false }
 vector-common = { path = "../vector-common", default-features = false, features = ["byte_size_of"] }
+dashmap = { version = "6.1", default-features = false }
+ordered-float = { version = "4.6.0", default-features = false }
 
 [dev-dependencies]
 clap.workspace = true

lib/vector-buffers/Cargo.toml

@@ -0,0 +1,19 @@
+// Maximum i64 value that can be represented exactly in f64 (2^53).
+pub const F64_SAFE_INT_MAX: i64 = 1_i64 << 53;
+
+pub fn i64_to_f64_safe(value: i64) -> f64 {
+    let capped = value.clamp(0, F64_SAFE_INT_MAX);
+    debug_assert!(capped <= F64_SAFE_INT_MAX);
+    #[allow(clippy::cast_precision_loss)]
+    {
+        capped as f64
+    }
+}
+
+pub fn u64_to_f64_safe(value: u64) -> f64 {
+    let capped = value.min(F64_SAFE_INT_MAX as u64);
+    #[allow(clippy::cast_precision_loss)]
+    {
+        capped as f64
+    }
+}

lib/vector-buffers/src/cast_utils.rs

@@ -1,27 +1,57 @@
+use dashmap::DashMap;
+use std::sync::atomic::{AtomicI64, Ordering};
+use std::sync::LazyLock;
 use std::time::Duration;
 
+use crate::cast_utils::{i64_to_f64_safe, u64_to_f64_safe};
 use metrics::{counter, gauge, histogram, Histogram};
 use vector_common::{
     internal_event::{error_type, InternalEvent},
     registered_event,
 };
 
+static BUFFER_COUNTERS: LazyLock<DashMap<usize, (AtomicI64, AtomicI64)>> =
+    LazyLock::new(DashMap::new);
+
+fn update_and_get(counter: &AtomicI64, delta: i64) -> i64 {
+    let mut new_val = 0;
+    counter
+        .fetch_update(Ordering::SeqCst, Ordering::SeqCst, |current| {
+            let updated = current.saturating_add(delta).clamp(0, i64::MAX);
+            new_val = updated;
+            Some(updated)
+        })
+        .ok();
+    new_val
+}
+
+fn update_buffer_gauge(stage: usize, events_delta: i64, bytes_delta: i64) {
+    let counters = BUFFER_COUNTERS
+        .entry(stage)
+        .or_insert_with(|| (AtomicI64::new(0), AtomicI64::new(0)));
+
+    let new_events = update_and_get(&counters.0, events_delta);
+    let new_bytes = update_and_get(&counters.1, bytes_delta);
+
+    gauge!("buffer_events", "stage" => stage.to_string()).set(i64_to_f64_safe(new_events));
+    gauge!("buffer_byte_size", "stage" => stage.to_string()).set(i64_to_f64_safe(new_bytes));
+}
+
 pub struct BufferCreated {
     pub idx: usize,
     pub max_size_events: usize,
     pub max_size_bytes: u64,
 }
 
 impl InternalEvent for BufferCreated {
-    #[allow(clippy::cast_precision_loss)]
     fn emit(self) {
         if self.max_size_events != 0 {
             gauge!("buffer_max_event_size", "stage" => self.idx.to_string())
-                .set(self.max_size_events as f64);
+                .set(u64_to_f64_safe(self.max_size_events as u64));
         }
         if self.max_size_bytes != 0 {
             gauge!("buffer_max_byte_size", "stage" => self.idx.to_string())
-                .set(self.max_size_bytes as f64);
+                .set(u64_to_f64_safe(self.max_size_bytes));
         }
     }
 }
@@ -33,15 +63,15 @@ pub struct BufferEventsReceived {
 }
 
 impl InternalEvent for BufferEventsReceived {
-    #[allow(clippy::cast_precision_loss)]
     fn emit(self) {
         counter!("buffer_received_events_total", "stage" => self.idx.to_string())
             .increment(self.count);
         counter!("buffer_received_bytes_total", "stage" => self.idx.to_string())
             .increment(self.byte_size);
-        gauge!("buffer_events", "stage" => self.idx.to_string()).increment(self.count as f64);
-        gauge!("buffer_byte_size", "stage" => self.idx.to_string())
-            .increment(self.byte_size as f64);
+
+        let count_delta = i64::try_from(self.count).unwrap_or(i64::MAX);
+        let bytes_delta = i64::try_from(self.byte_size).unwrap_or(i64::MAX);
+        update_buffer_gauge(self.idx, count_delta, bytes_delta);
     }
 }
 
@@ -52,14 +82,14 @@ pub struct BufferEventsSent {
 }
 
 impl InternalEvent for BufferEventsSent {
-    #[allow(clippy::cast_precision_loss)]
     fn emit(self) {
         counter!("buffer_sent_events_total", "stage" => self.idx.to_string()).increment(self.count);
         counter!("buffer_sent_bytes_total", "stage" => self.idx.to_string())
             .increment(self.byte_size);
-        gauge!("buffer_events", "stage" => self.idx.to_string()).decrement(self.count as f64);
-        gauge!("buffer_byte_size", "stage" => self.idx.to_string())
-            .decrement(self.byte_size as f64);
+
+        let count_delta = i64::try_from(self.count).unwrap_or(i64::MAX);
+        let bytes_delta = i64::try_from(self.byte_size).unwrap_or(i64::MAX);
+        update_buffer_gauge(self.idx, -count_delta, -bytes_delta);
     }
 }
 
@@ -72,7 +102,6 @@ pub struct BufferEventsDropped {
 }
 
 impl InternalEvent for BufferEventsDropped {
-    #[allow(clippy::cast_precision_loss)]
     fn emit(self) {
         let intentional_str = if self.intentional { "true" } else { "false" };
         if self.intentional {
@@ -96,9 +125,10 @@ impl InternalEvent for BufferEventsDropped {
             "buffer_discarded_events_total", "intentional" => intentional_str,
         )
         .increment(self.count);
-        gauge!("buffer_events", "stage" => self.idx.to_string()).decrement(self.count as f64);
-        gauge!("buffer_byte_size", "stage" => self.idx.to_string())
-            .decrement(self.byte_size as f64);
+
+        let count_delta = i64::try_from(self.count).unwrap_or(i64::MAX);
+        let bytes_delta = i64::try_from(self.byte_size).unwrap_or(i64::MAX);
+        update_buffer_gauge(self.idx, -count_delta, -bytes_delta);
     }
 }
 
@@ -115,7 +145,6 @@ impl InternalEvent for BufferReadError {
             error_code = self.error_code,
             error_type = error_type::READER_FAILED,
             stage = "processing",
-
         );
         counter!(
             "buffer_errors_total", "error_code" => self.error_code,
@@ -137,3 +166,190 @@ registered_event! {
         self.send_duration.record(duration);
     }
 }
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::cast_utils::F64_SAFE_INT_MAX;
+    use metrics::{Key, Label};
+    use metrics_util::debugging::{DebugValue, DebuggingRecorder};
+    use metrics_util::{CompositeKey, MetricKind};
+    use ordered_float::OrderedFloat;
+    use std::sync::Mutex;
+    use std::thread;
+
+    static TEST_LOCK: LazyLock<Mutex<()>> = LazyLock::new(|| Mutex::new(()));
+
+    fn reset_counters() {
+        BUFFER_COUNTERS.clear();
+    }
+
+    fn get_counter_values(stage: usize) -> (i64, i64) {
+        match BUFFER_COUNTERS.get(&stage) {
+            Some(counters) => {
+                let events = counters.0.load(Ordering::Relaxed);
+                let bytes = counters.1.load(Ordering::Relaxed);
+                (events, bytes)
+            }
+            None => (0, 0),
+        }
+    }
+
+    fn assert_gauge_state(
+        stage: usize,
+        updates: &[(i64, i64)],
+        expected_events: f64,
+        expected_bytes: f64,
+    ) {
+        let _guard = TEST_LOCK.lock().unwrap();
+        reset_counters();
+
+        let recorder = DebuggingRecorder::default();
+        let snapshotter = recorder.snapshotter();
+
+        metrics::with_local_recorder(&recorder, move || {
+            for (events_delta, bytes_delta) in updates {
+                update_buffer_gauge(stage, *events_delta, *bytes_delta);
+            }
+
+            let mut metrics = snapshotter.snapshot().into_vec();
+
+            let stage_label = Label::new("stage", stage.to_string());
+            let mut expected_metrics = vec![
+                (
+                    CompositeKey::new(
+                        MetricKind::Gauge,
+                        Key::from_parts("buffer_events", vec![stage_label.clone()]),
+                    ),
+                    None,
+                    None,
+                    DebugValue::Gauge(OrderedFloat(expected_events)),
+                ),
+                (
+                    CompositeKey::new(
+                        MetricKind::Gauge,
+                        Key::from_parts("buffer_byte_size", vec![stage_label]),
+                    ),
+                    None,
+                    None,
+                    DebugValue::Gauge(OrderedFloat(expected_bytes)),
+                ),
+            ];
+
+            metrics.sort_by_key(|(key, ..)| key.clone());
+            expected_metrics.sort_by_key(|(key, ..)| key.clone());
+
+            assert_eq!(metrics, expected_metrics);
+        });
+    }
+
+    #[test]
+    fn test_increment() {
+        let _guard = TEST_LOCK.lock().unwrap();
+        reset_counters();
+
+        update_buffer_gauge(0, 10, 1024);
+        let (events, bytes) = get_counter_values(0);
+        assert_eq!(events, 10);
+        assert_eq!(bytes, 1024);
+    }
+
+    #[test]
+    fn test_increment_and_decrement() {
+        let _guard = TEST_LOCK.lock().unwrap();
+        reset_counters();
+
+        update_buffer_gauge(1, 100, 2048);
+        update_buffer_gauge(1, -50, -1024);
+        let (events, bytes) = get_counter_values(1);
+        assert_eq!(events, 50);
+        assert_eq!(bytes, 1024);
+    }
+
+    #[test]
+    fn test_decrement_below_zero_clamped_to_zero() {
+        let _guard = TEST_LOCK.lock().unwrap();
+        reset_counters();
+
+        update_buffer_gauge(2, 5, 100);
+        update_buffer_gauge(2, -10, -200);
+        let (events, bytes) = get_counter_values(2);
+
+        assert_eq!(events, 0);
+        assert_eq!(bytes, 0);
+    }
+
+    #[test]
+    fn test_multiple_stages_are_independent() {
+        let _guard = TEST_LOCK.lock().unwrap();
+        reset_counters();
+
+        update_buffer_gauge(0, 10, 100);
+        update_buffer_gauge(1, 20, 200);
+        let (events0, bytes0) = get_counter_values(0);
+        let (events1, bytes1) = get_counter_values(1);
+        assert_eq!(events0, 10);
+        assert_eq!(bytes0, 100);
+        assert_eq!(events1, 20);
+        assert_eq!(bytes1, 200);
+    }
+
+    #[test]
+    fn test_multithreaded_updates_are_correct() {
+        let _guard = TEST_LOCK.lock().unwrap();
+        reset_counters();
+
+        let num_threads = 10;
+        let increments_per_thread = 1000;
+        let mut handles = vec![];
+
+        for _ in 0..num_threads {
+            let handle = thread::spawn(move || {
+                for _ in 0..increments_per_thread {
+                    update_buffer_gauge(0, 1, 10);
+                }
+            });
+            handles.push(handle);
+        }
+
+        for handle in handles {
+            handle.join().unwrap();
+        }
+
+        let (final_events, final_bytes) = get_counter_values(0);
+        let expected_events = i64::from(num_threads * increments_per_thread);
+        let expected_bytes = i64::from(num_threads * increments_per_thread * 10);
+
+        assert_eq!(final_events, expected_events);
+        assert_eq!(final_bytes, expected_bytes);
+    }
+
+    #[test]
+    fn test_large_values_capped_to_f64_safe_max() {
+        let _guard = TEST_LOCK.lock().unwrap();
+        reset_counters();
+
+        update_buffer_gauge(3, F64_SAFE_INT_MAX * 2, F64_SAFE_INT_MAX * 2);
+
+        let (events, bytes) = get_counter_values(3);
+
+        assert!(events > F64_SAFE_INT_MAX);
+        assert!(bytes > F64_SAFE_INT_MAX);
+
+        let capped_events = events.min(F64_SAFE_INT_MAX);
+        let capped_bytes = bytes.min(F64_SAFE_INT_MAX);
+
+        assert_eq!(capped_events, F64_SAFE_INT_MAX);
+        assert_eq!(capped_bytes, F64_SAFE_INT_MAX);
+    }
+
+    #[test]
+    fn test_increment_with_recorder() {
+        assert_gauge_state(0, &[(100, 2048), (200, 1024)], 300.0, 3072.0);
+    }
+
+    #[test]
+    fn test_should_not_be_negative_with_recorder() {
+        assert_gauge_state(0, &[(100, 1024), (-200, -4096)], 0.0, 0.0);
+    }
+}

lib/vector-buffers/src/internal_events.rs

@@ -31,6 +31,7 @@ mod internal_events;
 pub mod test;
 pub mod topology;
 
+mod cast_utils;
 pub(crate) mod variants;
 
 use std::fmt::Debug;