diff --git a/rustfs/src/connect/diagnostics/top_net.rs b/rustfs/src/connect/diagnostics/top_net.rs index 5f86362dc..13ca5e397 100644 --- a/rustfs/src/connect/diagnostics/top_net.rs +++ b/rustfs/src/connect/diagnostics/top_net.rs @@ -12,9 +12,10 @@ // See the License for the specific language governing permissions and // limitations under the License. -//! Bounded internode-network window backed by RustFS's monotonic counters. +//! Bounded host-network window backed by a persistent OS interface snapshot. use serde::Serialize; +use sysinfo::Networks; use tokio::time::Instant; use tokio_util::sync::CancellationToken; @@ -47,17 +48,31 @@ pub async fn capture_top_net( let Some(_permit) = request.acquire(cancel).await? else { return request.cancelled(TOOL_ID); }; - let Some(before) = network_snapshot() else { + if !sysinfo::IS_SUPPORTED_SYSTEM { return request.failed(TOOL_ID, 0, TopReasonCode::SourceUnavailable); - }; + } + let mut networks = Networks::new(); + networks.refresh(true); + if networks.is_empty() { + return request.failed(TOOL_ID, 0, TopReasonCode::SourceUnavailable); + } let started = Instant::now(); if !request.wait_window(TOOL_ID, cancel).await? { return request.cancelled(TOOL_ID); } - let Some(after) = network_snapshot() else { - return request.failed(TOOL_ID, elapsed_millis(started.elapsed()), TopReasonCode::SourceUnavailable); - }; - evaluate_network_window(request, before, after, elapsed_millis(started.elapsed())) + let observed = rustfs_obs::metrics::stats_collector::collect_host_network_stats(&mut networks); + evaluate_network_window( + request, + NetworkCounterSnapshot { + received_bytes: 0, + sent_bytes: 0, + }, + NetworkCounterSnapshot { + received_bytes: observed.total_received, + sent_bytes: observed.total_transmitted, + }, + elapsed_millis(started.elapsed()), + ) } pub fn evaluate_network_window( @@ -91,14 +106,6 @@ pub fn evaluate_network_window( ) } -fn network_snapshot() -> Option { - let snapshot = rustfs_obs::metrics::stats_collector::collect_internode_network_stats()?; - Some(NetworkCounterSnapshot { - received_bytes: snapshot.internode_recv_bytes_total, - sent_bytes: snapshot.internode_sent_bytes_total, - }) -} - fn elapsed_millis(duration: std::time::Duration) -> u64 { u64::try_from(duration.as_millis()).unwrap_or(u64::MAX).max(1) } diff --git a/rustfs/tests/connect_top_net.rs b/rustfs/tests/connect_top_net.rs index 309407d5a..3856fc008 100644 --- a/rustfs/tests/connect_top_net.rs +++ b/rustfs/tests/connect_top_net.rs @@ -13,7 +13,9 @@ // limitations under the License. use std::fs; +use std::io::Write as _; use std::io::{Cursor, Read as _}; +use std::net::{Shutdown, TcpListener, TcpStream}; #[cfg(unix)] use std::os::unix::fs::PermissionsExt as _; use std::path::Path; @@ -26,7 +28,7 @@ use p256::pkcs8::DecodePublicKey as _; use rustfs::connect::DeviceIdentity; use rustfs::connect::diagnostics::{ LocalTopConsent, NetworkCounterSnapshot, TopCaptureLimits, TopCaptureRequest, TopCaptureScope, TopOutcome, TopReasonCode, - evaluate_network_window, save_signed_top_export, sign_top_export, + capture_top_net, evaluate_network_window, save_signed_top_export, sign_top_export, }; use sha2::{Digest as _, Sha256}; use time::OffsetDateTime; @@ -106,6 +108,48 @@ fn top_network_uses_exact_counter_deltas_and_fails_closed_on_reset() { assert!(reset.data.is_none()); } +#[tokio::test] +async fn top_network_capture_observes_real_loopback_traffic() { + if !sysinfo::IS_SUPPORTED_SYSTEM { + return; + } + + let mut request = request(); + request.window = Duration::from_millis(500); + let traffic = std::thread::spawn(|| { + std::thread::sleep(Duration::from_millis(100)); + generate_loopback_traffic().expect("generate loopback traffic"); + }); + + let result = capture_top_net(&request, &CancellationToken::new()) + .await + .expect("capture host network traffic"); + traffic.join().expect("traffic thread"); + + assert_eq!(result.outcome, TopOutcome::Succeeded); + let data = result.data.expect("network data"); + assert!(data.received_bytes > 0, "real loopback traffic must increase received bytes"); + assert!(data.sent_bytes > 0, "real loopback traffic must increase sent bytes"); +} + +fn generate_loopback_traffic() -> std::io::Result<()> { + let listener = TcpListener::bind(("127.0.0.1", 0))?; + let address = listener.local_addr()?; + let server = std::thread::spawn(move || -> std::io::Result<()> { + let (mut stream, _) = listener.accept()?; + let mut received = Vec::new(); + stream.read_to_end(&mut received)?; + if received.len() != 256 * 1024 { + return Err(std::io::Error::other("loopback payload was truncated")); + } + Ok(()) + }); + let mut client = TcpStream::connect(address)?; + client.write_all(&vec![0x5a; 256 * 1024])?; + client.shutdown(Shutdown::Write)?; + server.join().map_err(|_| std::io::Error::other("loopback server panicked"))? +} + #[test] fn top_network_export_is_bounded_redacted_and_signed_over_exact_envelope_bytes() { let request = request();