diff --git a/crates/rio/src/http_reader.rs b/crates/rio/src/http_reader.rs index 33bdce62d..103dccc2e 100644 --- a/crates/rio/src/http_reader.rs +++ b/crates/rio/src/http_reader.rs @@ -1496,6 +1496,7 @@ mod tests { use axum::{Router, body::Body, extract::State, http::StatusCode, response::IntoResponse, routing::get}; use futures::stream::{self, StreamExt as _}; use http_body_util::BodyExt as _; + use rustfs_io_metrics::internode_metrics::global_internode_metrics; use std::io::{self, IoSlice}; use std::sync::{ Arc, @@ -1741,6 +1742,7 @@ mod tests { let addr = listener.local_addr().expect("listener local address should be available"); let app = Router::new() .route("/stream", get(get_stream).head(reject_head).put(accept_put)) + .route(WALK_DIR_PATH, get(get_stream)) .route("/reject-put", get(get_stream).put(reject_put)) .route("/stall", get(get_stalling_stream)) .route("/delayed-first", get(get_delayed_first_chunk)) @@ -1801,6 +1803,33 @@ mod tests { handle.abort(); } + #[tokio::test] + async fn http_reader_records_walk_dir_recv_bytes() { + let state = TestState::default(); + let Some((base_url, handle)) = start_test_server(state.clone()).await else { + return; + }; + let metrics = global_internode_metrics(); + let before = metrics.snapshot().recv_bytes_total; + let url = base_url.replace("/stream", WALK_DIR_PATH); + + let mut reader = HttpReader::new(url, Method::GET, HeaderMap::new(), None) + .await + .expect("walk_dir reader should open"); + let mut buf = Vec::new(); + reader.read_to_end(&mut buf).await.expect("walk_dir body should read to EOF"); + let after = metrics.snapshot().recv_bytes_total; + + assert_eq!(buf, b"hello"); + assert_eq!(state.get_count.load(Ordering::SeqCst), 1); + assert!( + after >= before.saturating_add(5), + "walk_dir HttpReader should record streamed bytes as internode recv bytes: before={before}, after={after}" + ); + + handle.abort(); + } + #[tokio::test] async fn http_reader_stall_timeout_triggers_after_progress_stops() { let state = TestState::default(); diff --git a/rustfs/src/storage/rpc/http_service.rs b/rustfs/src/storage/rpc/http_service.rs index dd0cfc887..4e16962e3 100644 --- a/rustfs/src/storage/rpc/http_service.rs +++ b/rustfs/src/storage/rpc/http_service.rs @@ -840,6 +840,7 @@ mod tests { use http_body_util::BodyExt; use rustfs_io_metrics::internode_metrics::{ INTERNODE_OPERATION_PUT_FILE_STREAM, INTERNODE_OPERATION_READ_FILE_STREAM, INTERNODE_OPERATION_WALK_DIR, + global_internode_metrics, }; use sha2::Digest as _; use tokio::io; @@ -988,6 +989,30 @@ mod tests { assert_eq!(bytes, Bytes::from_static(b"complete walk data")); } + #[tokio::test] + async fn walk_dir_body_records_operation_sent_bytes() { + let metrics = global_internode_metrics(); + let before = metrics.snapshot().sent_bytes_total; + let payload = Bytes::from_static(b"metered walk data"); + let expected_len = u64::try_from(payload.len()).expect("test payload length should fit u64"); + let body = walk_dir_response_body(true, move |mut writer| async move { + writer.write_all(&payload).await?; + Ok(()) + }); + + let bytes = BodyExt::collect(body) + .await + .expect("successful completion should preserve the metered body") + .to_bytes(); + let after = metrics.snapshot().sent_bytes_total; + + assert_eq!(bytes, Bytes::from_static(b"metered walk data")); + assert!( + after >= before.saturating_add(expected_len), + "walk_dir response body should record streamed bytes as internode sent bytes: before={before}, after={after}, expected_delta={expected_len}" + ); + } + #[tokio::test] async fn walk_dir_completion_stream_surfaces_cancelled_producer() { let (completion_tx, completion_rx) = tokio::sync::oneshot::channel(); diff --git a/rustfs/src/storage/runtime_sources.rs b/rustfs/src/storage/runtime_sources.rs index 494311735..49aca244d 100644 --- a/rustfs/src/storage/runtime_sources.rs +++ b/rustfs/src/storage/runtime_sources.rs @@ -17,7 +17,10 @@ use crate::runtime_sources as root_runtime_sources; use crate::storage::storage_api::runtime_sources_consumer::ECStore; use rustfs_credentials::Credentials; use rustfs_iam::{error::Result as IamResult, store::object::ObjectStore, sys::IamSys}; -use rustfs_io_metrics::{PerformanceMetrics, internode_metrics::InternodeMetrics}; +use rustfs_io_metrics::{ + PerformanceMetrics, + internode_metrics::{InternodeMetrics, global_internode_metrics}, +}; use rustfs_kms::ObjectEncryptionService; use rustfs_lock::LockClient; use std::sync::Arc; @@ -41,7 +44,7 @@ pub(crate) fn current_buffer_config() -> RustFSBufferConfig { } pub(crate) fn current_internode_metrics() -> Arc { - root_runtime_sources::current_internode_metrics().unwrap_or_else(|| Arc::new(InternodeMetrics::default())) + root_runtime_sources::current_internode_metrics().unwrap_or_else(|| global_internode_metrics().clone()) } pub(crate) async fn current_local_node_name() -> String {