fix(obs): reuse network snapshots across ticks (#4431)

Co-authored-by: heihutu <heihutu@gmail.com>

crates/obs/src/metrics/scheduler.rs

@@ -92,7 +92,7 @@ use serde::Serialize;
 use std::borrow::Cow;
 use std::collections::{HashMap, HashSet};
 use std::time::Duration;
-use sysinfo::System;
+use sysinfo::{Networks, System};
 use tokio::time::Instant;
 use tokio_util::sync::CancellationToken;
 use tracing::warn;
@@ -739,6 +739,7 @@ pub fn init_metrics_runtime(token: CancellationToken) {
     tokio::spawn(async move {
         let labels = current_process_metric_labels();
         let mut host_system = System::new_all();
+        let mut host_networks = Networks::new();
         let process_interval = config.process_interval;
         let mut interval = tokio::time::interval(process_interval);
         let now = Instant::now();
@@ -769,9 +770,11 @@ pub fn init_metrics_runtime(token: CancellationToken) {
 
                     if now >= next_system_run {
                         #[cfg(feature = "gpu")]
-                        let mut metrics = collect_system_monitoring_metrics(&bundle, &labels, &mut host_system);
+                        let mut metrics =
+                            collect_system_monitoring_metrics(&bundle, &labels, &mut host_system, &mut host_networks);
                         #[cfg(not(feature = "gpu"))]
-                        let metrics = collect_system_monitoring_metrics(&bundle, &labels, &mut host_system);
+                        let metrics =
+                            collect_system_monitoring_metrics(&bundle, &labels, &mut host_system, &mut host_networks);
 
                         #[cfg(feature = "gpu")]
                         if let Some(pid) = current_pid {
@@ -907,6 +910,7 @@ fn collect_system_monitoring_metrics(
     bundle: &ProcessMetricBundle,
     labels: &[(&'static str, Cow<'static, str>)],
     host_system: &mut System,
+    host_networks: &mut Networks,
 ) -> Vec<PrometheusMetric> {
     let cpu_stats = ProcessCpuStats {
         usage: bundle.resource.cpu_percent,
@@ -920,7 +924,7 @@ fn collect_system_monitoring_metrics(
         read_bytes: bundle.disk_read_bytes,
         written_bytes: bundle.disk_write_bytes,
     };
-    let network_stats = collect_host_network_stats();
+    let network_stats = collect_host_network_stats(host_networks);
     let (system_cpu_stats, system_memory_stats) = collect_system_cpu_and_memory_stats_with(host_system);
 
     let mut metrics = Vec::new();

crates/obs/src/metrics/stats_collector.rs

@@ -712,16 +712,15 @@ pub fn collect_process_system_stats() -> ProcessStats {
     collect_process_metric_bundle().process
 }
 
-/// Collect host network statistics from the current network interface snapshot.
+/// Collect host network statistics from a refreshed network interface snapshot.
 ///
 /// These counters come from system interfaces and are host-wide, not process-scoped.
-pub fn collect_host_network_stats() -> HostNetworkStats {
-    let networks = Networks::new_with_refreshed_list();
+pub fn collect_host_network_stats_with(networks: &Networks) -> HostNetworkStats {
     let mut total_received = 0u64;
     let mut total_transmitted = 0u64;
     let mut per_interface = Vec::with_capacity(networks.len());
 
-    for (interface_name, data) in &networks {
+    for (interface_name, data) in networks {
         let received = data.received();
         let transmitted = data.transmitted();
         total_received += received;
@@ -736,6 +735,15 @@ pub fn collect_host_network_stats() -> HostNetworkStats {
     }
 }
 
+/// Collect host network statistics using a persistent `sysinfo::Networks` snapshot.
+///
+/// `sysinfo` reports network I/O as deltas since the previous refresh, so
+/// callers must reuse the same `Networks` instance across collection ticks.
+pub fn collect_host_network_stats(networks: &mut Networks) -> HostNetworkStats {
+    networks.refresh(true);
+    collect_host_network_stats_with(networks)
+}
+
 /// Collect internode network metrics from the global internode metrics snapshot.
 ///
 /// The returned values come directly from `global_internode_metrics().snapshot()`
@@ -1068,6 +1076,43 @@ pub async fn collect_compression_cluster_stats() -> Option<CompressionClusterSta
 #[cfg(test)]
 mod tests {
     use super::*;
+    use std::io::{Read, Write};
+    use std::net::{Shutdown, TcpListener, TcpStream};
+    use std::thread;
+    use std::time::Duration;
+
+    fn generate_loopback_traffic() -> std::io::Result<()> {
+        let listener = TcpListener::bind(("127.0.0.1", 0))?;
+        let addr = listener.local_addr()?;
+        let payload = vec![0x5Au8; 64 * 1024];
+        let expected_len = payload.len();
+
+        let server = thread::spawn(move || -> std::io::Result<()> {
+            let (mut stream, _) = listener.accept()?;
+            let mut received = 0usize;
+            let mut buf = [0u8; 8192];
+
+            while received < expected_len {
+                let read = stream.read(&mut buf)?;
+                if read == 0 {
+                    break;
+                }
+                received += read;
+            }
+
+            Ok(())
+        });
+
+        let mut client = TcpStream::connect(addr)?;
+        client.write_all(&payload)?;
+        client.flush()?;
+        client.shutdown(Shutdown::Write)?;
+
+        server
+            .join()
+            .expect("loopback traffic server thread should complete successfully")?;
+        Ok(())
+    }
 
     #[test]
     fn disk_is_online_for_metrics_accepts_online_state_case_insensitive() {
@@ -1179,6 +1224,45 @@ mod tests {
         assert_eq!(scanner_bucket_scans_started(&life_time_ops, 5), 5);
     }
 
+    #[test]
+    fn host_network_stats_require_a_persistent_networks_snapshot() -> std::io::Result<()> {
+        if !sysinfo::IS_SUPPORTED_SYSTEM {
+            return Ok(());
+        }
+
+        let mut persistent_networks = Networks::new();
+        persistent_networks.refresh(true);
+        let initial_stats = collect_host_network_stats_with(&persistent_networks);
+        assert_eq!(
+            initial_stats.total_received + initial_stats.total_transmitted,
+            0,
+            "the first refresh only seeds sysinfo's baseline snapshot"
+        );
+
+        match generate_loopback_traffic() {
+            Ok(()) => {}
+            Err(err) if err.kind() == std::io::ErrorKind::PermissionDenied => {
+                return Ok(());
+            }
+            Err(err) => return Err(err),
+        }
+        thread::sleep(Duration::from_millis(100));
+
+        let refreshed_stats = collect_host_network_stats(&mut persistent_networks);
+        assert!(
+            refreshed_stats.total_received > 0 || refreshed_stats.total_transmitted > 0,
+            "a persistent Networks instance should report non-zero loopback deltas after traffic"
+        );
+
+        let recreated_stats = collect_host_network_stats_with(&Networks::new_with_refreshed_list());
+        assert_eq!(
+            recreated_stats.total_received + recreated_stats.total_transmitted,
+            0,
+            "recreating Networks loses the prior refresh baseline and yields zero deltas"
+        );
+        Ok(())
+    }
+
     #[test]
     fn scanner_work_rate_per_second_reports_rate() {
         assert_eq!(scanner_work_rate_per_second(90, 45.0), 2.0);