From 7bcf82aebd18147deed0786033beb0db70532a96 Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Tue, 19 May 2026 02:31:07 +0000 Subject: [PATCH] feat: add internode operation metrics --- crates/common/src/internode_metrics.rs | 66 +++++++++++++ crates/ecstore/src/rpc/remote_disk.rs | 43 ++++++-- crates/rio/src/http_reader.rs | 130 +++++++++++++++++++------ rustfs/src/storage/rpc/disk.rs | 56 +++++++---- rustfs/src/storage/rpc/http_service.rs | 61 +++++++++--- 5 files changed, 285 insertions(+), 71 deletions(-) diff --git a/crates/common/src/internode_metrics.rs b/crates/common/src/internode_metrics.rs index a089e3067..228a9e7a2 100644 --- a/crates/common/src/internode_metrics.rs +++ b/crates/common/src/internode_metrics.rs @@ -19,6 +19,19 @@ use std::sync::{ }; use std::time::{Duration, SystemTime, UNIX_EPOCH}; +pub const INTERNODE_OPERATION_READ_FILE_STREAM: &str = "read_file_stream"; +pub const INTERNODE_OPERATION_PUT_FILE_STREAM: &str = "put_file_stream"; +pub const INTERNODE_OPERATION_WALK_DIR: &str = "walk_dir"; +pub const INTERNODE_OPERATION_GRPC_READ_ALL: &str = "grpc_read_all"; +pub const INTERNODE_OPERATION_GRPC_WRITE_ALL: &str = "grpc_write_all"; + +const OPERATION_LABEL: &str = "operation"; +const INTERNODE_OPERATION_SENT_BYTES_TOTAL: &str = "rustfs_system_network_internode_operation_sent_bytes_total"; +const INTERNODE_OPERATION_RECV_BYTES_TOTAL: &str = "rustfs_system_network_internode_operation_recv_bytes_total"; +const INTERNODE_OPERATION_REQUESTS_OUTGOING_TOTAL: &str = "rustfs_system_network_internode_operation_requests_outgoing_total"; +const INTERNODE_OPERATION_REQUESTS_INCOMING_TOTAL: &str = "rustfs_system_network_internode_operation_requests_incoming_total"; +const INTERNODE_OPERATION_ERRORS_TOTAL: &str = "rustfs_system_network_internode_operation_errors_total"; + #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct InternodeMetricsSnapshot { pub sent_bytes_total: u64, @@ -54,6 +67,16 @@ impl InternodeMetrics { counter!("rustfs_system_network_internode_sent_bytes_total").increment(bytes); } + pub fn record_sent_bytes_for_operation(&self, operation: &'static str, bytes: usize) { + self.record_sent_bytes(bytes); + + let bytes = bytes as u64; + if bytes == 0 { + return; + } + counter!(INTERNODE_OPERATION_SENT_BYTES_TOTAL, OPERATION_LABEL => operation).increment(bytes); + } + pub fn record_recv_bytes(&self, bytes: usize) { let bytes = bytes as u64; if bytes == 0 { @@ -63,21 +86,46 @@ impl InternodeMetrics { counter!("rustfs_system_network_internode_recv_bytes_total").increment(bytes); } + pub fn record_recv_bytes_for_operation(&self, operation: &'static str, bytes: usize) { + self.record_recv_bytes(bytes); + + let bytes = bytes as u64; + if bytes == 0 { + return; + } + counter!(INTERNODE_OPERATION_RECV_BYTES_TOTAL, OPERATION_LABEL => operation).increment(bytes); + } + pub fn record_outgoing_request(&self) { self.outgoing_requests_total.fetch_add(1, Ordering::Relaxed); counter!("rustfs_system_network_internode_requests_outgoing_total").increment(1); } + pub fn record_outgoing_request_for_operation(&self, operation: &'static str) { + self.record_outgoing_request(); + counter!(INTERNODE_OPERATION_REQUESTS_OUTGOING_TOTAL, OPERATION_LABEL => operation).increment(1); + } + pub fn record_incoming_request(&self) { self.incoming_requests_total.fetch_add(1, Ordering::Relaxed); counter!("rustfs_system_network_internode_requests_incoming_total").increment(1); } + pub fn record_incoming_request_for_operation(&self, operation: &'static str) { + self.record_incoming_request(); + counter!(INTERNODE_OPERATION_REQUESTS_INCOMING_TOTAL, OPERATION_LABEL => operation).increment(1); + } + pub fn record_error(&self) { self.errors_total.fetch_add(1, Ordering::Relaxed); counter!("rustfs_system_network_internode_errors_total").increment(1); } + pub fn record_error_for_operation(&self, operation: &'static str) { + self.record_error(); + counter!(INTERNODE_OPERATION_ERRORS_TOTAL, OPERATION_LABEL => operation).increment(1); + } + pub fn record_dial_result(&self, duration: Duration, success: bool) { let elapsed_nanos = duration.as_nanos().min(u128::from(u64::MAX)) as u64; self.dial_total_time_nanos.fetch_add(elapsed_nanos, Ordering::Relaxed); @@ -163,4 +211,22 @@ mod tests { metrics.reset_for_test(); } + + #[test] + fn operation_metrics_also_update_aggregate_snapshot() { + let metrics = InternodeMetrics::default(); + + metrics.record_sent_bytes_for_operation(INTERNODE_OPERATION_READ_FILE_STREAM, 128); + metrics.record_recv_bytes_for_operation(INTERNODE_OPERATION_PUT_FILE_STREAM, 256); + metrics.record_outgoing_request_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL); + metrics.record_incoming_request_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL); + metrics.record_error_for_operation(INTERNODE_OPERATION_WALK_DIR); + + let snapshot = metrics.snapshot(); + assert_eq!(snapshot.sent_bytes_total, 128); + assert_eq!(snapshot.recv_bytes_total, 256); + assert_eq!(snapshot.outgoing_requests_total, 1); + assert_eq!(snapshot.incoming_requests_total, 1); + assert_eq!(snapshot.errors_total, 1); + } } diff --git a/crates/ecstore/src/rpc/remote_disk.rs b/crates/ecstore/src/rpc/remote_disk.rs index 3591e6353..3a4b71a56 100644 --- a/crates/ecstore/src/rpc/remote_disk.rs +++ b/crates/ecstore/src/rpc/remote_disk.rs @@ -36,6 +36,9 @@ use bytes::Bytes; use futures::lock::Mutex; use http::{HeaderMap, HeaderValue, Method, header::CONTENT_TYPE}; use metrics::counter; +use rustfs_common::internode_metrics::{ + INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_WRITE_ALL, global_internode_metrics, +}; use rustfs_filemeta::{FileInfo, ObjectPartInfo, RawFileInfo}; use rustfs_protos::evict_failed_connection; use rustfs_protos::proto_gen::node_service::RenamePartRequest; @@ -1576,11 +1579,12 @@ impl DiskAPI for RemoteDisk { self.execute_with_timeout( || async { + let data_len = data.len(); let disk = self.disk_ref().await; - let mut client = self - .get_client() - .await - .map_err(|err| Error::other(format!("can not get client, err: {err}")))?; + let mut client = self.get_client().await.map_err(|err| { + global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL); + Error::other(format!("can not get client, err: {err}")) + })?; let request = Request::new(WriteAllRequest { disk, volume: volume.to_string(), @@ -1588,12 +1592,21 @@ impl DiskAPI for RemoteDisk { data, }); - let response = client.write_all(request).await?.into_inner(); + global_internode_metrics().record_outgoing_request_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL); + let response = match client.write_all(request).await { + Ok(response) => response.into_inner(), + Err(err) => { + global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL); + return Err(err.into()); + } + }; if !response.success { + global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL); return Err(response.error.unwrap_or_default().into()); } + global_internode_metrics().record_sent_bytes_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL, data_len); Ok(()) }, get_max_timeout_duration(), @@ -1608,22 +1621,32 @@ impl DiskAPI for RemoteDisk { self.execute_with_timeout( || async { let disk = self.disk_ref().await; - let mut client = self - .get_client() - .await - .map_err(|err| Error::other(format!("can not get client, err: {err}")))?; + let mut client = self.get_client().await.map_err(|err| { + global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL); + Error::other(format!("can not get client, err: {err}")) + })?; let request = Request::new(ReadAllRequest { disk, volume: volume.to_string(), path: path.to_string(), }); - let response = client.read_all(request).await?.into_inner(); + global_internode_metrics().record_outgoing_request_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL); + let response = match client.read_all(request).await { + Ok(response) => response.into_inner(), + Err(err) => { + global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL); + return Err(err.into()); + } + }; if !response.success { + global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL); return Err(response.error.unwrap_or_default().into()); } + global_internode_metrics() + .record_recv_bytes_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL, response.data.len()); Ok(response.data) }, get_max_timeout_duration(), diff --git a/crates/rio/src/http_reader.rs b/crates/rio/src/http_reader.rs index bd770f24b..149a8b44a 100644 --- a/crates/rio/src/http_reader.rs +++ b/crates/rio/src/http_reader.rs @@ -18,7 +18,10 @@ use futures::{Stream, TryStreamExt as _}; use http::HeaderMap; use pin_project_lite::pin_project; use reqwest::{Certificate, Client, Identity, Method, RequestBuilder}; -use rustfs_common::internode_metrics::global_internode_metrics; +use rustfs_common::internode_metrics::{ + INTERNODE_OPERATION_PUT_FILE_STREAM, INTERNODE_OPERATION_READ_FILE_STREAM, INTERNODE_OPERATION_WALK_DIR, + global_internode_metrics, +}; use rustfs_utils::get_env_opt_str; use std::io::IoSlice; use std::io::{self, Error}; @@ -35,6 +38,10 @@ use tokio_util::io::StreamReader; use tokio_util::sync::PollSender; use tracing::error; +const READ_FILE_STREAM_PATH: &str = "/rustfs/rpc/read_file_stream"; +const PUT_FILE_STREAM_PATH: &str = "/rustfs/rpc/put_file_stream"; +const WALK_DIR_PATH: &str = "/rustfs/rpc/walk_dir"; + /// Get the TLS path from the RUSTFS_TLS_PATH environment variable. /// If the variable is not set, return None. fn tls_path() -> Option<&'static std::path::PathBuf> { @@ -145,6 +152,7 @@ pin_project! { method: Method, headers: HeaderMap, track_internode_metrics: bool, + internode_operation: Option<&'static str>, stall_timeout: Option, stall_timer: Option>>, #[pin] @@ -188,6 +196,7 @@ impl HttpReader { stall_timeout: Option, ) -> io::Result { let track_internode_metrics = is_internode_rpc_url(&url); + let internode_operation = internode_rpc_operation(&url); let client = get_http_client(&url); let mut request: RequestBuilder = client.request(method.clone(), url.clone()).headers(headers.clone()); if let Some(body) = body { @@ -195,30 +204,22 @@ impl HttpReader { } let resp = request.send().await.map_err(|e| { - if track_internode_metrics { - global_internode_metrics().record_error(); - } + record_internode_error(track_internode_metrics, internode_operation); Error::other(format!("HttpReader HTTP request error: {e}")) })?; if resp.status().is_success().not() { - if track_internode_metrics { - global_internode_metrics().record_error(); - } + record_internode_error(track_internode_metrics, internode_operation); return Err(Error::other(format!( "HttpReader HTTP request failed with non-200 status {}", resp.status() ))); } - if track_internode_metrics { - global_internode_metrics().record_outgoing_request(); - } + record_internode_outgoing_request(track_internode_metrics, internode_operation); let stream = resp.bytes_stream().map_err(move |e| { - if track_internode_metrics { - global_internode_metrics().record_error(); - } + record_internode_error(track_internode_metrics, internode_operation); Error::other(format!("HttpReader stream error: {e}")) }); @@ -228,6 +229,7 @@ impl HttpReader { method, headers, track_internode_metrics, + internode_operation, stall_timer: stall_timeout.map(|timeout| Box::pin(time::sleep(timeout))), stall_timeout, }) @@ -251,8 +253,8 @@ impl AsyncRead for HttpReader { match this.inner.as_mut().poll_read(cx, buf) { Poll::Ready(Ok(())) => { let bytes_read = buf.filled().len().saturating_sub(filled_before); - if *this.track_internode_metrics && bytes_read > 0 { - global_internode_metrics().record_recv_bytes(bytes_read); + if bytes_read > 0 { + record_internode_recv_bytes(*this.track_internode_metrics, *this.internode_operation, bytes_read); } if bytes_read > 0 { if let Some(stall_timeout) = *this.stall_timeout { @@ -267,9 +269,7 @@ impl AsyncRead for HttpReader { if let Some(timer) = this.stall_timer.as_mut() && timer.as_mut().poll(cx).is_ready() { - if *this.track_internode_metrics { - global_internode_metrics().record_error(); - } + record_internode_error(*this.track_internode_metrics, *this.internode_operation); Poll::Ready(Err(Error::new( io::ErrorKind::TimedOut, "HttpReader stall timeout: no data received before deadline", @@ -305,6 +305,7 @@ impl HashReaderDetector for HttpReader { struct ReceiverStream { receiver: mpsc::Receiver>, track_internode_metrics: bool, + internode_operation: Option<&'static str>, } impl Stream for ReceiverStream { @@ -327,9 +328,7 @@ impl Stream for ReceiverStream { // } match poll { Poll::Ready(Some(Some(bytes))) => { - if self.track_internode_metrics { - global_internode_metrics().record_sent_bytes(bytes.len()); - } + record_internode_sent_bytes(self.track_internode_metrics, self.internode_operation, bytes.len()); Poll::Ready(Some(Ok(bytes))) } Poll::Ready(Some(None)) => Poll::Ready(None), // Sender shutdown @@ -364,6 +363,7 @@ impl HttpWriter { let method_clone = method.clone(); let headers_clone = headers.clone(); let track_internode_metrics = is_internode_rpc_url(&url); + let internode_operation = internode_rpc_operation(&url); let (sender, receiver) = tokio::sync::mpsc::channel::>(HTTP_WRITER_CHANNEL_CAPACITY); let (err_tx, err_rx) = tokio::sync::oneshot::channel::(); @@ -372,6 +372,7 @@ impl HttpWriter { let stream = ReceiverStream { receiver, track_internode_metrics, + internode_operation, }; let body = reqwest::Body::wrap_stream(stream); // http_log!( @@ -391,9 +392,7 @@ impl HttpWriter { Ok(resp) => { // http_log!("[HttpWriter::spawn] got response: status={}", resp.status()); if !resp.status().is_success() { - if track_internode_metrics { - global_internode_metrics().record_error(); - } + record_internode_error(track_internode_metrics, internode_operation); let _ = err_tx.send(Error::other(format!( "HttpWriter HTTP request failed with non-200 status {}", resp.status() @@ -402,9 +401,7 @@ impl HttpWriter { } } Err(e) => { - if track_internode_metrics { - global_internode_metrics().record_error(); - } + record_internode_error(track_internode_metrics, internode_operation); // http_log!("[HttpWriter::spawn] HTTP request error: {e}"); let _ = err_tx.send(Error::other(format!("HTTP request failed: {e}"))); return Err(Error::other(format!("HTTP request failed: {e}"))); @@ -416,9 +413,7 @@ impl HttpWriter { }); // http_log!("[HttpWriter::new] connection established successfully"); - if track_internode_metrics { - global_internode_metrics().record_outgoing_request(); - } + record_internode_outgoing_request(track_internode_metrics, internode_operation); Ok(Self { url, method, @@ -448,6 +443,60 @@ fn is_internode_rpc_url(url: &str) -> bool { url.contains("/rustfs/rpc/") } +fn internode_rpc_operation(url: &str) -> Option<&'static str> { + let url = reqwest::Url::parse(url).ok()?; + match url.path() { + READ_FILE_STREAM_PATH => Some(INTERNODE_OPERATION_READ_FILE_STREAM), + PUT_FILE_STREAM_PATH => Some(INTERNODE_OPERATION_PUT_FILE_STREAM), + WALK_DIR_PATH => Some(INTERNODE_OPERATION_WALK_DIR), + _ => None, + } +} + +fn record_internode_outgoing_request(track: bool, operation: Option<&'static str>) { + if !track { + return; + } + + match operation { + Some(operation) => global_internode_metrics().record_outgoing_request_for_operation(operation), + None => global_internode_metrics().record_outgoing_request(), + } +} + +fn record_internode_sent_bytes(track: bool, operation: Option<&'static str>, bytes: usize) { + if !track { + return; + } + + match operation { + Some(operation) => global_internode_metrics().record_sent_bytes_for_operation(operation, bytes), + None => global_internode_metrics().record_sent_bytes(bytes), + } +} + +fn record_internode_recv_bytes(track: bool, operation: Option<&'static str>, bytes: usize) { + if !track { + return; + } + + match operation { + Some(operation) => global_internode_metrics().record_recv_bytes_for_operation(operation, bytes), + None => global_internode_metrics().record_recv_bytes(bytes), + } +} + +fn record_internode_error(track: bool, operation: Option<&'static str>) { + if !track { + return; + } + + match operation { + Some(operation) => global_internode_metrics().record_error_for_operation(operation), + None => global_internode_metrics().record_error(), + } +} + fn poll_send_error_to_io(err: tokio_util::sync::PollSendError, context: &str) -> io::Error { Error::other(format!("{context}: {err}")) } @@ -681,6 +730,27 @@ mod tests { (format!("http://{addr}/stream"), handle) } + #[test] + fn internode_rpc_operation_maps_known_routes() { + assert_eq!( + internode_rpc_operation(&format!("http://node:9000{READ_FILE_STREAM_PATH}?disk=d")), + Some(INTERNODE_OPERATION_READ_FILE_STREAM) + ); + assert_eq!( + internode_rpc_operation(&format!("http://node:9000{PUT_FILE_STREAM_PATH}?disk=d")), + Some(INTERNODE_OPERATION_PUT_FILE_STREAM) + ); + assert_eq!( + internode_rpc_operation(&format!("http://node:9000{WALK_DIR_PATH}?disk=d")), + Some(INTERNODE_OPERATION_WALK_DIR) + ); + assert_eq!(internode_rpc_operation("http://node:9000/rustfs/rpc/unknown"), None); + assert_eq!( + internode_rpc_operation("http://node:9000/rustfs/rpc/unknown?next=/rustfs/rpc/read_file_stream"), + None + ); + } + #[tokio::test] async fn http_reader_does_not_send_preflight_head() { let state = TestState::default(); diff --git a/rustfs/src/storage/rpc/disk.rs b/rustfs/src/storage/rpc/disk.rs index efc107a37..9debb3e01 100644 --- a/rustfs/src/storage/rpc/disk.rs +++ b/rustfs/src/storage/rpc/disk.rs @@ -13,6 +13,9 @@ // limitations under the License. use super::*; +use rustfs_common::internode_metrics::{ + INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_WRITE_ALL, global_internode_metrics, +}; use serde::de::DeserializeOwned; use std::io::Cursor; @@ -929,18 +932,27 @@ impl NodeService { pub(super) async fn handle_write_all(&self, request: Request) -> Result, Status> { let request = request.into_inner(); + let data_len = request.data.len(); if let Some(disk) = self.find_disk(&request.disk).await { match disk.write_all(&request.volume, &request.path, request.data).await { - Ok(_) => Ok(Response::new(WriteAllResponse { - success: true, - error: None, - })), - Err(err) => Ok(Response::new(WriteAllResponse { - success: false, - error: Some(err.into()), - })), + Ok(_) => { + global_internode_metrics().record_incoming_request_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL); + global_internode_metrics().record_recv_bytes_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL, data_len); + Ok(Response::new(WriteAllResponse { + success: true, + error: None, + })) + } + Err(err) => { + global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL); + Ok(Response::new(WriteAllResponse { + success: false, + error: Some(err.into()), + })) + } } } else { + global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL); Ok(Response::new(WriteAllResponse { success: false, error: Some(DiskError::other("can not find disk".to_string()).into()), @@ -954,18 +966,26 @@ impl NodeService { let request = request.into_inner(); if let Some(disk) = self.find_disk(&request.disk).await { match disk.read_all(&request.volume, &request.path).await { - Ok(data) => Ok(Response::new(ReadAllResponse { - success: true, - data, - error: None, - })), - Err(err) => Ok(Response::new(ReadAllResponse { - success: false, - data: Bytes::new(), - error: Some(err.into()), - })), + Ok(data) => { + global_internode_metrics().record_incoming_request_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL); + global_internode_metrics().record_sent_bytes_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL, data.len()); + Ok(Response::new(ReadAllResponse { + success: true, + data, + error: None, + })) + } + Err(err) => { + global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL); + Ok(Response::new(ReadAllResponse { + success: false, + data: Bytes::new(), + error: Some(err.into()), + })) + } } } else { + global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL); Ok(Response::new(ReadAllResponse { success: false, data: Bytes::new(), diff --git a/rustfs/src/storage/rpc/http_service.rs b/rustfs/src/storage/rpc/http_service.rs index 48e3713a1..e28c256fc 100644 --- a/rustfs/src/storage/rpc/http_service.rs +++ b/rustfs/src/storage/rpc/http_service.rs @@ -18,7 +18,10 @@ use futures_util::TryStreamExt; use http::{HeaderMap, Method, Request, Response, StatusCode, Uri}; use http_body_util::{BodyExt, Limited}; use hyper::body::Incoming; -use rustfs_common::internode_metrics::global_internode_metrics; +use rustfs_common::internode_metrics::{ + INTERNODE_OPERATION_PUT_FILE_STREAM, INTERNODE_OPERATION_READ_FILE_STREAM, INTERNODE_OPERATION_WALK_DIR, + global_internode_metrics, +}; use rustfs_config::MAX_ADMIN_REQUEST_BODY_SIZE; use rustfs_ecstore::disk::{DiskAPI, WalkDirOptions}; use rustfs_ecstore::rpc::verify_rpc_signature; @@ -106,8 +109,9 @@ fn is_internode_rpc_path(path: &str) -> bool { } async fn handle_internode_rpc(req: Request) -> Response { + let operation = internode_http_operation(req.uri().path()); if let Err(response) = verify_internode_rpc_signature(req.uri(), req.method(), req.headers()) { - global_internode_metrics().record_error(); + record_internode_rpc_error(operation); return *response; } @@ -122,12 +126,28 @@ async fn handle_internode_rpc(req: Request) -> Response { }; if !response.status().is_success() { - global_internode_metrics().record_error(); + record_internode_rpc_error(operation); } response } +fn internode_http_operation(path: &str) -> Option<&'static str> { + match path { + READ_FILE_STREAM_PATH => Some(INTERNODE_OPERATION_READ_FILE_STREAM), + PUT_FILE_STREAM_PATH => Some(INTERNODE_OPERATION_PUT_FILE_STREAM), + WALK_DIR_PATH => Some(INTERNODE_OPERATION_WALK_DIR), + _ => None, + } +} + +fn record_internode_rpc_error(operation: Option<&'static str>) { + match operation { + Some(operation) => global_internode_metrics().record_error_for_operation(operation), + None => global_internode_metrics().record_error(), + } +} + fn verify_internode_rpc_signature(uri: &Uri, method: &Method, headers: &HeaderMap) -> Result<(), RpcErrorResponse> { if method == Method::HEAD { return Ok(()); @@ -163,8 +183,8 @@ async fn handle_read_file(req: Request) -> Response { Err(e) => return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, format!("read file err {e}")), }; - global_internode_metrics().record_incoming_request(); - let stream = read_file_body_stream(file, query.length); + global_internode_metrics().record_incoming_request_for_operation(INTERNODE_OPERATION_READ_FILE_STREAM); + let stream = read_file_body_stream(file, query.length, INTERNODE_OPERATION_READ_FILE_STREAM); Response::builder() .status(StatusCode::OK) @@ -172,13 +192,17 @@ async fn handle_read_file(req: Request) -> Response { .expect("failed to build read file stream response") } -fn read_file_body_stream(reader: R, length: usize) -> Pin> + Send + Sync>> +fn read_file_body_stream( + reader: R, + length: usize, + operation: &'static str, +) -> Pin> + Send + Sync>> where R: tokio::io::AsyncRead + Unpin + Send + Sync + 'static, { let metrics = global_internode_metrics().clone(); let stream = ReaderStream::with_capacity(reader, DEFAULT_READ_BUFFER_SIZE).map_ok(move |bytes| { - metrics.record_sent_bytes(bytes.len()); + metrics.record_sent_bytes_for_operation(operation, bytes.len()); bytes }); @@ -220,10 +244,10 @@ async fn handle_walk_dir(req: Request) -> Response { } }); - global_internode_metrics().record_incoming_request(); + global_internode_metrics().record_incoming_request_for_operation(INTERNODE_OPERATION_WALK_DIR); let metrics = global_internode_metrics().clone(); let stream = ReaderStream::with_capacity(rd, DEFAULT_READ_BUFFER_SIZE).map_ok(move |bytes| { - metrics.record_sent_bytes(bytes.len()); + metrics.record_sent_bytes_for_operation(INTERNODE_OPERATION_WALK_DIR, bytes.len()); bytes }); @@ -260,8 +284,8 @@ async fn handle_put_file(req: Request) -> Response { Err(e) => return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, format!("write file err {e}")), }; - global_internode_metrics().record_incoming_request(); - global_internode_metrics().record_recv_bytes(copied as usize); + global_internode_metrics().record_incoming_request_for_operation(INTERNODE_OPERATION_PUT_FILE_STREAM); + global_internode_metrics().record_recv_bytes_for_operation(INTERNODE_OPERATION_PUT_FILE_STREAM, copied as usize); if let Err(e) = file.flush().await { return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, format!("write file err {e}")); @@ -337,6 +361,17 @@ mod tests { assert!(!is_internode_rpc_path("/rustfs/admin/v3/info")); } + #[test] + fn internode_http_operation_maps_only_known_routes() { + assert_eq!( + internode_http_operation(READ_FILE_STREAM_PATH), + Some(INTERNODE_OPERATION_READ_FILE_STREAM) + ); + assert_eq!(internode_http_operation(PUT_FILE_STREAM_PATH), Some(INTERNODE_OPERATION_PUT_FILE_STREAM)); + assert_eq!(internode_http_operation(WALK_DIR_PATH), Some(INTERNODE_OPERATION_WALK_DIR)); + assert_eq!(internode_http_operation("/rustfs/rpc/unknown"), None); + } + #[test] fn rpc_head_signature_verification_is_skipped() { let uri: Uri = READ_FILE_STREAM_PATH.parse().expect("uri"); @@ -377,7 +412,7 @@ mod tests { writer.write_all(b"hello world").await.expect("write succeeds"); }); - let mut stream = read_file_body_stream(reader, 0); + let mut stream = read_file_body_stream(reader, 0, INTERNODE_OPERATION_READ_FILE_STREAM); let mut out = Vec::new(); while let Some(chunk) = stream.next().await { out.extend_from_slice(&chunk.expect("chunk succeeds")); @@ -393,7 +428,7 @@ mod tests { writer.write_all(b"hello world").await.expect("write succeeds"); }); - let mut stream = read_file_body_stream(reader, 5); + let mut stream = read_file_body_stream(reader, 5, INTERNODE_OPERATION_READ_FILE_STREAM); let mut out = Vec::new(); while let Some(chunk) = stream.next().await { out.extend_from_slice(&chunk.expect("chunk succeeds"));