From ed36d1ed97a99ddebedee7260f21773eaf2a8d37 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Thu, 9 Jul 2026 05:42:27 +0800 Subject: [PATCH] fix(io-metrics): drop client-controlled bucket label from s3 ops counter (backlog#806) (#4580) fix(io-metrics): drop client-controlled bucket label from rustfs_s3_operations_total (backlog#806-22) The bucket dimension on the rustfs_s3_operations_total counter is client-controlled and unbounded, letting any client explode metric cardinality (a Prometheus/OTEL cardinality DoS that grows the in-process OTEL aggregation store even without a scrape). Drop the bucket label so the series is keyed by the bounded op dimension only (<=122 variants), matching MinIO which never labels its default operation counters with bucket. BREAKING: rustfs_s3_operations_total no longer carries a "bucket" label. record_s3_op(op, bucket) -> record_s3_op(op); all call sites in rustfs/src/storage/{ecfs.rs,helper.rs} updated (bucket bindings retained where still used by audit/notify paths). Add metrics-util debugging recorder dev-dep and tests asserting the series carries only the op label and that series count equals distinct-op count. --- Cargo.lock | 1 + crates/io-metrics/Cargo.toml | 1 + crates/io-metrics/src/s3_api_metrics.rs | 91 ++++++++++++++++++++++++- rustfs/src/storage/ecfs.rs | 64 ++++++++--------- rustfs/src/storage/helper.rs | 3 +- 5 files changed, 124 insertions(+), 36 deletions(-) diff --git a/Cargo.lock b/Cargo.lock index 736d5adb6..c960d79f3 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -9337,6 +9337,7 @@ version = "1.0.0-beta.8" dependencies = [ "criterion", "metrics", + "metrics-util", "num_cpus", "rustfs-s3-ops", "sysinfo", diff --git a/crates/io-metrics/Cargo.toml b/crates/io-metrics/Cargo.toml index 2e73ffc1a..ad4585e83 100644 --- a/crates/io-metrics/Cargo.toml +++ b/crates/io-metrics/Cargo.toml @@ -39,6 +39,7 @@ sysinfo = { workspace = true } [dev-dependencies] criterion = { workspace = true } +metrics-util = { version = "0.20", features = ["debugging"] } tokio = { workspace = true, features = ["test-util","rt","macros"] } [lints] diff --git a/crates/io-metrics/src/s3_api_metrics.rs b/crates/io-metrics/src/s3_api_metrics.rs index 90dd07042..251d76312 100644 --- a/crates/io-metrics/src/s3_api_metrics.rs +++ b/crates/io-metrics/src/s3_api_metrics.rs @@ -17,8 +17,16 @@ use std::sync::OnceLock; const S3_OPS_METRIC: &str = "rustfs_s3_operations_total"; -pub fn record_s3_op(op: S3Operation, bucket: &str) { - counter!(S3_OPS_METRIC, "op" => op.as_str(), "bucket" => bucket.to_owned()).increment(1); +/// Record a handled S3 API operation. +/// +/// The series is labeled by `op` only. The bucket name is deliberately NOT a +/// label: it is client-controlled and unbounded, so labeling by bucket would +/// let any client explode metric cardinality (a Prometheus/OTEL cardinality +/// DoS that grows the in-process OTEL aggregation store even without a scrape). +/// This mirrors MinIO, which never labels its default operation counters with +/// bucket. The `op` dimension is bounded (<= 122 variants). +pub fn record_s3_op(op: S3Operation) { + counter!(S3_OPS_METRIC, "op" => op.as_str()).increment(1); } pub fn init_s3_metrics() { @@ -27,3 +35,82 @@ pub fn init_s3_metrics() { describe_counter!(S3_OPS_METRIC, "Total number of S3 API operations handled"); }); } + +#[cfg(test)] +mod tests { + use super::*; + use metrics::with_local_recorder; + use metrics_util::debugging::DebuggingRecorder; + use std::collections::HashSet; + + /// Collect the label-key sets recorded against `rustfs_s3_operations_total`. + fn ops_metric_label_key_sets(recorder: &DebuggingRecorder) -> Vec> { + let snapshotter = recorder.snapshotter(); + with_local_recorder(recorder, || { + record_s3_op(S3Operation::GetObject); + }); + snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter(|(composite, _, _, _)| composite.key().name() == S3_OPS_METRIC) + .map(|(composite, _, _, _)| composite.key().labels().map(|label| label.key().to_string()).collect()) + .collect() + } + + #[test] + fn record_s3_op_labels_by_op_only_no_bucket() { + let recorder = DebuggingRecorder::new(); + let label_key_sets = ops_metric_label_key_sets(&recorder); + + assert_eq!(label_key_sets.len(), 1, "exactly one series expected for a single op"); + let keys = &label_key_sets[0]; + assert_eq!( + keys, + &HashSet::from(["op".to_string()]), + "series must carry the op label only; the client-controlled bucket label must be absent" + ); + assert!(!keys.contains("bucket"), "bucket label must never be emitted (cardinality DoS)"); + } + + #[test] + fn record_s3_op_cardinality_bounded_by_distinct_ops() { + let recorder = DebuggingRecorder::new(); + let snapshotter = recorder.snapshotter(); + + // Same op recorded twice + two other ops => distinct series == distinct ops. + let ops = [ + S3Operation::GetObject, + S3Operation::GetObject, + S3Operation::PutObject, + S3Operation::ListObjectsV2, + ]; + with_local_recorder(&recorder, || { + for op in ops { + record_s3_op(op); + } + }); + + let series: HashSet = snapshotter + .snapshot() + .into_vec() + .into_iter() + .filter(|(composite, _, _, _)| composite.key().name() == S3_OPS_METRIC) + .map(|(composite, _, _, _)| { + composite + .key() + .labels() + .map(|label| format!("{}={}", label.key(), label.value())) + .collect::>() + .join(",") + }) + .collect(); + + let distinct_ops: HashSet<&str> = ops.iter().map(|op| op.as_str()).collect(); + assert_eq!( + series.len(), + distinct_ops.len(), + "series count must equal the number of distinct ops, never the bucket count" + ); + } +} diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index 7b1f0226f..70f10b4c2 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -386,7 +386,7 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { - record_s3_op(S3Operation::DeleteObjectTagging, &req.input.bucket); + record_s3_op(S3Operation::DeleteObjectTagging); let start_time = std::time::Instant::now(); let mut helper = OperationHelper::new(&req, EventName::ObjectTaggingDelete, S3Operation::DeleteObjectTagging); let DeleteObjectTaggingInput { @@ -477,7 +477,7 @@ impl S3 for FS { } async fn get_bucket_acl(&self, req: S3Request) -> S3Result> { - record_s3_op(S3Operation::GetBucketAcl, &req.input.bucket); + record_s3_op(S3Operation::GetBucketAcl); let GetBucketAclInput { bucket, .. } = req.input; let Some(store) = runtime_sources::current_object_store_handle() else { @@ -517,7 +517,7 @@ impl S3 for FS { #[instrument(level = "debug", skip(self))] async fn get_bucket_cors(&self, req: S3Request) -> S3Result> { - record_s3_op(S3Operation::GetBucketCors, &req.input.bucket); + record_s3_op(S3Operation::GetBucketCors); let usecase = s3_api::default_bucket_usecase(); usecase.execute_get_bucket_cors(req).await } @@ -526,7 +526,7 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { - record_s3_op(S3Operation::GetBucketEncryption, &req.input.bucket); + record_s3_op(S3Operation::GetBucketEncryption); let usecase = s3_api::default_bucket_usecase(); usecase.execute_get_bucket_encryption(req).await } @@ -536,7 +536,7 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { - record_s3_op(S3Operation::GetBucketLifecycleConfiguration, &req.input.bucket); + record_s3_op(S3Operation::GetBucketLifecycleConfiguration); let usecase = s3_api::default_bucket_usecase(); usecase.execute_get_bucket_lifecycle_configuration(req).await } @@ -544,7 +544,7 @@ impl S3 for FS { /// Get bucket location #[instrument(level = "debug", skip(self, req))] async fn get_bucket_location(&self, req: S3Request) -> S3Result> { - record_s3_op(S3Operation::GetBucketLocation, &req.input.bucket); + record_s3_op(S3Operation::GetBucketLocation); let usecase = s3_api::default_bucket_usecase(); usecase.execute_get_bucket_location(req).await } @@ -553,13 +553,13 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { - record_s3_op(S3Operation::GetBucketNotificationConfiguration, &req.input.bucket); + record_s3_op(S3Operation::GetBucketNotificationConfiguration); let usecase = s3_api::default_bucket_usecase(); usecase.execute_get_bucket_notification_configuration(req).await } async fn get_bucket_policy(&self, req: S3Request) -> S3Result> { - record_s3_op(S3Operation::GetBucketPolicy, &req.input.bucket); + record_s3_op(S3Operation::GetBucketPolicy); let usecase = s3_api::default_bucket_usecase(); usecase.execute_get_bucket_policy(req).await } @@ -568,7 +568,7 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { - record_s3_op(S3Operation::GetBucketPolicyStatus, &req.input.bucket); + record_s3_op(S3Operation::GetBucketPolicyStatus); let usecase = s3_api::default_bucket_usecase(); usecase.execute_get_bucket_policy_status(req).await } @@ -577,7 +577,7 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { - record_s3_op(S3Operation::GetBucketReplication, &req.input.bucket); + record_s3_op(S3Operation::GetBucketReplication); let usecase = s3_api::default_bucket_usecase(); usecase.execute_get_bucket_replication(req).await } @@ -608,7 +608,7 @@ impl S3 for FS { #[instrument(level = "debug", skip(self))] async fn get_bucket_tagging(&self, req: S3Request) -> S3Result> { - record_s3_op(S3Operation::GetBucketTagging, &req.input.bucket); + record_s3_op(S3Operation::GetBucketTagging); let usecase = s3_api::default_bucket_usecase(); usecase.execute_get_bucket_tagging(req).await } @@ -618,7 +618,7 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { - record_s3_op(S3Operation::GetPublicAccessBlock, &req.input.bucket); + record_s3_op(S3Operation::GetPublicAccessBlock); let usecase = s3_api::default_bucket_usecase(); usecase.execute_get_public_access_block(req).await } @@ -628,7 +628,7 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { - record_s3_op(S3Operation::GetBucketVersioning, &req.input.bucket); + record_s3_op(S3Operation::GetBucketVersioning); let usecase = s3_api::default_bucket_usecase(); usecase.execute_get_bucket_versioning(req).await } @@ -668,7 +668,7 @@ impl S3 for FS { } async fn get_object_acl(&self, req: S3Request) -> S3Result> { - record_s3_op(S3Operation::GetObjectAcl, &req.input.bucket); + record_s3_op(S3Operation::GetObjectAcl); let GetObjectAclInput { bucket, key, version_id, .. } = req.input; @@ -761,7 +761,7 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { - record_s3_op(S3Operation::GetObjectLockConfiguration, &req.input.bucket); + record_s3_op(S3Operation::GetObjectLockConfiguration); let GetObjectLockConfigurationInput { bucket, .. } = req.input; let Some(store) = runtime_sources::current_object_store_handle() else { @@ -859,7 +859,7 @@ impl S3 for FS { #[instrument(level = "debug", skip(self))] async fn get_object_tagging(&self, req: S3Request) -> S3Result> { - record_s3_op(S3Operation::GetObjectTagging, &req.input.bucket); + record_s3_op(S3Operation::GetObjectTagging); let start_time = std::time::Instant::now(); let bucket = req.input.bucket.as_str(); let object = req.input.key.as_str(); @@ -930,12 +930,12 @@ impl S3 for FS { })) } - #[instrument(level = "debug", skip(self, req))] - async fn get_object_torrent(&self, req: S3Request) -> S3Result> { + #[instrument(level = "debug", skip(self, _req))] + async fn get_object_torrent(&self, _req: S3Request) -> S3Result> { // Torrent functionality is not implemented in RustFS // Per S3 API test expectations, return 404 NoSuchKey (not 501 Not Implemented) // This allows clients to gracefully handle the absence of torrent support - record_s3_op(S3Operation::GetObjectTorrent, &req.input.bucket); + record_s3_op(S3Operation::GetObjectTorrent); Err(S3Error::new(S3ErrorCode::NoSuchKey)) } @@ -955,7 +955,7 @@ impl S3 for FS { #[instrument(level = "debug", skip(self))] async fn list_buckets(&self, req: S3Request) -> S3Result> { // List buckets not associated with a bucket, give it bucket label "*" to denote "all". - record_s3_op(S3Operation::ListBuckets, "*"); + record_s3_op(S3Operation::ListBuckets); let usecase = s3_api::default_bucket_usecase(); usecase.execute_list_buckets(req).await } @@ -964,7 +964,7 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { - record_s3_op(S3Operation::ListMultipartUploads, &req.input.bucket); + record_s3_op(S3Operation::ListMultipartUploads); let usecase = s3_api::default_multipart_usecase(); usecase.execute_list_multipart_uploads(req).await } @@ -973,14 +973,14 @@ impl S3 for FS { &self, req: S3Request, ) -> S3Result> { - record_s3_op(S3Operation::ListObjectVersions, &req.input.bucket); + record_s3_op(S3Operation::ListObjectVersions); let usecase = s3_api::default_bucket_usecase(); usecase.execute_list_object_versions(req).await } #[instrument(level = "debug", skip(self, req))] async fn list_objects(&self, req: S3Request) -> S3Result> { - record_s3_op(S3Operation::ListObjects, &req.input.bucket); + record_s3_op(S3Operation::ListObjects); let usecase = s3_api::default_bucket_usecase(); usecase.execute_list_objects(req).await } @@ -988,14 +988,14 @@ impl S3 for FS { #[instrument(level = "debug", skip(self, req))] async fn list_objects_v2(&self, req: S3Request) -> S3Result> { crate::hp_guard!("S3::list_objects_v2"); - record_s3_op(S3Operation::ListObjectsV2, &req.input.bucket); + record_s3_op(S3Operation::ListObjectsV2); let usecase = s3_api::default_bucket_usecase(); usecase.execute_list_objects_v2(req).await } #[instrument(level = "debug", skip(self, req))] async fn list_parts(&self, req: S3Request) -> S3Result> { - record_s3_op(S3Operation::ListParts, &req.input.bucket); + record_s3_op(S3Operation::ListParts); let usecase = s3_api::default_multipart_usecase(); usecase.execute_list_parts(req).await } @@ -1006,7 +1006,7 @@ impl S3 for FS { access_control_policy, .. } = req.input; - record_s3_op(S3Operation::PutBucketAcl, &bucket); + record_s3_op(S3Operation::PutBucketAcl); let Some(store) = runtime_sources::current_object_store_handle() else { return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); @@ -1055,7 +1055,7 @@ impl S3 for FS { } async fn get_bucket_logging(&self, req: S3Request) -> S3Result> { - record_s3_op(S3Operation::GetBucketLogging, &req.input.bucket); + record_s3_op(S3Operation::GetBucketLogging); let Some(store) = runtime_sources::current_object_store_handle() else { return Err(s3_error!(InternalError, "Not init")); }; @@ -1074,7 +1074,7 @@ impl S3 for FS { } async fn put_bucket_logging(&self, req: S3Request) -> S3Result> { - record_s3_op(S3Operation::PutBucketLogging, &req.input.bucket); + record_s3_op(S3Operation::PutBucketLogging); let Some(store) = runtime_sources::current_object_store_handle() else { return Err(s3_error!(InternalError, "Not init")); }; @@ -1201,7 +1201,7 @@ impl S3 for FS { } async fn put_object_acl(&self, req: S3Request) -> S3Result> { - record_s3_op(S3Operation::PutObjectAcl, &req.input.bucket); + record_s3_op(S3Operation::PutObjectAcl); let mut helper = OperationHelper::new(&req, EventName::ObjectAclPut, S3Operation::PutObjectAcl); let bucket = &req.input.bucket; let key = &req.input.key; @@ -1463,7 +1463,7 @@ impl S3 for FS { #[instrument(level = "debug", skip(self, req))] async fn put_object_tagging(&self, req: S3Request) -> S3Result> { - record_s3_op(S3Operation::PutObjectTagging, &req.input.bucket); + record_s3_op(S3Operation::PutObjectTagging); let start_time = std::time::Instant::now(); let mut helper = OperationHelper::new(&req, EventName::ObjectTaggingPut, S3Operation::PutObjectTagging); let PutObjectTaggingInput { @@ -1557,14 +1557,14 @@ impl S3 for FS { #[instrument(level = "debug", skip(self, req))] async fn upload_part(&self, req: S3Request) -> S3Result> { crate::hp_guard!("S3::upload_part"); - record_s3_op(S3Operation::UploadPart, &req.input.bucket); + record_s3_op(S3Operation::UploadPart); let usecase = s3_api::default_multipart_usecase(); usecase.execute_upload_part(req).await } #[instrument(level = "debug", skip(self, req))] async fn upload_part_copy(&self, req: S3Request) -> S3Result> { - record_s3_op(S3Operation::UploadPartCopy, &req.input.bucket); + record_s3_op(S3Operation::UploadPartCopy); let usecase = s3_api::default_multipart_usecase(); Box::pin(usecase.execute_upload_part_copy(req)).await } diff --git a/rustfs/src/storage/helper.rs b/rustfs/src/storage/helper.rs index 4918c6570..ffe3ea700 100644 --- a/rustfs/src/storage/helper.rs +++ b/rustfs/src/storage/helper.rs @@ -121,8 +121,7 @@ impl OperationHelper { .filter(|value| !value.is_empty()) .unwrap_or(path_bucket); - let bucket_label = if bucket.is_empty() { "*" } else { &bucket }; - record_s3_op(op, bucket_label); + record_s3_op(op); // Fast path: when both chains are disabled, avoid all request parsing/builder work. if !audit_enabled && !notify_enabled {