diff --git a/crates/obs/src/metrics/scheduler.rs b/crates/obs/src/metrics/scheduler.rs index 428bba37b..ca487eb73 100644 --- a/crates/obs/src/metrics/scheduler.rs +++ b/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 { 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(); diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs index 4b9242600..df6725ce9 100644 --- a/crates/obs/src/metrics/stats_collector.rs +++ b/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 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);