diff --git a/crates/ecstore/src/rpc/remote_disk.rs b/crates/ecstore/src/rpc/remote_disk.rs index 7afed545d..5a8844bf1 100644 --- a/crates/ecstore/src/rpc/remote_disk.rs +++ b/crates/ecstore/src/rpc/remote_disk.rs @@ -35,7 +35,8 @@ use futures::lock::Mutex; use metrics::counter; use rustfs_filemeta::{FileInfo, ObjectPartInfo, RawFileInfo}; use rustfs_io_metrics::internode_metrics::{ - INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_WRITE_ALL, global_internode_metrics, + INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC, + global_internode_metrics, }; use rustfs_protos::evict_failed_connection; use rustfs_protos::proto_gen::node_service::RenamePartRequest; @@ -1558,7 +1559,10 @@ impl DiskAPI for RemoteDisk { let data_len = data.len(); let disk = self.disk_ref().await; let mut client = self.get_client().await.map_err(|err| { - global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL); + global_internode_metrics().record_error_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_WRITE_ALL, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); Error::other(format!("can not get client, err: {err}")) })?; let request = Request::new(WriteAllRequest { @@ -1568,19 +1572,32 @@ impl DiskAPI for RemoteDisk { data, }); - global_internode_metrics().record_outgoing_request_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL); + global_internode_metrics().record_outgoing_request_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_WRITE_ALL, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); 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); + global_internode_metrics().record_error_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_WRITE_ALL, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); return Err(err.into()); } }; - global_internode_metrics().record_sent_bytes_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL, data_len); + global_internode_metrics().record_sent_bytes_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_WRITE_ALL, + INTERNODE_TRANSPORT_BACKEND_GRPC, + data_len, + ); if !response.success { - global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL); + global_internode_metrics().record_error_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_WRITE_ALL, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); return Err(response.error.unwrap_or_default().into()); } @@ -1599,7 +1616,10 @@ impl DiskAPI for RemoteDisk { || async { let disk = self.disk_ref().await; let mut client = self.get_client().await.map_err(|err| { - global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL); + global_internode_metrics().record_error_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_ALL, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); Error::other(format!("can not get client, err: {err}")) })?; let request = Request::new(ReadAllRequest { @@ -1608,22 +1628,34 @@ impl DiskAPI for RemoteDisk { path: path.to_string(), }); - global_internode_metrics().record_outgoing_request_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL); + global_internode_metrics().record_outgoing_request_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_ALL, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); 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); + global_internode_metrics().record_error_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_ALL, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); return Err(err.into()); } }; if !response.success { - global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL); + global_internode_metrics().record_error_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_ALL, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); return Err(response.error.unwrap_or_default().into()); } - global_internode_metrics() - .record_recv_bytes_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL, response.data.len()); + global_internode_metrics().record_recv_bytes_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_ALL, + INTERNODE_TRANSPORT_BACKEND_GRPC, + response.data.len(), + ); Ok(response.data) }, get_max_timeout_duration(), diff --git a/crates/io-metrics/README.md b/crates/io-metrics/README.md index b8698904f..8da89cef0 100644 --- a/crates/io-metrics/README.md +++ b/crates/io-metrics/README.md @@ -143,6 +143,47 @@ record_backpressure_event("warning", 0.85); record_timeout_event("GetObject", Duration::from_secs(30)); ``` +### Internode Transport Metrics + +Internode metrics are recorded by `src/internode_metrics.rs`. Aggregate metrics +remain unlabeled for compatibility with existing dashboards: + +| Metric | Meaning | +| --- | --- | +| `rustfs_system_network_internode_sent_bytes_total` | Total internode bytes sent by this node. | +| `rustfs_system_network_internode_recv_bytes_total` | Total internode bytes received by this node. | +| `rustfs_system_network_internode_requests_outgoing_total` | Total outgoing internode requests. | +| `rustfs_system_network_internode_requests_incoming_total` | Total incoming internode requests. | +| `rustfs_system_network_internode_errors_total` | Total internode errors. | +| `rustfs_system_network_internode_dial_errors_total` | Failed internode connection attempts. | +| `rustfs_system_network_internode_dial_avg_time_nanos` | Average internode dial duration. | + +Operation-level metrics use the same low-cardinality label set: + +| Metric | Labels | Meaning | +| --- | --- | --- | +| `rustfs_system_network_internode_operation_sent_bytes_total` | `operation`, `backend` | Bytes sent for an internode operation. | +| `rustfs_system_network_internode_operation_recv_bytes_total` | `operation`, `backend` | Bytes received for an internode operation. | +| `rustfs_system_network_internode_operation_requests_outgoing_total` | `operation`, `backend` | Outgoing request attempts for an internode operation. | +| `rustfs_system_network_internode_operation_requests_incoming_total` | `operation`, `backend` | Incoming request attempts for an internode operation. | +| `rustfs_system_network_internode_operation_errors_total` | `operation`, `backend` | Failed internode operation attempts. | + +Current `operation` values are `read_file_stream`, `put_file_stream`, +`walk_dir`, `grpc_read_all`, and `grpc_write_all`. Current `backend` values are +`tcp-http` for the `InternodeDataTransport` TCP/HTTP path and `grpc` for the +remaining gRPC byte paths. The compatibility wrapper uses `unknown` only for +callers that have not been classified yet. + +Success/failure is intentionally not a high-cardinality label today. Failures +are represented by `rustfs_system_network_internode_operation_errors_total`; +successful completions are not emitted as a dedicated result-labeled metric. +Adding completion/result labels is a follow-up once stream completion semantics +are defined consistently for request setup, body transfer, and shutdown. + +`scripts/run_internode_transport_baseline.sh --metrics-url ...` records metric +deltas with `operation` and `backend` columns, so the TCP baseline can attribute +bytes and request/error counts to `tcp-http` transport operations. + ### Unified Configuration Centralized configuration management: @@ -181,6 +222,7 @@ rustfs-io-metrics/ │ ├── deadlock_metrics.rs # Deadlock metrics │ ├── lock_metrics.rs # Lock metrics │ ├── timeout_metrics.rs # Timeout metrics +│ ├── internode_metrics.rs # Internode transport metrics │ ├── bandwidth.rs # Bandwidth monitoring │ ├── global_metrics.rs # Global metrics │ └── performance.rs # Performance metrics diff --git a/crates/io-metrics/src/internode_metrics.rs b/crates/io-metrics/src/internode_metrics.rs index 228a9e7a2..c7c0a5b3d 100644 --- a/crates/io-metrics/src/internode_metrics.rs +++ b/crates/io-metrics/src/internode_metrics.rs @@ -24,14 +24,49 @@ 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"; +pub const INTERNODE_TRANSPORT_BACKEND_TCP_HTTP: &str = "tcp-http"; +pub const INTERNODE_TRANSPORT_BACKEND_GRPC: &str = "grpc"; +pub const INTERNODE_TRANSPORT_BACKEND_UNKNOWN: &str = "unknown"; const OPERATION_LABEL: &str = "operation"; +const BACKEND_LABEL: &str = "backend"; 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, PartialEq, Eq)] +pub struct InternodeOperationMetricDescriptor { + pub name: &'static str, + pub labels: &'static [&'static str], +} + +const OPERATION_BACKEND_LABELS: &[&str] = &[OPERATION_LABEL, BACKEND_LABEL]; + +pub const INTERNODE_OPERATION_METRICS: &[InternodeOperationMetricDescriptor] = &[ + InternodeOperationMetricDescriptor { + name: INTERNODE_OPERATION_SENT_BYTES_TOTAL, + labels: OPERATION_BACKEND_LABELS, + }, + InternodeOperationMetricDescriptor { + name: INTERNODE_OPERATION_RECV_BYTES_TOTAL, + labels: OPERATION_BACKEND_LABELS, + }, + InternodeOperationMetricDescriptor { + name: INTERNODE_OPERATION_REQUESTS_OUTGOING_TOTAL, + labels: OPERATION_BACKEND_LABELS, + }, + InternodeOperationMetricDescriptor { + name: INTERNODE_OPERATION_REQUESTS_INCOMING_TOTAL, + labels: OPERATION_BACKEND_LABELS, + }, + InternodeOperationMetricDescriptor { + name: INTERNODE_OPERATION_ERRORS_TOTAL, + labels: OPERATION_BACKEND_LABELS, + }, +]; + #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct InternodeMetricsSnapshot { pub sent_bytes_total: u64, @@ -68,13 +103,17 @@ impl InternodeMetrics { } pub fn record_sent_bytes_for_operation(&self, operation: &'static str, bytes: usize) { + self.record_sent_bytes_for_operation_and_backend(operation, INTERNODE_TRANSPORT_BACKEND_UNKNOWN, bytes); + } + + pub fn record_sent_bytes_for_operation_and_backend(&self, operation: &'static str, backend: &'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); + counter!(INTERNODE_OPERATION_SENT_BYTES_TOTAL, OPERATION_LABEL => operation, BACKEND_LABEL => backend).increment(bytes); } pub fn record_recv_bytes(&self, bytes: usize) { @@ -87,13 +126,17 @@ impl InternodeMetrics { } pub fn record_recv_bytes_for_operation(&self, operation: &'static str, bytes: usize) { + self.record_recv_bytes_for_operation_and_backend(operation, INTERNODE_TRANSPORT_BACKEND_UNKNOWN, bytes); + } + + pub fn record_recv_bytes_for_operation_and_backend(&self, operation: &'static str, backend: &'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); + counter!(INTERNODE_OPERATION_RECV_BYTES_TOTAL, OPERATION_LABEL => operation, BACKEND_LABEL => backend).increment(bytes); } pub fn record_outgoing_request(&self) { @@ -102,8 +145,13 @@ impl InternodeMetrics { } pub fn record_outgoing_request_for_operation(&self, operation: &'static str) { + self.record_outgoing_request_for_operation_and_backend(operation, INTERNODE_TRANSPORT_BACKEND_UNKNOWN); + } + + pub fn record_outgoing_request_for_operation_and_backend(&self, operation: &'static str, backend: &'static str) { self.record_outgoing_request(); - counter!(INTERNODE_OPERATION_REQUESTS_OUTGOING_TOTAL, OPERATION_LABEL => operation).increment(1); + counter!(INTERNODE_OPERATION_REQUESTS_OUTGOING_TOTAL, OPERATION_LABEL => operation, BACKEND_LABEL => backend) + .increment(1); } pub fn record_incoming_request(&self) { @@ -112,8 +160,13 @@ impl InternodeMetrics { } pub fn record_incoming_request_for_operation(&self, operation: &'static str) { + self.record_incoming_request_for_operation_and_backend(operation, INTERNODE_TRANSPORT_BACKEND_UNKNOWN); + } + + pub fn record_incoming_request_for_operation_and_backend(&self, operation: &'static str, backend: &'static str) { self.record_incoming_request(); - counter!(INTERNODE_OPERATION_REQUESTS_INCOMING_TOTAL, OPERATION_LABEL => operation).increment(1); + counter!(INTERNODE_OPERATION_REQUESTS_INCOMING_TOTAL, OPERATION_LABEL => operation, BACKEND_LABEL => backend) + .increment(1); } pub fn record_error(&self) { @@ -122,8 +175,12 @@ impl InternodeMetrics { } pub fn record_error_for_operation(&self, operation: &'static str) { + self.record_error_for_operation_and_backend(operation, INTERNODE_TRANSPORT_BACKEND_UNKNOWN); + } + + pub fn record_error_for_operation_and_backend(&self, operation: &'static str, backend: &'static str) { self.record_error(); - counter!(INTERNODE_OPERATION_ERRORS_TOTAL, OPERATION_LABEL => operation).increment(1); + counter!(INTERNODE_OPERATION_ERRORS_TOTAL, OPERATION_LABEL => operation, BACKEND_LABEL => backend).increment(1); } pub fn record_dial_result(&self, duration: Duration, success: bool) { @@ -216,11 +273,25 @@ mod tests { 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); + metrics.record_sent_bytes_for_operation_and_backend( + INTERNODE_OPERATION_READ_FILE_STREAM, + INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, + 128, + ); + metrics.record_recv_bytes_for_operation_and_backend( + INTERNODE_OPERATION_PUT_FILE_STREAM, + INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, + 256, + ); + metrics.record_outgoing_request_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_WRITE_ALL, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); + metrics.record_incoming_request_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_ALL, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); + metrics.record_error_for_operation_and_backend(INTERNODE_OPERATION_WALK_DIR, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP); let snapshot = metrics.snapshot(); assert_eq!(snapshot.sent_bytes_total, 128); @@ -229,4 +300,12 @@ mod tests { assert_eq!(snapshot.incoming_requests_total, 1); assert_eq!(snapshot.errors_total, 1); } + + #[test] + fn operation_metric_descriptors_include_backend_and_operation_labels() { + assert_eq!(INTERNODE_OPERATION_METRICS.len(), 5); + for metric in INTERNODE_OPERATION_METRICS { + assert_eq!(metric.labels, &[OPERATION_LABEL, BACKEND_LABEL]); + } + } } diff --git a/crates/rio/src/http_reader.rs b/crates/rio/src/http_reader.rs index f9cdb8e23..2a62477a5 100644 --- a/crates/rio/src/http_reader.rs +++ b/crates/rio/src/http_reader.rs @@ -20,7 +20,7 @@ use pin_project_lite::pin_project; use reqwest::{Certificate, Client, Identity, Method, RequestBuilder}; use rustfs_io_metrics::internode_metrics::{ INTERNODE_OPERATION_PUT_FILE_STREAM, INTERNODE_OPERATION_READ_FILE_STREAM, INTERNODE_OPERATION_WALK_DIR, - global_internode_metrics, + INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, global_internode_metrics, }; use rustfs_utils::get_env_opt_str; use std::io::IoSlice; @@ -459,7 +459,8 @@ fn record_internode_outgoing_request(track: bool, operation: Option<&'static str } match operation { - Some(operation) => global_internode_metrics().record_outgoing_request_for_operation(operation), + Some(operation) => global_internode_metrics() + .record_outgoing_request_for_operation_and_backend(operation, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP), None => global_internode_metrics().record_outgoing_request(), } } @@ -470,7 +471,11 @@ fn record_internode_sent_bytes(track: bool, operation: Option<&'static str>, byt } match operation { - Some(operation) => global_internode_metrics().record_sent_bytes_for_operation(operation, bytes), + Some(operation) => global_internode_metrics().record_sent_bytes_for_operation_and_backend( + operation, + INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, + bytes, + ), None => global_internode_metrics().record_sent_bytes(bytes), } } @@ -481,7 +486,11 @@ fn record_internode_recv_bytes(track: bool, operation: Option<&'static str>, byt } match operation { - Some(operation) => global_internode_metrics().record_recv_bytes_for_operation(operation, bytes), + Some(operation) => global_internode_metrics().record_recv_bytes_for_operation_and_backend( + operation, + INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, + bytes, + ), None => global_internode_metrics().record_recv_bytes(bytes), } } @@ -492,7 +501,9 @@ fn record_internode_error(track: bool, operation: Option<&'static str>) { } match operation { - Some(operation) => global_internode_metrics().record_error_for_operation(operation), + Some(operation) => { + global_internode_metrics().record_error_for_operation_and_backend(operation, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP) + } None => global_internode_metrics().record_error(), } } diff --git a/rustfs/src/storage/rpc/disk.rs b/rustfs/src/storage/rpc/disk.rs index 0b4c4cd6d..f338f48cd 100644 --- a/rustfs/src/storage/rpc/disk.rs +++ b/rustfs/src/storage/rpc/disk.rs @@ -14,7 +14,8 @@ use super::*; use rustfs_io_metrics::internode_metrics::{ - INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_WRITE_ALL, global_internode_metrics, + INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC, + global_internode_metrics, }; use serde::de::DeserializeOwned; use std::io::Cursor; @@ -933,8 +934,15 @@ 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(); - 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); + global_internode_metrics().record_incoming_request_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_WRITE_ALL, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); + global_internode_metrics().record_recv_bytes_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_WRITE_ALL, + INTERNODE_TRANSPORT_BACKEND_GRPC, + 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 { @@ -942,7 +950,10 @@ impl NodeService { error: None, })), Err(err) => { - global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL); + global_internode_metrics().record_error_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_WRITE_ALL, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); Ok(Response::new(WriteAllResponse { success: false, error: Some(err.into()), @@ -950,7 +961,8 @@ impl NodeService { } } } else { - global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_WRITE_ALL); + global_internode_metrics() + .record_error_for_operation_and_backend(INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC); Ok(Response::new(WriteAllResponse { success: false, error: Some(DiskError::other("can not find disk".to_string()).into()), @@ -962,11 +974,18 @@ impl NodeService { debug!("read all"); let request = request.into_inner(); - global_internode_metrics().record_incoming_request_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL); + global_internode_metrics().record_incoming_request_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_ALL, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); if let Some(disk) = self.find_disk(&request.disk).await { match disk.read_all(&request.volume, &request.path).await { Ok(data) => { - global_internode_metrics().record_sent_bytes_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL, data.len()); + global_internode_metrics().record_sent_bytes_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_ALL, + INTERNODE_TRANSPORT_BACKEND_GRPC, + data.len(), + ); Ok(Response::new(ReadAllResponse { success: true, data, @@ -974,7 +993,10 @@ impl NodeService { })) } Err(err) => { - global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL); + global_internode_metrics().record_error_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_ALL, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); Ok(Response::new(ReadAllResponse { success: false, data: Bytes::new(), @@ -983,7 +1005,8 @@ impl NodeService { } } } else { - global_internode_metrics().record_error_for_operation(INTERNODE_OPERATION_GRPC_READ_ALL); + global_internode_metrics() + .record_error_for_operation_and_backend(INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC); 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 90824cde5..dfbe5a0e0 100644 --- a/rustfs/src/storage/rpc/http_service.rs +++ b/rustfs/src/storage/rpc/http_service.rs @@ -25,7 +25,7 @@ use rustfs_ecstore::set_disk::DEFAULT_READ_BUFFER_SIZE; use rustfs_ecstore::store::find_local_disk_by_ref; use rustfs_io_metrics::internode_metrics::{ INTERNODE_OPERATION_PUT_FILE_STREAM, INTERNODE_OPERATION_READ_FILE_STREAM, INTERNODE_OPERATION_WALK_DIR, - global_internode_metrics, + INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, global_internode_metrics, }; use rustfs_utils::net::bytes_stream; use s3s::Body; @@ -143,7 +143,9 @@ fn internode_http_operation(path: &str) -> Option<&'static str> { fn record_internode_rpc_error(operation: Option<&'static str>) { match operation { - Some(operation) => global_internode_metrics().record_error_for_operation(operation), + Some(operation) => { + global_internode_metrics().record_error_for_operation_and_backend(operation, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP) + } None => global_internode_metrics().record_error(), } } @@ -183,7 +185,10 @@ 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_for_operation(INTERNODE_OPERATION_READ_FILE_STREAM); + global_internode_metrics().record_incoming_request_for_operation_and_backend( + INTERNODE_OPERATION_READ_FILE_STREAM, + INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, + ); let stream = read_file_body_stream(file, query.length, INTERNODE_OPERATION_READ_FILE_STREAM); Response::builder() @@ -202,7 +207,7 @@ where { let metrics = global_internode_metrics().clone(); let stream = ReaderStream::with_capacity(reader, DEFAULT_READ_BUFFER_SIZE).map_ok(move |bytes| { - metrics.record_sent_bytes_for_operation(operation, bytes.len()); + metrics.record_sent_bytes_for_operation_and_backend(operation, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, bytes.len()); bytes }); @@ -244,10 +249,15 @@ async fn handle_walk_dir(req: Request) -> Response { } }); - global_internode_metrics().record_incoming_request_for_operation(INTERNODE_OPERATION_WALK_DIR); + global_internode_metrics() + .record_incoming_request_for_operation_and_backend(INTERNODE_OPERATION_WALK_DIR, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP); let metrics = global_internode_metrics().clone(); let stream = ReaderStream::with_capacity(rd, DEFAULT_READ_BUFFER_SIZE).map_ok(move |bytes| { - metrics.record_sent_bytes_for_operation(INTERNODE_OPERATION_WALK_DIR, bytes.len()); + metrics.record_sent_bytes_for_operation_and_backend( + INTERNODE_OPERATION_WALK_DIR, + INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, + bytes.len(), + ); bytes }); @@ -284,8 +294,15 @@ 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_for_operation(INTERNODE_OPERATION_PUT_FILE_STREAM); - global_internode_metrics().record_recv_bytes_for_operation(INTERNODE_OPERATION_PUT_FILE_STREAM, copied as usize); + global_internode_metrics().record_incoming_request_for_operation_and_backend( + INTERNODE_OPERATION_PUT_FILE_STREAM, + INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, + ); + global_internode_metrics().record_recv_bytes_for_operation_and_backend( + INTERNODE_OPERATION_PUT_FILE_STREAM, + INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, + copied as usize, + ); if let Err(e) = file.flush().await { return response_with_status(StatusCode::INTERNAL_SERVER_ERROR, format!("write file err {e}")); diff --git a/scripts/run_internode_transport_baseline.sh b/scripts/run_internode_transport_baseline.sh index e9e4022fc..93166268c 100755 --- a/scripts/run_internode_transport_baseline.sh +++ b/scripts/run_internode_transport_baseline.sh @@ -127,7 +127,7 @@ setup_output() { mkdir -p "${OUT_DIR}" echo "scenario,endpoint,workload,concurrency,size,status,throughput,requests_per_sec,avg_latency,error_count,log_file,run_dir" > "${OUT_DIR}/summary.csv" if [[ -n "${INTERNODE_METRICS_URL}" ]]; then - echo "scenario,workload,concurrency,size,metric,operation,before,after,delta" > "${OUT_DIR}/internode_metric_deltas.csv" + echo "scenario,workload,concurrency,size,metric,operation,backend,before,after,delta" > "${OUT_DIR}/internode_metric_deltas.csv" fi } @@ -190,15 +190,20 @@ extract_internode_rows() { awk ' $1 ~ /^rustfs_system_network_internode_operation_/ { metric = $1 + sub(/\{.*/, "", metric) op = "all" + backend = "unknown" if (match($0, /operation="[^"]+"/)) { op = substr($0, RSTART + 11, RLENGTH - 12) } + if (match($0, /backend="[^"]+"/)) { + backend = substr($0, RSTART + 9, RLENGTH - 10) + } n = split($0, parts, " ") value = parts[n] gsub(/[[:space:]]+/, "", value) if (value ~ /^[0-9]+([.][0-9]+)?$/) { - print metric "," op "," value + print metric "," op "," backend "," value } }' "${src}" } @@ -219,18 +224,19 @@ append_metric_deltas() { awk -F',' -v scenario="${scenario}" -v workload="${workload}" -v conc="${conc}" -v size="${size}" ' FNR==NR { - key = $1 SUBSEP $2 - before[key] = $3 + key = $1 SUBSEP $2 SUBSEP $3 + before[key] = $4 next } { - key = $1 SUBSEP $2 + key = $1 SUBSEP $2 SUBSEP $3 metric = $1 operation = $2 - afterv = $3 + 0 + backend = $3 + afterv = $4 + 0 beforev = (key in before ? before[key] + 0 : 0) delta = afterv - beforev - printf "%s,%s,%s,%s,%s,%s,%.0f,%.0f,%.0f\n", scenario, workload, conc, size, metric, operation, beforev, afterv, delta + printf "%s,%s,%s,%s,%s,%s,%s,%.0f,%.0f,%.0f\n", scenario, workload, conc, size, metric, operation, backend, beforev, afterv, delta } ' "${before_rows}" "${after_rows}" >> "${OUT_DIR}/internode_metric_deltas.csv"