gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-06 21:33:14 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf(metrics): count internode RPC auth failures (#5777)

Browse Source 
Co-authored-by: heihutu <heihutu@gmail.com>
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
houseme
2026-08-06 23:42:59 +08:00
committed by GitHub
parent da82fd995e
commit 5e7e25b7d1
3 changed files with 190 additions and 8 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/io-metrics/src/internode_metrics.rs
+86 -6
View File
@@ -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]
rustfs/src/server/http.rs
+61
View File
@@ -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<Request<()>, 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 <https://github.com/rustfs/backlog/issues/1327>:
/// a legacy-only peer (constant-target signature, no v2 headers) must keep authenticating
/// through the real production path (`check_auth` + `RpcRequestTarget` extension), and every
rustfs/src/storage/rpc/http_service.rs
+43 -2
View File
@@ -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 {