From d900e11a095201d722156c9e1a5640da52a40d4c Mon Sep 17 00:00:00 2001 From: houseme Date: Mon, 10 Aug 2026 15:13:27 +0800 Subject: [PATCH] perf(ecstore): expose replay cache RPC sources (#5926) Track accepted replay cache records by gRPC operation and split Lock/Unlock and ReadVersion methods out of grpc_other so hotpath validation can attribute nonce pressure without changing replay protection semantics. Co-authored-by: heihutu --- crates/ecstore/src/cluster/rpc/http_auth.rs | 106 +++++++++++++++++++- crates/io-metrics/src/internode_metrics.rs | 96 ++++++++++++++++-- 2 files changed, 190 insertions(+), 12 deletions(-) diff --git a/crates/ecstore/src/cluster/rpc/http_auth.rs b/crates/ecstore/src/cluster/rpc/http_auth.rs index 8fa813c2d..48c3ce696 100644 --- a/crates/ecstore/src/cluster/rpc/http_auth.rs +++ b/crates/ecstore/src/cluster/rpc/http_auth.rs @@ -40,8 +40,11 @@ use http::{HeaderMap, HeaderValue, Method, Uri}; use rustfs_credentials::{DEFAULT_SECRET_KEY, RPC_SECRET_REQUIRED_MESSAGE}; use rustfs_credentials::{RPC_SECRET_REQUIRED_OPERATOR_MESSAGE, try_get_rpc_token}; use rustfs_io_metrics::internode_metrics::{ - INTERNODE_OPERATION_GRPC_OTHER, INTERNODE_OPERATION_GRPC_READ_ALL, INTERNODE_OPERATION_GRPC_READ_MULTIPLE, - INTERNODE_OPERATION_GRPC_WRITE_ALL, INTERNODE_TRANSPORT_BACKEND_GRPC, global_internode_metrics, + INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION, INTERNODE_OPERATION_GRPC_FORCE_UNLOCK, INTERNODE_OPERATION_GRPC_LOCK, + INTERNODE_OPERATION_GRPC_LOCK_BATCH, INTERNODE_OPERATION_GRPC_OTHER, INTERNODE_OPERATION_GRPC_READ_ALL, + INTERNODE_OPERATION_GRPC_READ_MULTIPLE, INTERNODE_OPERATION_GRPC_READ_VERSION, INTERNODE_OPERATION_GRPC_REFRESH, + INTERNODE_OPERATION_GRPC_UNLOCK, INTERNODE_OPERATION_GRPC_UNLOCK_BATCH, INTERNODE_OPERATION_GRPC_WRITE_ALL, + INTERNODE_TRANSPORT_BACKEND_GRPC, global_internode_metrics, }; use rustfs_object_data_cache::{MemoryBasis, resolve_effective_memory}; use rustfs_utils::get_env_bool; @@ -340,6 +343,7 @@ struct RpcNonceCacheMetrics<'a> { expired: usize, entries: usize, capacity: usize, + record_scope: Option>, overflow_scope: Option>, } @@ -350,6 +354,13 @@ fn publish_nonce_cache_metrics(metrics: Option>) { let internode_metrics = global_internode_metrics(); internode_metrics.record_replay_cache_evictions("expired", metrics.expired); internode_metrics.record_replay_cache_state(metrics.entries, metrics.capacity); + if let Some(scope) = metrics.record_scope { + internode_metrics.record_replay_cache_record_for_operation_and_backend_path( + scope.operation, + scope.backend, + scope.rpc_path, + ); + } if let Some(scope) = metrics.overflow_scope { internode_metrics.record_replay_cache_overflow_for_operation_and_backend_path( scope.operation, @@ -385,6 +396,7 @@ impl RpcNonceCache { expired, entries: self.nonces.len(), capacity: record.capacity, + record_scope: None, overflow_scope: None, }; if self.nonces.contains(&record.nonce) { @@ -409,6 +421,7 @@ impl RpcNonceCache { Ok(()), Some(RpcNonceCacheMetrics { entries: self.nonces.len(), + record_scope: Some(record.metric_scope), ..metrics }), ) @@ -913,7 +926,15 @@ fn tonic_rpc_metric_operation(path: &str) -> &'static str { match parse_tonic_rpc_path(path).ok().map(|(_, rpc_method)| rpc_method) { Some("ReadAll") => INTERNODE_OPERATION_GRPC_READ_ALL, Some("ReadMultiple") => INTERNODE_OPERATION_GRPC_READ_MULTIPLE, + Some("ReadVersion") => INTERNODE_OPERATION_GRPC_READ_VERSION, + Some("BatchReadVersion") => INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION, Some("WriteAll") => INTERNODE_OPERATION_GRPC_WRITE_ALL, + Some("Lock") => INTERNODE_OPERATION_GRPC_LOCK, + Some("UnLock") => INTERNODE_OPERATION_GRPC_UNLOCK, + Some("LockBatch") => INTERNODE_OPERATION_GRPC_LOCK_BATCH, + Some("UnLockBatch") => INTERNODE_OPERATION_GRPC_UNLOCK_BATCH, + Some("Refresh") => INTERNODE_OPERATION_GRPC_REFRESH, + Some("ForceUnLock") => INTERNODE_OPERATION_GRPC_FORCE_UNLOCK, _ => INTERNODE_OPERATION_GRPC_OTHER, } } @@ -2457,10 +2478,42 @@ mod tests { tonic_rpc_metric_operation("/node_service.NodeService/ReadMultiple"), INTERNODE_OPERATION_GRPC_READ_MULTIPLE ); + assert_eq!( + tonic_rpc_metric_operation("/node_service.NodeService/ReadVersion"), + INTERNODE_OPERATION_GRPC_READ_VERSION + ); + assert_eq!( + tonic_rpc_metric_operation("/node_service.NodeService/BatchReadVersion"), + INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION + ); assert_eq!( tonic_rpc_metric_operation("/node_service.NodeService/WriteAll"), INTERNODE_OPERATION_GRPC_WRITE_ALL ); + assert_eq!( + tonic_rpc_metric_operation("/node_service.NodeService/Lock"), + INTERNODE_OPERATION_GRPC_LOCK + ); + assert_eq!( + tonic_rpc_metric_operation("/node_service.NodeService/UnLock"), + INTERNODE_OPERATION_GRPC_UNLOCK + ); + assert_eq!( + tonic_rpc_metric_operation("/node_service.NodeService/LockBatch"), + INTERNODE_OPERATION_GRPC_LOCK_BATCH + ); + assert_eq!( + tonic_rpc_metric_operation("/node_service.NodeService/UnLockBatch"), + INTERNODE_OPERATION_GRPC_UNLOCK_BATCH + ); + assert_eq!( + tonic_rpc_metric_operation("/node_service.NodeService/Refresh"), + INTERNODE_OPERATION_GRPC_REFRESH + ); + assert_eq!( + tonic_rpc_metric_operation("/node_service.NodeService/ForceUnLock"), + INTERNODE_OPERATION_GRPC_FORCE_UNLOCK + ); assert_eq!( tonic_rpc_metric_operation("/node_service.NodeService/SignalService"), INTERNODE_OPERATION_GRPC_OTHER @@ -2554,6 +2607,13 @@ mod tests { result } + fn check_test_nonce_record_with_metrics<'a>( + cache: &mut RpcNonceCache, + record: RpcNonceRecord<'a>, + ) -> (std::io::Result<()>, Option>) { + cache.check_and_record(record) + } + fn test_nonce_record( nonce: Uuid, signed_at: i64, @@ -2597,6 +2657,48 @@ mod tests { assert!(cache.nonces.contains(&nonce_b)); } + #[test] + fn nonce_cache_metrics_mark_successful_records_only() { + let now = Instant::now(); + let expiry = now.checked_add(REPLAY_CACHE_RETENTION).expect("test expiry should fit"); + let nonce_a = Uuid::new_v4(); + let nonce_b = Uuid::new_v4(); + let mut cache = RpcNonceCache::default(); + + let (recorded, metrics) = + check_test_nonce_record_with_metrics(&mut cache, test_nonce_record(nonce_a, 100, now, 100, expiry, 1)); + recorded.expect("first nonce should be recorded"); + let metrics = metrics.expect("successful nonce should publish metrics"); + let record_scope = metrics.record_scope.expect("successful nonce should carry record scope"); + assert_eq!(record_scope.operation, INTERNODE_OPERATION_GRPC_READ_ALL); + assert_eq!(record_scope.backend, INTERNODE_TRANSPORT_BACKEND_GRPC); + assert_eq!(record_scope.rpc_path, "/node_service.NodeService/ReadAll"); + assert!(metrics.overflow_scope.is_none()); + + let (replay, metrics) = + check_test_nonce_record_with_metrics(&mut cache, test_nonce_record(nonce_a, 100, now, 100, expiry, 1)); + assert_eq!( + replay.expect_err("duplicate nonce must fail closed").to_string(), + "RPC request replay detected" + ); + let metrics = metrics.expect("replay rejection should still publish cache state"); + assert!(metrics.record_scope.is_none()); + assert!(metrics.overflow_scope.is_none()); + + let (overflow, metrics) = + check_test_nonce_record_with_metrics(&mut cache, test_nonce_record(nonce_b, 100, now, 100, expiry, 1)); + assert_eq!( + overflow.expect_err("full cache must fail closed").to_string(), + "RPC replay cache capacity exceeded" + ); + let metrics = metrics.expect("overflow should publish cache state"); + assert!(metrics.record_scope.is_none()); + let overflow_scope = metrics.overflow_scope.expect("overflow should keep diagnostic scope"); + assert_eq!(overflow_scope.operation, INTERNODE_OPERATION_GRPC_READ_ALL); + assert_eq!(overflow_scope.backend, INTERNODE_TRANSPORT_BACKEND_GRPC); + assert_eq!(overflow_scope.rpc_path, "/node_service.NodeService/ReadAll"); + } + // The `rpc_body_digest_fallback_counter` serial group covers every test that drives (or // asserts on) the process-global body-digest fallback counter, so exact-delta assertions // cannot race with each other. diff --git a/crates/io-metrics/src/internode_metrics.rs b/crates/io-metrics/src/internode_metrics.rs index 62ffe091c..41ce7ceac 100644 --- a/crates/io-metrics/src/internode_metrics.rs +++ b/crates/io-metrics/src/internode_metrics.rs @@ -28,6 +28,14 @@ pub const INTERNODE_OPERATION_NS_SCANNER: &str = "ns_scanner"; 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_OPERATION_GRPC_READ_MULTIPLE: &str = "grpc_read_multiple"; +pub const INTERNODE_OPERATION_GRPC_READ_VERSION: &str = "grpc_read_version"; +pub const INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION: &str = "grpc_batch_read_version"; +pub const INTERNODE_OPERATION_GRPC_LOCK: &str = "grpc_lock"; +pub const INTERNODE_OPERATION_GRPC_UNLOCK: &str = "grpc_unlock"; +pub const INTERNODE_OPERATION_GRPC_LOCK_BATCH: &str = "grpc_lock_batch"; +pub const INTERNODE_OPERATION_GRPC_UNLOCK_BATCH: &str = "grpc_unlock_batch"; +pub const INTERNODE_OPERATION_GRPC_REFRESH: &str = "grpc_refresh"; +pub const INTERNODE_OPERATION_GRPC_FORCE_UNLOCK: &str = "grpc_force_unlock"; pub const INTERNODE_OPERATION_GRPC_OTHER: &str = "grpc_other"; pub const INTERNODE_TRANSPORT_BACKEND_TCP_HTTP: &str = "tcp-http"; pub const INTERNODE_TRANSPORT_BACKEND_GRPC: &str = "grpc"; @@ -78,6 +86,7 @@ const INTERNODE_REPLAY_SCOPE_FALLBACK_TOTAL: &str = "rustfs_system_network_inter const INTERNODE_REPLAY_CACHE_OVERFLOW_TOTAL: &str = "rustfs_system_network_internode_replay_cache_overflow_total"; const INTERNODE_REPLAY_CACHE_OVERFLOW_BY_OPERATION_TOTAL: &str = "rustfs_system_network_internode_replay_cache_overflow_by_operation_total"; +const INTERNODE_REPLAY_CACHE_RECORDS_TOTAL: &str = "rustfs_system_network_internode_replay_cache_records_total"; const INTERNODE_REPLAY_CACHE_ENTRIES: &str = "rustfs_system_network_internode_replay_cache_entries"; const INTERNODE_REPLAY_CACHE_CAPACITY: &str = "rustfs_system_network_internode_replay_cache_capacity"; const INTERNODE_REPLAY_CACHE_EVICTIONS_TOTAL: &str = "rustfs_system_network_internode_replay_cache_evictions_total"; @@ -157,6 +166,10 @@ pub const INTERNODE_OPERATION_METRICS: &[InternodeOperationMetricDescriptor] = & name: INTERNODE_REPLAY_CACHE_OVERFLOW_BY_OPERATION_TOTAL, labels: SERVER_OPERATION_BACKEND_RPC_PATH_LABELS, }, + InternodeOperationMetricDescriptor { + name: INTERNODE_REPLAY_CACHE_RECORDS_TOTAL, + labels: SERVER_OPERATION_BACKEND_RPC_PATH_LABELS, + }, InternodeOperationMetricDescriptor { name: INTERNODE_REPLAY_CACHE_ENTRIES, labels: SERVER_LABELS, @@ -619,6 +632,22 @@ impl InternodeMetrics { .increment(1); } + pub fn record_replay_cache_record_for_operation_and_backend_path( + &self, + operation: &'static str, + backend: &'static str, + rpc_path: &str, + ) { + counter!( + INTERNODE_REPLAY_CACHE_RECORDS_TOTAL, + SERVER_LABEL => current_server_label(), + OPERATION_LABEL => operation, + BACKEND_LABEL => backend, + RPC_PATH_LABEL => rpc_path.to_owned() + ) + .increment(1); + } + pub fn record_replay_cache_state(&self, entries: usize, capacity: usize) { let entries = usize_to_u64_saturating(entries); let capacity = usize_to_u64_saturating(capacity); @@ -964,7 +993,7 @@ mod tests { #[test] fn operation_metric_descriptors_include_backend_and_operation_labels() { - assert_eq!(INTERNODE_OPERATION_METRICS.len(), 20); + assert_eq!(INTERNODE_OPERATION_METRICS.len(), 21); for metric in &INTERNODE_OPERATION_METRICS[..6] { assert_eq!(metric.labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL]); } @@ -986,14 +1015,18 @@ mod tests { INTERNODE_OPERATION_METRICS[13].labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL, RPC_PATH_LABEL] ); - for metric in &INTERNODE_OPERATION_METRICS[14..16] { + 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!(metric.labels, &[SERVER_LABEL]); } - assert_eq!(INTERNODE_OPERATION_METRICS[16].labels, &[SERVER_LABEL, REASON_LABEL]); - assert_eq!(INTERNODE_OPERATION_METRICS[17].labels, &[SERVER_LABEL, STAGE_LABEL, DOMINANT_ERROR_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]); // Payload histogram + large-payload counter carry operation+backend labels. - assert_eq!(INTERNODE_OPERATION_METRICS[18].labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL]); 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]); } #[test] @@ -1004,6 +1037,14 @@ mod tests { assert_eq!(INTERNODE_OPERATION_WALK_DIR, "walk_dir"); assert_eq!(INTERNODE_OPERATION_GRPC_READ_ALL, "grpc_read_all"); assert_eq!(INTERNODE_OPERATION_GRPC_WRITE_ALL, "grpc_write_all"); + assert_eq!(INTERNODE_OPERATION_GRPC_READ_VERSION, "grpc_read_version"); + assert_eq!(INTERNODE_OPERATION_GRPC_BATCH_READ_VERSION, "grpc_batch_read_version"); + assert_eq!(INTERNODE_OPERATION_GRPC_LOCK, "grpc_lock"); + assert_eq!(INTERNODE_OPERATION_GRPC_UNLOCK, "grpc_unlock"); + assert_eq!(INTERNODE_OPERATION_GRPC_LOCK_BATCH, "grpc_lock_batch"); + assert_eq!(INTERNODE_OPERATION_GRPC_UNLOCK_BATCH, "grpc_unlock_batch"); + assert_eq!(INTERNODE_OPERATION_GRPC_REFRESH, "grpc_refresh"); + assert_eq!(INTERNODE_OPERATION_GRPC_FORCE_UNLOCK, "grpc_force_unlock"); assert_eq!(INTERNODE_OPERATION_GRPC_OTHER, "grpc_other"); assert_eq!(INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, "tcp-http"); @@ -1048,26 +1089,30 @@ mod tests { ); assert_eq!( INTERNODE_OPERATION_METRICS[14].name, - "rustfs_system_network_internode_replay_cache_entries" + "rustfs_system_network_internode_replay_cache_records_total" ); assert_eq!( INTERNODE_OPERATION_METRICS[15].name, - "rustfs_system_network_internode_replay_cache_capacity" + "rustfs_system_network_internode_replay_cache_entries" ); assert_eq!( INTERNODE_OPERATION_METRICS[16].name, - "rustfs_system_network_internode_replay_cache_evictions_total" + "rustfs_system_network_internode_replay_cache_capacity" ); assert_eq!( INTERNODE_OPERATION_METRICS[17].name, - "rustfs_system_storage_erasure_write_quorum_failures_total" + "rustfs_system_network_internode_replay_cache_evictions_total" ); assert_eq!( INTERNODE_OPERATION_METRICS[18].name, - "rustfs_system_network_internode_operation_payload_bytes" + "rustfs_system_storage_erasure_write_quorum_failures_total" ); assert_eq!( INTERNODE_OPERATION_METRICS[19].name, + "rustfs_system_network_internode_operation_payload_bytes" + ); + assert_eq!( + INTERNODE_OPERATION_METRICS[20].name, "rustfs_system_network_internode_operation_large_payloads_total" ); assert_eq!(INTERNODE_OPERATION_GRPC_READ_MULTIPLE, "grpc_read_multiple"); @@ -1091,6 +1136,10 @@ mod tests { INTERNODE_SIGNATURE_V1_FALLBACK_TOTAL, "rustfs_system_network_internode_signature_v1_fallback_total" ); + assert_eq!( + INTERNODE_REPLAY_CACHE_RECORDS_TOTAL, + "rustfs_system_network_internode_replay_cache_records_total" + ); assert_eq!(FAILURE_REASON_LABEL, "failure_reason"); assert_eq!(RPC_PATH_LABEL, "rpc_path"); assert_eq!(REASON_LABEL, "reason"); @@ -1144,6 +1193,11 @@ mod tests { INTERNODE_TRANSPORT_BACKEND_GRPC, "/node_service.NodeService/ReadAll", ); + metrics.record_replay_cache_record_for_operation_and_backend_path( + INTERNODE_OPERATION_GRPC_READ_VERSION, + INTERNODE_TRANSPORT_BACKEND_GRPC, + "/node_service.NodeService/ReadVersion", + ); }); let snapshot = metrics.snapshot(); @@ -1179,6 +1233,28 @@ mod tests { assert_eq!(labels.get(BACKEND_LABEL).map(String::as_str), Some(INTERNODE_TRANSPORT_BACKEND_GRPC)); assert_eq!(labels.get(RPC_PATH_LABEL).map(String::as_str), Some("/node_service.NodeService/ReadAll")); assert!(labels.get(SERVER_LABEL).is_some_and(|value| !value.is_empty())); + + let records: Vec<_> = entries + .iter() + .filter(|(composite, _, _, _)| composite.key().name() == INTERNODE_REPLAY_CACHE_RECORDS_TOTAL) + .collect(); + assert_eq!(records.len(), 1); + let labels: HashMap<_, _> = records[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(RPC_PATH_LABEL).map(String::as_str), + Some("/node_service.NodeService/ReadVersion") + ); + assert!(labels.get(SERVER_LABEL).is_some_and(|value| !value.is_empty())); } #[test]