changes

src/rust/lqosd/src/throughput_tracker/flow_data/netflow5/protocol.rs

@@ -5,10 +5,10 @@ use lqos_utils::unix_time::{time_since_boot, unix_now};
 use nix::sys::time::TimeValLike;
 use std::net::IpAddr;
 
-use crate::throughput_tracker::flow_data::FlowbeeLocalData;
 use crate::throughput_tracker::flow_data::netflow_common::{
-    boot_nanos_to_netflow_millis, clamp_i64_to_u32, clamp_u64_to_u32,
+    boot_nanos_to_netflow_millis, saturating_u64_to_netflow_u32, uptime_millis_to_netflow_u32,
 };
+use crate::throughput_tracker::flow_data::FlowbeeLocalData;
 
 const NETFLOW5_MAX_RECORDS_PER_PACKET: usize = 30;
 pub(crate) const NETFLOW5_MAX_FLOWS_PER_PACKET: usize = NETFLOW5_MAX_RECORDS_PER_PACKET / 2;
@@ -30,16 +30,18 @@ pub(crate) struct Netflow5Header {
 impl Netflow5Header {
     /// Create a new Netflow 5 header
     pub(crate) fn new(flow_sequence: u32, num_records: u16) -> Self {
-        let uptime_ms: u32 = time_since_boot()
-            .map(|u| clamp_i64_to_u32("NetFlow5", "sys_uptime", u.num_milliseconds()))
-            .unwrap_or(0);
+        let uptime_ms = time_since_boot().map(|u| u.num_milliseconds()).unwrap_or(0);
         let unix_secs = unix_now().unwrap_or(0);
 
+        Self::from_times(flow_sequence, num_records, uptime_ms, unix_secs)
+    }
+
+    fn from_times(flow_sequence: u32, num_records: u16, uptime_ms: i64, unix_secs: u64) -> Self {
         Self {
             version: (5u16).to_be(),
             count: num_records.to_be(),
-            sys_uptime: uptime_ms.to_be(),
-            unix_secs: clamp_u64_to_u32("NetFlow5", "unix_secs", unix_secs).to_be(),
+            sys_uptime: uptime_millis_to_netflow_u32(uptime_ms).to_be(),
+            unix_secs: saturating_u64_to_netflow_u32(unix_secs).to_be(),
             unix_nsecs: 0,
             flow_sequence,
             engine_type: 0,
@@ -84,12 +86,12 @@ pub(crate) fn to_netflow_5(
     if let (IpAddr::V4(local), IpAddr::V4(remote)) = (local, remote) {
         let src_ip = u32::from_ne_bytes(local.octets());
         let dst_ip = u32::from_ne_bytes(remote.octets());
-        let d_pkts2 = clamp_u64_to_u32("NetFlow5", "down packets", data.packets_sent.down).to_be();
-        let d_octets2 = clamp_u64_to_u32("NetFlow5", "down octets", data.bytes_sent.down).to_be();
-        let d_pkts = clamp_u64_to_u32("NetFlow5", "up packets", data.packets_sent.up).to_be();
-        let d_octets = clamp_u64_to_u32("NetFlow5", "up octets", data.bytes_sent.up).to_be();
-        let first = boot_nanos_to_netflow_millis("NetFlow5", "first", data.start_time).to_be();
-        let last = boot_nanos_to_netflow_millis("NetFlow5", "last", data.last_seen).to_be();
+        let d_pkts2 = saturating_u64_to_netflow_u32(data.packets_sent.down).to_be();
+        let d_octets2 = saturating_u64_to_netflow_u32(data.bytes_sent.down).to_be();
+        let d_pkts = saturating_u64_to_netflow_u32(data.packets_sent.up).to_be();
+        let d_octets = saturating_u64_to_netflow_u32(data.bytes_sent.up).to_be();
+        let first = boot_nanos_to_netflow_millis(data.start_time).to_be();
+        let last = boot_nanos_to_netflow_millis(data.last_seen).to_be();
 
         let record = Netflow5Record {
             src_addr: src_ip,
@@ -146,9 +148,9 @@ pub(crate) fn to_netflow_5(
 #[cfg(test)]
 mod tests {
     use super::*;
-    use crate::throughput_tracker::flow_data::FlowbeeLocalData;
     use crate::throughput_tracker::flow_data::netflow_common::NANOS_PER_MILLI;
-    use lqos_utils::{XdpIpAddress, units::DownUpOrder};
+    use crate::throughput_tracker::flow_data::FlowbeeLocalData;
+    use lqos_utils::{units::DownUpOrder, XdpIpAddress};
     use std::net::IpAddr;
 
     fn test_key() -> FlowbeeKey {
@@ -181,6 +183,16 @@ mod tests {
         }
     }
 
+    #[test]
+    fn netflow5_header_wraps_large_uptime_and_clamps_large_unix_time() {
+        let header =
+            Netflow5Header::from_times(10, 2, i64::from(u32::MAX) + 1, u64::from(u32::MAX) + 1);
+
+        assert_eq!(u32::from_be(header.sys_uptime), 0);
+        assert_eq!(u32::from_be(header.unix_secs), u32::MAX);
+        assert_eq!(header.flow_sequence, 10);
+    }
+
     #[test]
     fn netflow5_records_clamp_counters_and_use_milliseconds() {
         let key = test_key();
@@ -200,12 +212,18 @@ mod tests {
     }
 
     #[test]
-    fn netflow5_timestamp_conversion_clamps_after_milliseconds() {
-        let too_many_millis = (u64::from(u32::MAX) + 1) * NANOS_PER_MILLI;
+    fn netflow5_records_wrap_timestamps_after_u32_milliseconds() {
+        let key = test_key();
+        let mut data = test_flow_data();
+        data.start_time = (u64::from(u32::MAX) + 1) * NANOS_PER_MILLI;
+        data.last_seen = (u64::from(u32::MAX) + 2) * NANOS_PER_MILLI;
 
-        assert_eq!(
-            boot_nanos_to_netflow_millis("NetFlow5", "test timestamp", too_many_millis),
-            u32::MAX
-        );
+        let (forward, reverse) =
+            to_netflow_5(&key, &data).expect("IPv4 flow should convert to NetFlow5 records");
+
+        assert_eq!(u32::from_be(forward.first), 0);
+        assert_eq!(u32::from_be(forward.last), 1);
+        assert_eq!(u32::from_be(reverse.first), 0);
+        assert_eq!(u32::from_be(reverse.last), 1);
     }
 }

src/rust/lqosd/src/throughput_tracker/flow_data/netflow9/protocol/header.rs

@@ -1,7 +1,9 @@
 use lqos_utils::unix_time::{time_since_boot, unix_now};
 use nix::sys::time::TimeValLike;
 
-use crate::throughput_tracker::flow_data::netflow_common::{clamp_i64_to_u32, clamp_u64_to_u32};
+use crate::throughput_tracker::flow_data::netflow_common::{
+    saturating_u64_to_netflow_u32, uptime_millis_to_netflow_u32,
+};
 
 #[repr(C)]
 pub(crate) struct Netflow9Header {
@@ -36,8 +38,8 @@ impl Netflow9Header {
         Self {
             version: (9u16).to_be(),
             count: record_count_including_templates.to_be(),
-            sys_uptime: clamp_i64_to_u32("NetFlow9", "sys_uptime", uptime_ms).to_be(),
-            unix_secs: clamp_u64_to_u32("NetFlow9", "unix_secs", unix_secs).to_be(),
+            sys_uptime: uptime_millis_to_netflow_u32(uptime_ms).to_be(),
+            unix_secs: saturating_u64_to_netflow_u32(unix_secs).to_be(),
             package_sequence: flow_sequence.to_be(),
             source_id: 0,
         }
@@ -49,11 +51,11 @@ mod tests {
     use super::*;
 
     #[test]
-    fn netflow9_header_clamps_large_times() {
+    fn netflow9_header_wraps_large_uptime_and_clamps_large_unix_time() {
         let header =
             Netflow9Header::from_times(10, 2, i64::from(u32::MAX) + 1, u64::from(u32::MAX) + 1);
 
-        assert_eq!(u32::from_be(header.sys_uptime), u32::MAX);
+        assert_eq!(u32::from_be(header.sys_uptime), 0);
         assert_eq!(u32::from_be(header.unix_secs), u32::MAX);
         assert_eq!(u32::from_be(header.package_sequence), 10);
     }

src/rust/lqosd/src/throughput_tracker/flow_data/netflow_common.rs

@@ -2,27 +2,77 @@
 
 pub(crate) const NANOS_PER_MILLI: u64 = 1_000_000;
 
-pub(crate) fn clamp_i64_to_u32(protocol: &str, field: &str, value: i64) -> u32 {
+pub(crate) fn uptime_millis_to_netflow_u32(value: i64) -> u32 {
     if value < 0 {
-        tracing::warn!("{protocol} {field} value {value} is negative; clamping to 0");
         return 0;
     }
 
-    clamp_u64_to_u32(protocol, field, value as u64)
+    value as u32
 }
 
-pub(crate) fn clamp_u64_to_u32(protocol: &str, field: &str, value: u64) -> u32 {
-    match u32::try_from(value) {
-        Ok(value) => value,
-        Err(_) => {
-            tracing::warn!(
-                "{protocol} {field} value {value} exceeds u32::MAX; clamping to u32::MAX"
-            );
-            u32::MAX
+pub(crate) fn saturating_u64_to_netflow_u32(value: u64) -> u32 {
+    u32::try_from(value).unwrap_or(u32::MAX)
+}
+
+pub(crate) fn boot_nanos_to_netflow_millis(value: u64) -> u32 {
+    (value / NANOS_PER_MILLI) as u32
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use std::sync::{
+        atomic::{AtomicUsize, Ordering},
+        Arc,
+    };
+    use tracing::{Event, Level, Subscriber};
+    use tracing_subscriber::{
+        layer::{Context, Layer, SubscriberExt},
+        registry::LookupSpan,
+    };
+
+    struct WarningCounter {
+        count: Arc<AtomicUsize>,
+    }
+
+    impl<S> Layer<S> for WarningCounter
+    where
+        S: Subscriber + for<'span> LookupSpan<'span>,
+    {
+        fn on_event(&self, event: &Event<'_>, _context: Context<'_, S>) {
+            if *event.metadata().level() == Level::WARN {
+                self.count.fetch_add(1, Ordering::Relaxed);
+            }
         }
     }
-}
 
-pub(crate) fn boot_nanos_to_netflow_millis(protocol: &str, field: &str, value: u64) -> u32 {
-    clamp_u64_to_u32(protocol, field, value / NANOS_PER_MILLI)
+    #[test]
+    fn uptime_millis_wraps_like_netflow_wire_fields() {
+        assert_eq!(uptime_millis_to_netflow_u32(i64::from(u32::MAX)), u32::MAX);
+        assert_eq!(uptime_millis_to_netflow_u32(i64::from(u32::MAX) + 1), 0);
+        assert_eq!(uptime_millis_to_netflow_u32(-1), 0);
+    }
+
+    #[test]
+    fn boundary_conversions_do_not_emit_warnings() {
+        let warning_count = Arc::new(AtomicUsize::new(0));
+        let subscriber = tracing_subscriber::registry().with(WarningCounter {
+            count: Arc::clone(&warning_count),
+        });
+
+        tracing::subscriber::with_default(subscriber, || {
+            assert_eq!(uptime_millis_to_netflow_u32(-1), 0);
+            assert_eq!(uptime_millis_to_netflow_u32(i64::from(u32::MAX) + 1), 0);
+            assert_eq!(
+                saturating_u64_to_netflow_u32(u64::from(u32::MAX) + 1),
+                u32::MAX
+            );
+            assert_eq!(
+                boot_nanos_to_netflow_millis((u64::from(u32::MAX) + 1) * NANOS_PER_MILLI),
+                0
+            );
+        });
+
+        assert_eq!(warning_count.load(Ordering::Relaxed), 0);
+    }
 }