fix(connect): observe host traffic in top net (#7814)

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<NetworkCounterSnapshot> {
-    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)
 }

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();