diff --git a/crates/ecstore/src/cluster/rpc/remote_disk.rs b/crates/ecstore/src/cluster/rpc/remote_disk.rs index 26e9f899a..ed8772f53 100644 --- a/crates/ecstore/src/cluster/rpc/remote_disk.rs +++ b/crates/ecstore/src/cluster/rpc/remote_disk.rs @@ -41,6 +41,10 @@ use bytes::Bytes; use futures::lock::Mutex; use metrics::counter; use rustfs_filemeta::{FileInfo, ObjectPartInfo, RawFileInfo}; +use rustfs_io_metrics::internode_metrics::{ + INTERNODE_STAGE_READ_VERSION_REQUEST_ENCODE, INTERNODE_STAGE_READ_VERSION_RESPONSE_DECODE, + INTERNODE_STAGE_READ_VERSION_RPC_ROUNDTRIP, +}; use rustfs_protos::ChannelClass; use rustfs_protos::evict_failed_connection; use rustfs_protos::proto_gen::node_service::RenamePartRequest; @@ -64,7 +68,7 @@ use std::{ atomic::{AtomicBool, AtomicU32, Ordering}, }, task::{Context, Poll}, - time::Duration, + time::{Duration, Instant}, }; use tokio::time; use tokio::{ @@ -1790,6 +1794,16 @@ fn decode_msgpack_or_json(binary: &[u8], json: &str, value_ } } +fn read_version_stage_timer(attribution_enabled: bool) -> Option { + attribution_enabled.then(Instant::now) +} + +fn record_read_version_stage(stage: &'static str, started_at: Option) { + if let Some(started_at) = started_at { + crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_read_version_stage(stage, started_at.elapsed()); + } +} + /// Aggregate encoded size (bytes) of a `ReadMultiple` response, preferring the msgpack payloads /// and falling back to the JSON compatibility strings. Used to size the RPC for the payload /// histogram / large-payload alerting (grpc-optimization P0 instrumentation). @@ -2705,8 +2719,11 @@ impl DiskAPI for RemoteDisk { state = "started", "Remote disk RPC started" ); - let opts_str = compat_json(opts)?; - let opts_bin = encode_msgpack(opts)?; + let read_version_attribution_enabled = rustfs_io_metrics::get_stage_metrics_enabled(); + let encode_started = read_version_stage_timer(read_version_attribution_enabled); + let encoded_opts = compat_json(opts).and_then(|opts_str| encode_msgpack(opts).map(|opts_bin| (opts_str, opts_bin))); + record_read_version_stage(INTERNODE_STAGE_READ_VERSION_REQUEST_ENCODE, encode_started); + let (opts_str, opts_bin) = encoded_opts?; // Idempotent version read: eligible for the bounded transient-network retry so a single // reset-by-peer during the read-after-write window does not erode the metadata read @@ -2722,6 +2739,14 @@ impl DiskAPI for RemoteDisk { .get_client() .await .map_err(|err| Error::other(format!("can not get client, err: {err}")))?; + let request_payload_bytes = read_version_attribution_enabled.then(|| { + disk.len() + .saturating_add(volume.len()) + .saturating_add(path.len()) + .saturating_add(version_id.len()) + .saturating_add(opts_str.len()) + .saturating_add(opts_bin.len()) + }); let request = Request::new(ReadVersionRequest { disk, volume: volume.to_string(), @@ -2731,14 +2756,47 @@ impl DiskAPI for RemoteDisk { opts_bin: opts_bin.into(), }); - let response = client.read_version(request).await?.into_inner(); + crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_read_version_request(); + if let Some(request_payload_bytes) = request_payload_bytes { + crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_read_version_sent_bytes(request_payload_bytes); + } + let rpc_started = read_version_stage_timer(read_version_attribution_enabled); + let response = match client.read_version(request).await { + Ok(response) => { + record_read_version_stage(INTERNODE_STAGE_READ_VERSION_RPC_ROUNDTRIP, rpc_started); + response.into_inner() + } + Err(err) => { + record_read_version_stage(INTERNODE_STAGE_READ_VERSION_RPC_ROUNDTRIP, rpc_started); + crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_read_version_error(); + return Err(err.into()); + } + }; if !response.success { + crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_read_version_error(); return Err(response.error.unwrap_or_default().into()); } - let file_info = decode_msgpack_or_json::(&response.file_info_bin, &response.file_info, "FileInfo")?; - validate_decoded_file_info(&file_info)?; + crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_read_version_recv_bytes( + response.file_info.len().saturating_add(response.file_info_bin.len()), + ); + let decode_started = read_version_stage_timer(read_version_attribution_enabled); + let file_info = match decode_msgpack_or_json::(&response.file_info_bin, &response.file_info, "FileInfo") + .and_then(|file_info| { + validate_decoded_file_info(&file_info)?; + Ok(file_info) + }) { + Ok(file_info) => { + record_read_version_stage(INTERNODE_STAGE_READ_VERSION_RESPONSE_DECODE, decode_started); + file_info + } + Err(err) => { + record_read_version_stage(INTERNODE_STAGE_READ_VERSION_RESPONSE_DECODE, decode_started); + crate::cluster::rpc::runtime_sources::record_remote_disk_grpc_read_version_error(); + return Err(err); + } + }; Ok(file_info) }, @@ -7931,12 +7989,17 @@ mod tests { } #[tokio::test] + #[serial] async fn read_version_uses_the_metadata_timeout_on_a_stalled_peer() { runtime_sources::ensure_test_rpc_secret(); let Some((base_addr, accept_task)) = spawn_stalled_grpc_peer().await else { return; }; let remote_disk = remote_disk_for_addr(&base_addr).await; + let metrics = rustfs_io_metrics::internode_metrics::global_internode_metrics(); + let previous_stage_metrics = rustfs_io_metrics::get_stage_metrics_enabled(); + metrics.reset_for_test(); + rustfs_io_metrics::set_get_stage_metrics_enabled(true); temp_env::async_with_vars( [ @@ -7960,6 +8023,18 @@ mod tests { ) .await; + rustfs_io_metrics::set_get_stage_metrics_enabled(previous_stage_metrics); + let snapshot = metrics.snapshot(); + assert!( + snapshot.outgoing_requests_total >= 1, + "ReadVersion call site should record outgoing attempts when attribution is enabled" + ); + assert!( + snapshot.sent_bytes_total > 0, + "ReadVersion call site should record request payload bytes when attribution is enabled" + ); + metrics.reset_for_test(); + remote_disk.cancel_token.cancel(); accept_task.abort(); } diff --git a/crates/ecstore/src/cluster/rpc/runtime_sources.rs b/crates/ecstore/src/cluster/rpc/runtime_sources.rs index 03d19f60f..0c8393a2e 100644 --- a/crates/ecstore/src/cluster/rpc/runtime_sources.rs +++ b/crates/ecstore/src/cluster/rpc/runtime_sources.rs @@ -14,10 +14,11 @@ use rustfs_io_metrics::internode_metrics::{ INTERNODE_MSGPACK_CODEC_JSON, INTERNODE_MSGPACK_CODEC_MSGPACK, INTERNODE_MSGPACK_DIRECTION_RESPONSE, - INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_READ_MULTIPLE, INTERNODE_OPERATION_GRPC_WRITE_ALL, - INTERNODE_OPERATION_PUT_FILE_STREAM, INTERNODE_OPERATION_READ_FILE_STREAM, INTERNODE_TRANSPORT_BACKEND_GRPC, - INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, global_internode_metrics, + INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_READ_MULTIPLE, INTERNODE_OPERATION_GRPC_READ_VERSION, + INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_OPERATION_PUT_FILE_STREAM, INTERNODE_OPERATION_READ_FILE_STREAM, + INTERNODE_TRANSPORT_BACKEND_GRPC, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, global_internode_metrics, }; +use std::time::Duration; #[cfg(test)] use rustfs_io_metrics::internode_metrics::InternodeMetricsSnapshot; @@ -82,6 +83,59 @@ pub(crate) fn record_remote_disk_grpc_read_all_request() { .record_outgoing_request_for_operation_and_backend(INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC); } +pub(crate) fn record_remote_disk_grpc_read_version_request() { + if !rustfs_io_metrics::get_stage_metrics_enabled() { + return; + } + global_internode_metrics().record_outgoing_request_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); +} + +pub(crate) fn record_remote_disk_grpc_read_version_error() { + if !rustfs_io_metrics::get_stage_metrics_enabled() { + return; + } + global_internode_metrics() + .record_error_for_operation_and_backend(INTERNODE_OPERATION_GRPC_READ_VERSION, INTERNODE_TRANSPORT_BACKEND_GRPC); +} + +pub(crate) fn record_remote_disk_grpc_read_version_sent_bytes(bytes: usize) { + if !rustfs_io_metrics::get_stage_metrics_enabled() { + return; + } + global_internode_metrics().record_sent_bytes_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + bytes, + ); +} + +pub(crate) fn record_remote_disk_grpc_read_version_recv_bytes(bytes: usize) { + if !rustfs_io_metrics::get_stage_metrics_enabled() { + return; + } + global_internode_metrics().record_recv_bytes_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + bytes, + ); + record_grpc_payload_size(INTERNODE_OPERATION_GRPC_READ_VERSION, bytes); +} + +pub(crate) fn record_remote_disk_grpc_read_version_stage(stage: &'static str, duration: Duration) { + if !rustfs_io_metrics::get_stage_metrics_enabled() { + return; + } + global_internode_metrics().record_stage_duration_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + stage, + duration, + ); +} + pub(crate) fn record_remote_disk_grpc_read_all_recv_bytes(bytes: usize) { global_internode_metrics().record_recv_bytes_for_operation_and_backend( INTERNODE_OPERATION_GRPC_READ_ALL, diff --git a/crates/io-metrics/src/internode_metrics.rs b/crates/io-metrics/src/internode_metrics.rs index a12a8edfe..fb12a9911 100644 --- a/crates/io-metrics/src/internode_metrics.rs +++ b/crates/io-metrics/src/internode_metrics.rs @@ -47,6 +47,13 @@ pub const INTERNODE_MSGPACK_DIRECTION_REQUEST: &str = "request"; pub const INTERNODE_MSGPACK_DIRECTION_RESPONSE: &str = "response"; pub const INTERNODE_MSGPACK_CODEC_MSGPACK: &str = "msgpack"; pub const INTERNODE_MSGPACK_CODEC_JSON: &str = "json"; +pub const INTERNODE_STAGE_READ_VERSION_REQUEST_ENCODE: &str = "read_version_request_encode"; +pub const INTERNODE_STAGE_READ_VERSION_REQUEST_DECODE: &str = "read_version_request_decode"; +pub const INTERNODE_STAGE_READ_VERSION_DISK_READ: &str = "read_version_disk_read"; +pub const INTERNODE_STAGE_READ_VERSION_RESPONSE_JSON_ENCODE: &str = "read_version_response_json_encode"; +pub const INTERNODE_STAGE_READ_VERSION_RESPONSE_MSGPACK_ENCODE: &str = "read_version_response_msgpack_encode"; +pub const INTERNODE_STAGE_READ_VERSION_RPC_ROUNDTRIP: &str = "read_version_rpc_roundtrip"; +pub const INTERNODE_STAGE_READ_VERSION_RESPONSE_DECODE: &str = "read_version_response_decode"; const OPERATION_LABEL: &str = "operation"; const BACKEND_LABEL: &str = "backend"; @@ -67,6 +74,7 @@ const INTERNODE_OPERATION_REQUESTS_OUTGOING_TOTAL: &str = "rustfs_system_network 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"; const INTERNODE_OPERATION_DURATION_MS: &str = "rustfs_system_network_internode_operation_duration_ms"; +const INTERNODE_OPERATION_STAGE_DURATION_MS: &str = "rustfs_system_network_internode_operation_stage_duration_ms"; const INTERNODE_OPERATION_CLASSIFIED_ERRORS_TOTAL: &str = "rustfs_system_network_internode_operation_classified_errors_total"; const INTERNODE_OPERATION_RETRIES_TOTAL: &str = "rustfs_system_network_internode_operation_retries_total"; const INTERNODE_OPERATION_RETRY_SUCCESSES_TOTAL: &str = "rustfs_system_network_internode_operation_retry_successes_total"; @@ -105,6 +113,7 @@ const SERVER_OPERATION_BACKEND_HTTP_VERSION_LABELS: &[&str] = &[SERVER_LABEL, OP const SERVER_OPERATION_BACKEND_FAILURE_REASON_LABELS: &[&str] = &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL, FAILURE_REASON_LABEL]; const SERVER_OPERATION_BACKEND_RPC_PATH_LABELS: &[&str] = &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL, RPC_PATH_LABEL]; +const SERVER_OPERATION_BACKEND_STAGE_LABELS: &[&str] = &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL, STAGE_LABEL]; const SERVER_LABELS: &[&str] = &[SERVER_LABEL]; const SERVER_REASON_LABELS: &[&str] = &[SERVER_LABEL, REASON_LABEL]; const SERVER_QUORUM_FAILURE_LABELS: &[&str] = &[SERVER_LABEL, STAGE_LABEL, DOMINANT_ERROR_LABEL]; @@ -134,6 +143,10 @@ pub const INTERNODE_OPERATION_METRICS: &[InternodeOperationMetricDescriptor] = & name: INTERNODE_OPERATION_DURATION_MS, labels: SERVER_OPERATION_BACKEND_LABELS, }, + InternodeOperationMetricDescriptor { + name: INTERNODE_OPERATION_STAGE_DURATION_MS, + labels: SERVER_OPERATION_BACKEND_STAGE_LABELS, + }, InternodeOperationMetricDescriptor { name: INTERNODE_OPERATION_CLASSIFIED_ERRORS_TOTAL, labels: SERVER_OPERATION_BACKEND_CLASSIFICATION_LABELS, @@ -394,6 +407,24 @@ impl InternodeMetrics { .record(duration_ms); } + pub fn record_stage_duration_for_operation_and_backend( + &self, + operation: &'static str, + backend: &'static str, + stage: &'static str, + duration: Duration, + ) { + let duration_ms = duration.as_secs_f64() * 1000.0; + metrics::histogram!( + INTERNODE_OPERATION_STAGE_DURATION_MS, + SERVER_LABEL => current_server_label(), + OPERATION_LABEL => operation, + BACKEND_LABEL => backend, + STAGE_LABEL => stage + ) + .record(duration_ms); + } + pub fn record_classified_error_for_operation_and_backend( &self, operation: &'static str, @@ -988,42 +1019,90 @@ mod tests { assert_eq!(snapshot.replay_cache_evictions_total, 3); } + #[test] + fn operation_stage_duration_records_low_cardinality_stage_labels() { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let metrics = InternodeMetrics::default(); + + with_local_recorder(&recorder, || { + metrics.record_stage_duration_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + INTERNODE_STAGE_READ_VERSION_RPC_ROUNDTRIP, + Duration::from_micros(125), + ); + }); + + let entries: Vec<_> = snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter(|(composite, _, _, _)| composite.key().name() == INTERNODE_OPERATION_STAGE_DURATION_MS) + .collect(); + assert_eq!(entries.len(), 1); + let labels: HashMap<_, _> = entries[0] + .0 + .key() + .labels() + .map(|label| (label.key().to_string(), label.value().to_string())) + .collect(); + assert_eq!( + labels.get(OPERATION_LABEL).map(String::as_str), + Some(INTERNODE_OPERATION_GRPC_READ_VERSION) + ); + assert_eq!(labels.get(BACKEND_LABEL).map(String::as_str), Some(INTERNODE_TRANSPORT_BACKEND_GRPC)); + assert_eq!( + labels.get(STAGE_LABEL).map(String::as_str), + Some(INTERNODE_STAGE_READ_VERSION_RPC_ROUNDTRIP) + ); + assert!(labels.get(SERVER_LABEL).is_some_and(|value| !value.is_empty())); + match &entries[0].3 { + DebugValue::Histogram(samples) => assert_eq!(samples.iter().map(|sample| sample.0).collect::>(), vec![0.125]), + other => panic!("{INTERNODE_OPERATION_STAGE_DURATION_MS} must be a histogram, got {other:?}"), + } + } + #[test] fn operation_metric_descriptors_include_backend_and_operation_labels() { - assert_eq!(INTERNODE_OPERATION_METRICS.len(), 21); + assert_eq!(INTERNODE_OPERATION_METRICS.len(), 22); for metric in &INTERNODE_OPERATION_METRICS[..6] { assert_eq!(metric.labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL]); } - for metric in &INTERNODE_OPERATION_METRICS[6..9] { + assert_eq!( + INTERNODE_OPERATION_METRICS[6].labels, + &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL, STAGE_LABEL] + ); + for metric in &INTERNODE_OPERATION_METRICS[7..10] { assert_eq!(metric.labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL, CLASSIFICATION_LABEL]); } assert_eq!( - INTERNODE_OPERATION_METRICS[9].labels, + INTERNODE_OPERATION_METRICS[10].labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL, HTTP_VERSION_LABEL] ); - for metric in &INTERNODE_OPERATION_METRICS[10..12] { + for metric in &INTERNODE_OPERATION_METRICS[11..13] { assert_eq!(metric.labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL]); } - assert_eq!( - INTERNODE_OPERATION_METRICS[12].labels, - &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL, FAILURE_REASON_LABEL] - ); assert_eq!( INTERNODE_OPERATION_METRICS[13].labels, - &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL, RPC_PATH_LABEL] + &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL, FAILURE_REASON_LABEL] ); assert_eq!( INTERNODE_OPERATION_METRICS[14].labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL, RPC_PATH_LABEL] ); - for metric in &INTERNODE_OPERATION_METRICS[15..17] { + assert_eq!( + INTERNODE_OPERATION_METRICS[15].labels, + &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL, RPC_PATH_LABEL] + ); + for metric in &INTERNODE_OPERATION_METRICS[16..18] { assert_eq!(metric.labels, &[SERVER_LABEL]); } - assert_eq!(INTERNODE_OPERATION_METRICS[17].labels, &[SERVER_LABEL, REASON_LABEL]); - assert_eq!(INTERNODE_OPERATION_METRICS[18].labels, &[SERVER_LABEL, STAGE_LABEL, DOMINANT_ERROR_LABEL]); + assert_eq!(INTERNODE_OPERATION_METRICS[18].labels, &[SERVER_LABEL, REASON_LABEL]); + assert_eq!(INTERNODE_OPERATION_METRICS[19].labels, &[SERVER_LABEL, STAGE_LABEL, DOMINANT_ERROR_LABEL]); // Payload histogram + large-payload counter carry operation+backend labels. - assert_eq!(INTERNODE_OPERATION_METRICS[19].labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL]); assert_eq!(INTERNODE_OPERATION_METRICS[20].labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL]); + assert_eq!(INTERNODE_OPERATION_METRICS[21].labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL]); } #[test] @@ -1054,62 +1133,66 @@ mod tests { ); assert_eq!( INTERNODE_OPERATION_METRICS[6].name, - "rustfs_system_network_internode_operation_classified_errors_total" + "rustfs_system_network_internode_operation_stage_duration_ms" ); assert_eq!( INTERNODE_OPERATION_METRICS[7].name, - "rustfs_system_network_internode_operation_retries_total" + "rustfs_system_network_internode_operation_classified_errors_total" ); assert_eq!( INTERNODE_OPERATION_METRICS[8].name, - "rustfs_system_network_internode_operation_retry_successes_total" + "rustfs_system_network_internode_operation_retries_total" ); assert_eq!( INTERNODE_OPERATION_METRICS[9].name, - "rustfs_system_network_internode_operation_http_versions_total" + "rustfs_system_network_internode_operation_retry_successes_total" ); assert_eq!( INTERNODE_OPERATION_METRICS[10].name, - "rustfs_system_network_internode_operation_stall_timeouts_total" + "rustfs_system_network_internode_operation_http_versions_total" ); assert_eq!( INTERNODE_OPERATION_METRICS[11].name, - "rustfs_system_network_internode_operation_write_shutdown_errors_total" + "rustfs_system_network_internode_operation_stall_timeouts_total" ); assert_eq!( INTERNODE_OPERATION_METRICS[12].name, - "rustfs_system_network_internode_rpc_auth_failures_total" + "rustfs_system_network_internode_operation_write_shutdown_errors_total" ); assert_eq!( INTERNODE_OPERATION_METRICS[13].name, - "rustfs_system_network_internode_replay_cache_overflow_by_operation_total" + "rustfs_system_network_internode_rpc_auth_failures_total" ); assert_eq!( INTERNODE_OPERATION_METRICS[14].name, - "rustfs_system_network_internode_replay_cache_records_total" + "rustfs_system_network_internode_replay_cache_overflow_by_operation_total" ); assert_eq!( INTERNODE_OPERATION_METRICS[15].name, - "rustfs_system_network_internode_replay_cache_entries" + "rustfs_system_network_internode_replay_cache_records_total" ); assert_eq!( INTERNODE_OPERATION_METRICS[16].name, - "rustfs_system_network_internode_replay_cache_capacity" + "rustfs_system_network_internode_replay_cache_entries" ); assert_eq!( INTERNODE_OPERATION_METRICS[17].name, - "rustfs_system_network_internode_replay_cache_evictions_total" + "rustfs_system_network_internode_replay_cache_capacity" ); assert_eq!( INTERNODE_OPERATION_METRICS[18].name, - "rustfs_system_storage_erasure_write_quorum_failures_total" + "rustfs_system_network_internode_replay_cache_evictions_total" ); assert_eq!( INTERNODE_OPERATION_METRICS[19].name, - "rustfs_system_network_internode_operation_payload_bytes" + "rustfs_system_storage_erasure_write_quorum_failures_total" ); assert_eq!( INTERNODE_OPERATION_METRICS[20].name, + "rustfs_system_network_internode_operation_payload_bytes" + ); + assert_eq!( + INTERNODE_OPERATION_METRICS[21].name, "rustfs_system_network_internode_operation_large_payloads_total" ); assert_eq!(INTERNODE_OPERATION_GRPC_READ_MULTIPLE, "grpc_read_multiple"); @@ -1129,6 +1212,16 @@ mod tests { assert_eq!(INTERNODE_MSGPACK_DIRECTION_RESPONSE, "response"); assert_eq!(INTERNODE_MSGPACK_CODEC_MSGPACK, "msgpack"); assert_eq!(INTERNODE_MSGPACK_CODEC_JSON, "json"); + assert_eq!(INTERNODE_STAGE_READ_VERSION_REQUEST_ENCODE, "read_version_request_encode"); + assert_eq!(INTERNODE_STAGE_READ_VERSION_REQUEST_DECODE, "read_version_request_decode"); + assert_eq!(INTERNODE_STAGE_READ_VERSION_DISK_READ, "read_version_disk_read"); + assert_eq!(INTERNODE_STAGE_READ_VERSION_RESPONSE_JSON_ENCODE, "read_version_response_json_encode"); + assert_eq!( + INTERNODE_STAGE_READ_VERSION_RESPONSE_MSGPACK_ENCODE, + "read_version_response_msgpack_encode" + ); + assert_eq!(INTERNODE_STAGE_READ_VERSION_RPC_ROUNDTRIP, "read_version_rpc_roundtrip"); + assert_eq!(INTERNODE_STAGE_READ_VERSION_RESPONSE_DECODE, "read_version_response_decode"); assert_eq!( INTERNODE_SIGNATURE_V1_FALLBACK_TOTAL, "rustfs_system_network_internode_signature_v1_fallback_total" diff --git a/rustfs/src/storage/rpc/node_service/disk.rs b/rustfs/src/storage/rpc/node_service/disk.rs index ce0ba42e5..5d9fbc969 100644 --- a/rustfs/src/storage/rpc/node_service/disk.rs +++ b/rustfs/src/storage/rpc/node_service/disk.rs @@ -23,12 +23,15 @@ use bytes::Bytes; use rustfs_filemeta::FileInfo; use rustfs_io_metrics::internode_metrics::{ INTERNODE_MSGPACK_CODEC_JSON, INTERNODE_MSGPACK_CODEC_MSGPACK, INTERNODE_MSGPACK_DIRECTION_REQUEST, - INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC, - global_internode_metrics, + INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_READ_VERSION, INTERNODE_OPERATION_GRPC_WRITE_ALL, + INTERNODE_STAGE_READ_VERSION_DISK_READ, INTERNODE_STAGE_READ_VERSION_REQUEST_DECODE, + INTERNODE_STAGE_READ_VERSION_RESPONSE_JSON_ENCODE, INTERNODE_STAGE_READ_VERSION_RESPONSE_MSGPACK_ENCODE, + INTERNODE_TRANSPORT_BACKEND_GRPC, global_internode_metrics, }; use rustfs_protos::proto_gen::node_service::*; use serde::de::DeserializeOwned; use std::io::Cursor; +use std::time::Instant; use tonic::{Request, Response, Status}; use tracing::debug; @@ -201,6 +204,21 @@ fn encode_read_multiple_response_payloads( Ok((read_multiple_resps_json, read_multiple_resps_bin)) } +fn internode_stage_timer(attribution_enabled: bool) -> Option { + attribution_enabled.then(Instant::now) +} + +fn record_read_version_stage(stage: &'static str, started_at: Option) { + if let Some(started_at) = started_at { + global_internode_metrics().record_stage_duration_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + stage, + started_at.elapsed(), + ); + } +} + fn encode_batch_read_version_response_payloads( batch_read_version_resps: &[BatchReadVersionResp], request_decoded_from_msgpack: bool, @@ -685,11 +703,42 @@ impl NodeService { request: Request, ) -> Result, Status> { let request = request.into_inner(); + let metrics = global_internode_metrics(); + let read_version_attribution_enabled = rustfs_io_metrics::get_stage_metrics_enabled(); + if read_version_attribution_enabled { + metrics.record_incoming_request_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); + metrics.record_recv_bytes_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + request + .disk + .len() + .saturating_add(request.volume.len()) + .saturating_add(request.path.len()) + .saturating_add(request.version_id.len()) + .saturating_add(request.opts.len()) + .saturating_add(request.opts_bin.len()), + ); + } if let Some(disk) = self.find_disk(&request.disk).await { let request_had_msgpack_payload = !request.opts_bin.is_empty(); + let decode_started = internode_stage_timer(read_version_attribution_enabled); let opts = match decode_msgpack_or_json::(&request.opts_bin, &request.opts, "ReadOptions") { - Ok(options) => options, + Ok(options) => { + record_read_version_stage(INTERNODE_STAGE_READ_VERSION_REQUEST_DECODE, decode_started); + options + } Err(err) => { + record_read_version_stage(INTERNODE_STAGE_READ_VERSION_REQUEST_DECODE, decode_started); + if read_version_attribution_enabled { + metrics.record_error_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); + } return Ok(Response::new(ReadVersionResponse { success: false, file_info: String::new(), @@ -698,42 +747,88 @@ impl NodeService { })); } }; + let disk_read_started = internode_stage_timer(read_version_attribution_enabled); match disk .read_version("", &request.volume, &request.path, &request.version_id, &opts) .await { Ok(file_info) => { + record_read_version_stage(INTERNODE_STAGE_READ_VERSION_DISK_READ, disk_read_started); + let json_encode_started = internode_stage_timer(read_version_attribution_enabled); let file_info_json = compat_response_json(&file_info, request_had_msgpack_payload); + record_read_version_stage(INTERNODE_STAGE_READ_VERSION_RESPONSE_JSON_ENCODE, json_encode_started); + let msgpack_encode_started = internode_stage_timer(read_version_attribution_enabled); let file_info_bin = encode_file_info_msgpack(&file_info); + record_read_version_stage(INTERNODE_STAGE_READ_VERSION_RESPONSE_MSGPACK_ENCODE, msgpack_encode_started); match (file_info_json, file_info_bin) { - (Ok(file_info), Ok(file_info_bin)) => Ok(Response::new(ReadVersionResponse { - success: true, - file_info, - file_info_bin: file_info_bin.into(), - error: None, - })), - (Err(err), _) => Ok(Response::new(ReadVersionResponse { - success: false, - file_info: String::new(), - file_info_bin: Vec::new().into(), - error: Some(DiskError::other(format!("encode data failed: {err}")).into()), - })), - (_, Err(err)) => Ok(Response::new(ReadVersionResponse { - success: false, - file_info: String::new(), - file_info_bin: Vec::new().into(), - error: Some(DiskError::other(format!("encode data failed: {err}")).into()), - })), + (Ok(file_info), Ok(file_info_bin)) => { + if read_version_attribution_enabled { + metrics.record_sent_bytes_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + file_info.len().saturating_add(file_info_bin.len()), + ); + } + Ok(Response::new(ReadVersionResponse { + success: true, + file_info, + file_info_bin: file_info_bin.into(), + error: None, + })) + } + (Err(err), _) => { + if read_version_attribution_enabled { + metrics.record_error_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); + } + Ok(Response::new(ReadVersionResponse { + success: false, + file_info: String::new(), + file_info_bin: Vec::new().into(), + error: Some(DiskError::other(format!("encode data failed: {err}")).into()), + })) + } + (_, Err(err)) => { + if read_version_attribution_enabled { + metrics.record_error_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); + } + Ok(Response::new(ReadVersionResponse { + success: false, + file_info: String::new(), + file_info_bin: Vec::new().into(), + error: Some(DiskError::other(format!("encode data failed: {err}")).into()), + })) + } } } - Err(err) => Ok(Response::new(ReadVersionResponse { - success: false, - file_info: String::new(), - file_info_bin: Vec::new().into(), - error: Some(err.into()), - })), + Err(err) => { + record_read_version_stage(INTERNODE_STAGE_READ_VERSION_DISK_READ, disk_read_started); + if read_version_attribution_enabled { + metrics.record_error_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); + } + Ok(Response::new(ReadVersionResponse { + success: false, + file_info: String::new(), + file_info_bin: Vec::new().into(), + error: Some(err.into()), + })) + } } } else { + if read_version_attribution_enabled { + metrics.record_error_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + ); + } Ok(Response::new(ReadVersionResponse { success: false, file_info: String::new(), @@ -1520,12 +1615,16 @@ mod tests { encode_batch_read_version_response_payloads, encode_file_info_msgpack, encode_msgpack, encode_msgpack_named, encode_read_multiple_response_payloads, encode_rename_data_response_payloads, }; + use crate::storage::rpc::node_service::make_server; use crate::storage::storage_api::ReadMultipleResp; use crate::storage::storage_api::RenameDataResp; use crate::storage::storage_api::rpc_consumer::node_service::BatchReadVersionResp; use rustfs_filemeta::FileInfo; use rustfs_io_metrics::internode_metrics::global_internode_metrics; + use rustfs_protos::proto_gen::node_service::ReadVersionRequest; use serde::{Deserialize, Serialize}; + use serial_test::serial; + use tonic::Request; #[derive(Debug, PartialEq, Eq, Serialize, Deserialize)] struct SamplePayload { @@ -1533,6 +1632,36 @@ mod tests { count: u32, } + #[tokio::test] + #[serial] + async fn handle_read_version_records_attribution_for_missing_disk() { + let metrics = global_internode_metrics(); + let previous_stage_metrics = rustfs_io_metrics::get_stage_metrics_enabled(); + metrics.reset_for_test(); + rustfs_io_metrics::set_get_stage_metrics_enabled(true); + + let response = make_server() + .handle_read_version(Request::new(ReadVersionRequest { + disk: "missing-disk".to_string(), + volume: "bucket".to_string(), + path: "object".to_string(), + version_id: String::new(), + opts: String::new(), + opts_bin: Vec::new().into(), + })) + .await + .expect("ReadVersion handler should return a response") + .into_inner(); + + rustfs_io_metrics::set_get_stage_metrics_enabled(previous_stage_metrics); + let snapshot = metrics.snapshot(); + assert!(!response.success); + assert_eq!(snapshot.incoming_requests_total, 1); + assert_eq!(snapshot.errors_total, 1); + assert!(snapshot.recv_bytes_total > 0); + metrics.reset_for_test(); + } + #[test] fn decode_msgpack_or_json_prefers_binary_payload() { let payload = SamplePayload {