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

@@ -9337,6 +9337,7 @@ version = "1.0.0-beta.8"
 dependencies = [
  "criterion",
  "metrics",
+ "metrics-util",
  "num_cpus",
  "rustfs-s3-ops",
  "sysinfo",

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]

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<HashSet<String>> {
+        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<String> = 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::<Vec<_>>()
+                    .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"
+        );
+    }
+}

rustfs/src/storage/ecfs.rs

@@ -386,7 +386,7 @@ impl S3 for FS {
         &self,
         req: S3Request<DeleteObjectTaggingInput>,
     ) -> S3Result<S3Response<DeleteObjectTaggingOutput>> {
-        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<GetBucketAclInput>) -> S3Result<S3Response<GetBucketAclOutput>> {
-        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<GetBucketCorsInput>) -> S3Result<S3Response<GetBucketCorsOutput>> {
-        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<GetBucketEncryptionInput>,
     ) -> S3Result<S3Response<GetBucketEncryptionOutput>> {
-        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<GetBucketLifecycleConfigurationInput>,
     ) -> S3Result<S3Response<GetBucketLifecycleConfigurationOutput>> {
-        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<GetBucketLocationInput>) -> S3Result<S3Response<GetBucketLocationOutput>> {
-        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<GetBucketNotificationConfigurationInput>,
     ) -> S3Result<S3Response<GetBucketNotificationConfigurationOutput>> {
-        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<GetBucketPolicyInput>) -> S3Result<S3Response<GetBucketPolicyOutput>> {
-        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<GetBucketPolicyStatusInput>,
     ) -> S3Result<S3Response<GetBucketPolicyStatusOutput>> {
-        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<GetBucketReplicationInput>,
     ) -> S3Result<S3Response<GetBucketReplicationOutput>> {
-        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<GetBucketTaggingInput>) -> S3Result<S3Response<GetBucketTaggingOutput>> {
-        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<GetPublicAccessBlockInput>,
     ) -> S3Result<S3Response<GetPublicAccessBlockOutput>> {
-        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<GetBucketVersioningInput>,
     ) -> S3Result<S3Response<GetBucketVersioningOutput>> {
-        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<GetObjectAclInput>) -> S3Result<S3Response<GetObjectAclOutput>> {
-        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<GetObjectLockConfigurationInput>,
     ) -> S3Result<S3Response<GetObjectLockConfigurationOutput>> {
-        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<GetObjectTaggingInput>) -> S3Result<S3Response<GetObjectTaggingOutput>> {
-        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<GetObjectTorrentInput>) -> S3Result<S3Response<GetObjectTorrentOutput>> {
+    #[instrument(level = "debug", skip(self, _req))]
+    async fn get_object_torrent(&self, _req: S3Request<GetObjectTorrentInput>) -> S3Result<S3Response<GetObjectTorrentOutput>> {
         // 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<ListBucketsInput>) -> S3Result<S3Response<ListBucketsOutput>> {
         // 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<ListMultipartUploadsInput>,
     ) -> S3Result<S3Response<ListMultipartUploadsOutput>> {
-        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<ListObjectVersionsInput>,
     ) -> S3Result<S3Response<ListObjectVersionsOutput>> {
-        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<ListObjectsInput>) -> S3Result<S3Response<ListObjectsOutput>> {
-        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<ListObjectsV2Input>) -> S3Result<S3Response<ListObjectsV2Output>> {
         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<ListPartsInput>) -> S3Result<S3Response<ListPartsOutput>> {
-        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<GetBucketLoggingInput>) -> S3Result<S3Response<GetBucketLoggingOutput>> {
-        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<PutBucketLoggingInput>) -> S3Result<S3Response<PutBucketLoggingOutput>> {
-        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<PutObjectAclInput>) -> S3Result<S3Response<PutObjectAclOutput>> {
-        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<PutObjectTaggingInput>) -> S3Result<S3Response<PutObjectTaggingOutput>> {
-        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<UploadPartInput>) -> S3Result<S3Response<UploadPartOutput>> {
         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<UploadPartCopyInput>) -> S3Result<S3Response<UploadPartCopyOutput>> {
-        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
     }

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 {