From 5e7e25b7d1a439313e869e607334902423fd5d5d Mon Sep 17 00:00:00 2001 From: houseme Date: Thu, 6 Aug 2026 23:42:59 +0800 Subject: [PATCH] perf(metrics): count internode RPC auth failures (#5777) Co-authored-by: heihutu Co-authored-by: zhi22915 --- crates/io-metrics/src/internode_metrics.rs | 92 ++++++++++++++++++++-- rustfs/src/server/http.rs | 61 ++++++++++++++ rustfs/src/storage/rpc/http_service.rs | 45 ++++++++++- 3 files changed, 190 insertions(+), 8 deletions(-) diff --git a/crates/io-metrics/src/internode_metrics.rs b/crates/io-metrics/src/internode_metrics.rs index 790536b4a..4cebf3202 100644 --- a/crates/io-metrics/src/internode_metrics.rs +++ b/crates/io-metrics/src/internode_metrics.rs @@ -27,6 +27,7 @@ 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_OTHER: &str = "grpc_other"; 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"; @@ -45,6 +46,7 @@ const CLASSIFICATION_LABEL: &str = "classification"; const STAGE_LABEL: &str = "stage"; const DOMINANT_ERROR_LABEL: &str = "dominant_error"; const HTTP_VERSION_LABEL: &str = "http_version"; +const FAILURE_REASON_LABEL: &str = "failure_reason"; const DIRECTION_LABEL: &str = "direction"; const MESSAGE_LABEL: &str = "message"; const CODEC_LABEL: &str = "codec"; @@ -61,6 +63,7 @@ const INTERNODE_OPERATION_HTTP_VERSIONS_TOTAL: &str = "rustfs_system_network_int const INTERNODE_OPERATION_STALL_TIMEOUTS_TOTAL: &str = "rustfs_system_network_internode_operation_stall_timeouts_total"; const INTERNODE_OPERATION_WRITE_SHUTDOWN_ERRORS_TOTAL: &str = "rustfs_system_network_internode_operation_write_shutdown_errors_total"; +const INTERNODE_RPC_AUTH_FAILURES_TOTAL: &str = "rustfs_system_network_internode_rpc_auth_failures_total"; const INTERNODE_OPERATION_PAYLOAD_BYTES: &str = "rustfs_system_network_internode_operation_payload_bytes"; const INTERNODE_OPERATION_LARGE_PAYLOADS_TOTAL: &str = "rustfs_system_network_internode_operation_large_payloads_total"; const INTERNODE_MSGPACK_JSON_DECODE_TOTAL: &str = "rustfs_system_network_internode_msgpack_json_decode_total"; @@ -82,6 +85,8 @@ const SERVER_OPERATION_BACKEND_LABELS: &[&str] = &[SERVER_LABEL, OPERATION_LABEL const SERVER_OPERATION_BACKEND_CLASSIFICATION_LABELS: &[&str] = &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL, CLASSIFICATION_LABEL]; const SERVER_OPERATION_BACKEND_HTTP_VERSION_LABELS: &[&str] = &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL, HTTP_VERSION_LABEL]; +const SERVER_OPERATION_BACKEND_FAILURE_REASON_LABELS: &[&str] = + &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL, FAILURE_REASON_LABEL]; const SERVER_QUORUM_FAILURE_LABELS: &[&str] = &[SERVER_LABEL, STAGE_LABEL, DOMINANT_ERROR_LABEL]; pub const INTERNODE_OPERATION_METRICS: &[InternodeOperationMetricDescriptor] = &[ @@ -133,6 +138,10 @@ pub const INTERNODE_OPERATION_METRICS: &[InternodeOperationMetricDescriptor] = & name: INTERNODE_OPERATION_WRITE_SHUTDOWN_ERRORS_TOTAL, labels: SERVER_OPERATION_BACKEND_LABELS, }, + InternodeOperationMetricDescriptor { + name: INTERNODE_RPC_AUTH_FAILURES_TOTAL, + labels: SERVER_OPERATION_BACKEND_FAILURE_REASON_LABELS, + }, InternodeOperationMetricDescriptor { name: ERASURE_WRITE_QUORUM_FAILURES_TOTAL, labels: SERVER_QUORUM_FAILURE_LABELS, @@ -178,6 +187,7 @@ pub struct InternodeMetricsSnapshot { pub operation_http_versions_total: u64, pub operation_stall_timeouts_total: u64, pub operation_write_shutdown_errors_total: u64, + pub rpc_auth_failures_total: u64, pub signature_v1_fallback_total: u64, pub body_digest_fallback_total: u64, pub replay_scope_fallback_total: u64, @@ -198,6 +208,7 @@ pub struct InternodeMetrics { operation_http_versions_total: AtomicU64, operation_stall_timeouts_total: AtomicU64, operation_write_shutdown_errors_total: AtomicU64, + rpc_auth_failures_total: AtomicU64, msgpack_json_decode_total: AtomicU64, msgpack_json_decode_error_total: AtomicU64, signature_v1_fallback_total: AtomicU64, @@ -423,6 +434,23 @@ impl InternodeMetrics { .increment(1); } + pub fn record_rpc_auth_failure_for_operation_and_backend( + &self, + operation: &'static str, + backend: &'static str, + failure_reason: &'static str, + ) { + self.rpc_auth_failures_total.fetch_add(1, Ordering::Relaxed); + counter!( + INTERNODE_RPC_AUTH_FAILURES_TOTAL, + SERVER_LABEL => current_server_label(), + OPERATION_LABEL => operation, + BACKEND_LABEL => backend, + FAILURE_REASON_LABEL => failure_reason + ) + .increment(1); + } + /// Record the payload size (bytes) of a completed internode operation into a histogram /// keyed by operation+backend. Used to size which unary `bytes`-carrying RPCs /// (`ReadAll`/`ReadMultiple`/`WriteAll`) would benefit from being moved off the shared @@ -585,6 +613,7 @@ impl InternodeMetrics { operation_http_versions_total: self.operation_http_versions_total.load(Ordering::Relaxed), operation_stall_timeouts_total: self.operation_stall_timeouts_total.load(Ordering::Relaxed), operation_write_shutdown_errors_total: self.operation_write_shutdown_errors_total.load(Ordering::Relaxed), + rpc_auth_failures_total: self.rpc_auth_failures_total.load(Ordering::Relaxed), signature_v1_fallback_total: self.signature_v1_fallback_total.load(Ordering::Relaxed), body_digest_fallback_total: self.body_digest_fallback_total.load(Ordering::Relaxed), replay_scope_fallback_total: self.replay_scope_fallback_total.load(Ordering::Relaxed), @@ -606,6 +635,7 @@ impl InternodeMetrics { self.operation_http_versions_total.store(0, Ordering::Relaxed); self.operation_stall_timeouts_total.store(0, Ordering::Relaxed); self.operation_write_shutdown_errors_total.store(0, Ordering::Relaxed); + self.rpc_auth_failures_total.store(0, Ordering::Relaxed); self.msgpack_json_decode_total.store(0, Ordering::Relaxed); self.msgpack_json_decode_error_total.store(0, Ordering::Relaxed); self.signature_v1_fallback_total.store(0, Ordering::Relaxed); @@ -778,7 +808,7 @@ mod tests { use super::*; use metrics::with_local_recorder; use metrics_util::debugging::DebuggingRecorder; - use std::collections::HashSet; + use std::collections::{HashMap, HashSet}; #[test] fn snapshot_reports_recorded_values() { @@ -829,6 +859,11 @@ mod tests { INTERNODE_TRANSPORT_BACKEND_GRPC, ); metrics.record_error_for_operation_and_backend(INTERNODE_OPERATION_WALK_DIR, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP); + metrics.record_rpc_auth_failure_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_OTHER, + INTERNODE_TRANSPORT_BACKEND_GRPC, + "missing_v2_signature", + ); let snapshot = metrics.snapshot(); assert_eq!(snapshot.sent_bytes_total, 128); @@ -836,11 +871,12 @@ mod tests { assert_eq!(snapshot.outgoing_requests_total, 1); assert_eq!(snapshot.incoming_requests_total, 1); assert_eq!(snapshot.errors_total, 1); + assert_eq!(snapshot.rpc_auth_failures_total, 1); } #[test] fn operation_metric_descriptors_include_backend_and_operation_labels() { - assert_eq!(INTERNODE_OPERATION_METRICS.len(), 15); + assert_eq!(INTERNODE_OPERATION_METRICS.len(), 16); for metric in &INTERNODE_OPERATION_METRICS[..6] { assert_eq!(metric.labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL]); } @@ -854,10 +890,14 @@ mod tests { for metric in &INTERNODE_OPERATION_METRICS[10..12] { assert_eq!(metric.labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL]); } - assert_eq!(INTERNODE_OPERATION_METRICS[12].labels, &[SERVER_LABEL, STAGE_LABEL, DOMINANT_ERROR_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, STAGE_LABEL, DOMINANT_ERROR_LABEL]); // Payload histogram + large-payload counter carry operation+backend labels. - assert_eq!(INTERNODE_OPERATION_METRICS[13].labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL]); assert_eq!(INTERNODE_OPERATION_METRICS[14].labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL]); + assert_eq!(INTERNODE_OPERATION_METRICS[15].labels, &[SERVER_LABEL, OPERATION_LABEL, BACKEND_LABEL]); } #[test] @@ -867,6 +907,7 @@ 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_OTHER, "grpc_other"); assert_eq!(INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, "tcp-http"); assert_eq!(INTERNODE_TRANSPORT_BACKEND_GRPC, "grpc"); @@ -902,14 +943,18 @@ mod tests { ); assert_eq!( INTERNODE_OPERATION_METRICS[12].name, - "rustfs_system_storage_erasure_write_quorum_failures_total" + "rustfs_system_network_internode_rpc_auth_failures_total" ); assert_eq!( INTERNODE_OPERATION_METRICS[13].name, - "rustfs_system_network_internode_operation_payload_bytes" + "rustfs_system_storage_erasure_write_quorum_failures_total" ); assert_eq!( INTERNODE_OPERATION_METRICS[14].name, + "rustfs_system_network_internode_operation_payload_bytes" + ); + assert_eq!( + INTERNODE_OPERATION_METRICS[15].name, "rustfs_system_network_internode_operation_large_payloads_total" ); assert_eq!(INTERNODE_OPERATION_GRPC_READ_MULTIPLE, "grpc_read_multiple"); @@ -933,6 +978,41 @@ mod tests { INTERNODE_SIGNATURE_V1_FALLBACK_TOTAL, "rustfs_system_network_internode_signature_v1_fallback_total" ); + assert_eq!(FAILURE_REASON_LABEL, "failure_reason"); + } + + #[test] + fn rpc_auth_failure_counter_records_low_cardinality_labels() { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let metrics = InternodeMetrics::default(); + + with_local_recorder(&recorder, || { + metrics.record_rpc_auth_failure_for_operation_and_backend( + INTERNODE_OPERATION_GRPC_READ_ALL, + INTERNODE_TRANSPORT_BACKEND_GRPC, + "invalid_v2_signature", + ); + }); + + assert_eq!(metrics.snapshot().rpc_auth_failures_total, 1); + let entries: Vec<_> = snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter(|(composite, _, _, _)| composite.key().name() == INTERNODE_RPC_AUTH_FAILURES_TOTAL) + .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_ALL)); + assert_eq!(labels.get(BACKEND_LABEL).map(String::as_str), Some(INTERNODE_TRANSPORT_BACKEND_GRPC)); + assert_eq!(labels.get(FAILURE_REASON_LABEL).map(String::as_str), Some("invalid_v2_signature")); + assert!(labels.get(SERVER_LABEL).is_some_and(|value| !value.is_empty())); } #[test] diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index c832e3a92..62ddb3736 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -53,6 +53,10 @@ use metrics::{counter, gauge, histogram}; use opentelemetry::global; use opentelemetry::trace::TraceContextExt; use rustfs_common::GlobalReadiness; +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, +}; use rustfs_keystone::KeystoneAuthLayer; #[cfg(feature = "swift")] use rustfs_protocols::SwiftService; @@ -1903,6 +1907,17 @@ fn check_auth(req: Request<()>) -> std::result::Result, Status> { .map(|addr| addr.0.to_string()) .unwrap_or_else(|| "unknown".to_string()); let failure_reason = storage::tonic_rpc_auth_failure_reason(&e); + let operation = match rpc_method { + "ReadAll" => INTERNODE_OPERATION_GRPC_READ_ALL, + "ReadMultiple" => INTERNODE_OPERATION_GRPC_READ_MULTIPLE, + "WriteAll" => INTERNODE_OPERATION_GRPC_WRITE_ALL, + _ => INTERNODE_OPERATION_GRPC_OTHER, + }; + global_internode_metrics().record_rpc_auth_failure_for_operation_and_backend( + operation, + INTERNODE_TRANSPORT_BACKEND_GRPC, + failure_reason, + ); error!( event = EVENT_RPC_SIGNATURE_VERIFICATION_FAILED, component = LOG_COMPONENT_SERVER, @@ -2025,7 +2040,10 @@ mod tests { use http::Request as HttpRequest; use http::{HeaderMap, StatusCode}; use http_body_util::{Empty, Full}; + use metrics::with_local_recorder; + use metrics_util::debugging::DebuggingRecorder; use opentelemetry::propagation::Extractor; + use std::collections::HashMap; use std::convert::Infallible; use std::future::Ready; use std::sync::{Arc, Mutex}; @@ -2394,6 +2412,49 @@ mod tests { rustfs_common::set_global_local_node_name(&previous_node_name).await; } + #[tokio::test] + #[serial_test::serial] + async fn rpc_auth_rejection_records_failure_reason_metric() { + let _ = rustfs_credentials::set_global_rpc_secret("rpc-http-test-secret".to_string()); + let previous_node_name = rustfs_common::get_global_local_node_name().await; + rustfs_common::set_global_local_node_name("127.0.0.1:9000").await; + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + + with_local_recorder(&recorder, || { + let mut request = Request::new(()); + request.extensions_mut().insert(RpcRequestTarget { + uri: "http://127.0.0.1:9000/node_service.NodeService/ReadAll" + .parse() + .expect("test RPC URI should parse"), + method: Method::POST, + }); + let error = check_auth(request).expect_err("missing signature must be rejected"); + assert_eq!(error.code(), tonic::Code::Unauthenticated); + assert_eq!(error.message(), "No valid auth token"); + }); + + let entries: Vec<_> = snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter(|(composite, _, _, _)| composite.key().name() == "rustfs_system_network_internode_rpc_auth_failures_total") + .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").map(String::as_str), Some(INTERNODE_OPERATION_GRPC_READ_ALL)); + assert_eq!(labels.get("backend").map(String::as_str), Some(INTERNODE_TRANSPORT_BACKEND_GRPC)); + assert_eq!(labels.get("failure_reason").map(String::as_str), Some("missing_v1_signature")); + assert!(labels.get("server").is_some_and(|value| !value.is_empty())); + + rustfs_common::set_global_local_node_name(&previous_node_name).await; + } + /// Rolling-upgrade compatibility anchor for : /// a legacy-only peer (constant-target signature, no v2 headers) must keep authenticating /// through the real production path (`check_auth` + `RpcRequestTarget` extension), and every diff --git a/rustfs/src/storage/rpc/http_service.rs b/rustfs/src/storage/rpc/http_service.rs index 7fdeaea1a..dad31bc4c 100644 --- a/rustfs/src/storage/rpc/http_service.rs +++ b/rustfs/src/storage/rpc/http_service.rs @@ -26,6 +26,7 @@ use crate::storage::storage_api::rpc_consumer::http_service::{ WALK_DIR_BODY_SHA256_QUERY, }; use crate::storage::storage_api::runtime_sources_consumer::runtime_sources; +use crate::storage::storage_api::tonic_rpc_auth_failure_reason; use bytes::{Bytes, BytesMut}; use futures_util::{Stream, StreamExt, TryStreamExt, stream}; use http::{HeaderMap, HeaderValue, Method, Request, Response, StatusCode, Uri}; @@ -472,11 +473,17 @@ fn verify_internode_rpc_signature(uri: &Uri, method: &Method, headers: &HeaderMa verify_rpc_signature(&uri.to_string(), method, headers).map_err(|e| { let message = format!("rpc signature verification failed: {e}"); + let operation = internode_http_operation(uri.path()); + runtime_sources::current_internode_metrics().record_rpc_auth_failure_for_operation_and_backend( + operation.unwrap_or(RPC_OPERATION_UNKNOWN), + INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, + tonic_rpc_auth_failure_reason(&e), + ); log_internode_rpc_response_failure!( StatusCode::FORBIDDEN, uri.path(), method, - internode_http_operation(uri.path()), + operation, "signature_verification_failed", "rejected", None, @@ -1309,11 +1316,14 @@ mod tests { use bytes::Bytes; use http::{HeaderMap, HeaderValue, Method, StatusCode, Uri}; use http_body_util::BodyExt; + use metrics::with_local_recorder; + use metrics_util::debugging::DebuggingRecorder; use rustfs_io_metrics::internode_metrics::{ INTERNODE_OPERATION_NS_SCANNER, INTERNODE_OPERATION_PUT_FILE_STREAM, INTERNODE_OPERATION_READ_FILE_STREAM, - INTERNODE_OPERATION_WALK_DIR, global_internode_metrics, + INTERNODE_OPERATION_WALK_DIR, INTERNODE_TRANSPORT_BACKEND_TCP_HTTP, global_internode_metrics, }; use sha2::Digest as _; + use std::collections::HashMap; use tokio::io; use tokio::io::{AsyncReadExt, AsyncWriteExt}; use tokio_stream::StreamExt; @@ -1384,6 +1394,37 @@ mod tests { assert_eq!(response.status(), StatusCode::FORBIDDEN); } + #[test] + fn rpc_get_request_auth_failure_records_failure_reason_metric() { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + let uri: Uri = READ_FILE_STREAM_PATH.parse().expect("uri"); + let headers = HeaderMap::new(); + + with_local_recorder(&recorder, || { + let response = verify_internode_rpc_signature(&uri, &Method::GET, &headers).expect_err("response"); + assert_eq!(response.status(), StatusCode::FORBIDDEN); + }); + + let entries: Vec<_> = snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter(|(composite, _, _, _)| composite.key().name() == "rustfs_system_network_internode_rpc_auth_failures_total") + .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").map(String::as_str), Some(INTERNODE_OPERATION_READ_FILE_STREAM)); + assert_eq!(labels.get("backend").map(String::as_str), Some(INTERNODE_TRANSPORT_BACKEND_TCP_HTTP)); + assert_eq!(labels.get("failure_reason").map(String::as_str), Some("missing_v1_signature")); + assert!(labels.get("server").is_some_and(|value| !value.is_empty())); + } + #[test] fn put_file_stage_error_message_includes_stage_and_request_context() { let query = PutFileQuery {