From 59f41eb86af40513040c72ddd9d955b5459960d6 Mon Sep 17 00:00:00 2001 From: houseme Date: Sun, 26 Apr 2026 02:51:29 +0800 Subject: [PATCH] feat(obs): improve metrics coverage and dashboard performance (#2682) --- .docker/observability/docker-compose.yml | 1 + .../grafana/dashboards/rustfs.json | 48 +-- .../prometheus-rules/rustfs-dashboard.yml | 40 +++ .docker/observability/prometheus.yml | 3 + Cargo.lock | 3 + crates/common/src/internode_metrics.rs | 14 +- .../replication/replication_resyncer.rs | 14 + .../bucket/replication/replication_state.rs | 30 ++ crates/ecstore/src/erasure_coding/encode.rs | 37 +- crates/iam/src/manager.rs | 53 ++- crates/iam/src/oidc.rs | 135 ++++++- crates/iam/src/sys.rs | 6 +- crates/io-metrics/src/adaptive_ttl.rs | 16 +- crates/io-metrics/src/backpressure_metrics.rs | 10 +- crates/io-metrics/src/capacity_metrics.rs | 66 ++-- crates/io-metrics/src/deadlock_metrics.rs | 20 +- crates/io-metrics/src/io_metrics.rs | 34 +- crates/io-metrics/src/lib.rs | 129 ++++--- crates/io-metrics/src/lock_metrics.rs | 14 +- crates/io-metrics/src/metric_names.rs | 4 +- crates/io-metrics/src/timeout_metrics.rs | 12 +- crates/obs/Cargo.toml | 3 + crates/obs/README.md | 24 +- crates/obs/src/cleaner/README.md | 17 +- .../obs/src/metrics/collectors/cluster_iam.rs | 15 +- crates/obs/src/metrics/scheduler.rs | 107 +++++- crates/obs/src/metrics/stats_collector.rs | 332 +++++++++++++++++- crates/obs/src/telemetry/local.rs | 2 +- crates/obs/src/telemetry/otel.rs | 2 +- crates/s3select-api/src/query/execution.rs | 4 +- rustfs/src/admin/handlers/site_replication.rs | 4 +- rustfs/src/app/object_usecase.rs | 4 +- rustfs/src/server/http.rs | 113 ++++-- rustfs/src/storage/access.rs | 4 +- rustfs/src/storage/backpressure.rs | 8 +- rustfs/src/storage/concurrency/io_schedule.rs | 2 +- rustfs/src/storage/deadlock_detector.rs | 2 +- rustfs/src/storage/ecfs.rs | 43 ++- rustfs/src/storage/ecfs_extend.rs | 8 +- rustfs/src/storage/lock_optimizer.rs | 4 +- rustfs/src/storage/request_context.rs | 4 +- scripts/run.sh | 9 +- 42 files changed, 1108 insertions(+), 292 deletions(-) create mode 100644 .docker/observability/prometheus-rules/rustfs-dashboard.yml diff --git a/.docker/observability/docker-compose.yml b/.docker/observability/docker-compose.yml index c84babfa2..26123ae49 100644 --- a/.docker/observability/docker-compose.yml +++ b/.docker/observability/docker-compose.yml @@ -84,6 +84,7 @@ services: container_name: prometheus volumes: - ./prometheus.yml:/etc/prometheus/prometheus.yml:ro + - ./prometheus-rules:/etc/prometheus/rules:ro - prometheus-data:/prometheus ports: - "9090:9090" diff --git a/.docker/observability/grafana/dashboards/rustfs.json b/.docker/observability/grafana/dashboards/rustfs.json index 739320f7b..18c9b1fa6 100644 --- a/.docker/observability/grafana/dashboards/rustfs.json +++ b/.docker/observability/grafana/dashboards/rustfs.json @@ -157,7 +157,7 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "sum(increase(rustfs_api_requests_total{job=~\"$job\"}[$__range]))", + "expr": "sum(increase(rustfs_http_server_requests_total{job=~\"$job\"}[$__range]))", "legendFormat": "__auto", "range": true, "refId": "A" @@ -518,7 +518,7 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "sum(rate(rustfs_api_requests_total{job=~\"$job\"}[5m]))", + "expr": "sum(rate(rustfs_http_server_requests_total{job=~\"$job\"}[5m]))", "hide": false, "instant": false, "legendFormat": "__auto", @@ -605,7 +605,7 @@ }, "id": 11, "panels": [], - "title": "Requests", + "title": "API RED", "type": "row" }, { @@ -699,8 +699,8 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "sum by (request_method) (rate(rustfs_api_requests_total{job=~\"$job\", request_method=~\"$method\"}[$__rate_interval]))", - "legendFormat": "{{request_method}}", + "expr": "sum by (method) (rate(rustfs_http_server_requests_total{job=~\"$job\", method=~\"$method\"}[$__rate_interval]))", + "legendFormat": "{{method}}", "range": true, "refId": "A" } @@ -851,7 +851,7 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "histogram_quantile(0.50, sum by (le, job) (rate(rustfs_request_latency_ms_bucket{job=~\"$job\"}[5m])))", + "expr": "1000 * histogram_quantile(0.50, sum by (le, job) (rate(rustfs_http_server_request_duration_seconds_bucket{job=~\"$job\"}[5m])))", "legendFormat": "P50", "range": true, "refId": "A" @@ -862,7 +862,7 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "histogram_quantile(0.95, sum by (le, job) (rate(rustfs_request_latency_ms_bucket{job=~\"$job\"}[5m])))", + "expr": "1000 * histogram_quantile(0.95, sum by (le, job) (rate(rustfs_http_server_request_duration_seconds_bucket{job=~\"$job\"}[5m])))", "legendFormat": "P95", "range": true, "refId": "B" @@ -873,7 +873,7 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "histogram_quantile(0.99, sum by (le, job) (rate(rustfs_request_latency_ms_bucket{job=~\"$job\"}[5m])))", + "expr": "1000 * histogram_quantile(0.99, sum by (le, job) (rate(rustfs_http_server_request_duration_seconds_bucket{job=~\"$job\"}[5m])))", "legendFormat": "P99", "range": true, "refId": "C" @@ -1025,7 +1025,7 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "histogram_quantile(0.50, sum by (le, job) (rate(rustfs_request_body_len_bucket{job=~\"$job\"}[5m])))", + "expr": "histogram_quantile(0.50, sum by (le, job) (rate(rustfs_http_server_response_body_size_bytes_bucket{job=~\"$job\"}[5m])))", "legendFormat": "P50", "range": true, "refId": "A" @@ -1036,7 +1036,7 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "histogram_quantile(0.95, sum by (le, job) (rate(rustfs_request_body_len_bucket{job=~\"$job\"}[5m])))", + "expr": "histogram_quantile(0.95, sum by (le, job) (rate(rustfs_http_server_response_body_size_bytes_bucket{job=~\"$job\"}[5m])))", "legendFormat": "P95", "range": true, "refId": "B" @@ -1047,13 +1047,13 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "histogram_quantile(0.99, sum by (le, job) (rate(rustfs_request_body_len_bucket{job=~\"$job\"}[5m])))", + "expr": "histogram_quantile(0.99, sum by (le, job) (rate(rustfs_http_server_response_body_size_bytes_bucket{job=~\"$job\"}[5m])))", "legendFormat": "P99", "range": true, "refId": "C" } ], - "title": "Request Body Percentiles", + "title": "Response Body Percentiles", "type": "timeseries" }, { @@ -1566,7 +1566,7 @@ }, "id": 34, "panels": [], - "title": "Buckets", + "title": "Storage and Capacity", "type": "row" }, { @@ -2314,7 +2314,7 @@ }, "id": 12, "panels": [], - "title": "Resource Usage", + "title": "Host and Process USE", "type": "row" }, { @@ -4964,7 +4964,7 @@ }, "id": 114, "panels": [], - "title": "Delivery Targets", + "title": "Security and Delivery", "type": "row" }, { @@ -5463,7 +5463,7 @@ }, "id": 121, "panels": [], - "title": "Scanner Activity", + "title": "Background Services", "type": "row" }, { @@ -6566,7 +6566,7 @@ }, "editorMode": "code", "expr": "rustfs_bucket_replication_proxied_put_tagging_requests_total{job=~\"$job\",bucket=~\"$bucket\"}", - "legendFormat": "put - {{bucket}}", + "legendFormat": "put via tagging - {{bucket}}", "range": true, "refId": "E" }, @@ -6577,12 +6577,12 @@ }, "editorMode": "code", "expr": "rustfs_bucket_replication_proxied_put_tagging_requests_failures_total{job=~\"$job\",bucket=~\"$bucket\"}", - "legendFormat": "put failures - {{bucket}}", + "legendFormat": "put via tagging failures - {{bucket}}", "range": true, "refId": "F" } ], - "title": "Bucket Replication Proxy Requests (Get/Head/Put)", + "title": "Bucket Replication Proxy Requests (Get/Head/Put+Tagging)", "type": "timeseries" }, { @@ -6954,7 +6954,7 @@ }, "id": 137, "panels": [], - "title": "System / Cluster Coverage Gap (By Subsystem)", + "title": "Debug / Raw Explorer", "type": "row" }, { @@ -8155,7 +8155,7 @@ "type": "prometheus", "uid": "${datasource}" }, - "definition": "label_values(rustfs_api_requests_total, job)", + "definition": "label_values(rustfs_http_server_requests_total, job)", "includeAll": true, "label": "Job", "multi": true, @@ -8163,7 +8163,7 @@ "options": [], "query": { "qryType": 1, - "query": "label_values(rustfs_api_requests_total, job)", + "query": "label_values(rustfs_http_server_requests_total, job)", "refId": "PrometheusVariableQueryEditor-VariableQuery" }, "refresh": 2, @@ -8181,7 +8181,7 @@ "type": "prometheus", "uid": "${datasource}" }, - "definition": "label_values(rustfs_api_requests_total,request_method)", + "definition": "label_values(rustfs_http_server_requests_total,method)", "includeAll": true, "label": "Method", "multi": true, @@ -8189,7 +8189,7 @@ "options": [], "query": { "qryType": 1, - "query": "label_values(rustfs_api_requests_total,request_method)", + "query": "label_values(rustfs_http_server_requests_total,method)", "refId": "PrometheusVariableQueryEditor-VariableQuery" }, "refresh": 2, diff --git a/.docker/observability/prometheus-rules/rustfs-dashboard.yml b/.docker/observability/prometheus-rules/rustfs-dashboard.yml new file mode 100644 index 000000000..898b82b2a --- /dev/null +++ b/.docker/observability/prometheus-rules/rustfs-dashboard.yml @@ -0,0 +1,40 @@ +groups: + - name: rustfs-dashboard + interval: 30s + rules: + - record: rustfs:http_server_requests:rate5m + expr: sum by (job) (rate(rustfs_http_server_requests_total[5m])) + + - record: rustfs:http_server_request_duration_seconds:p50_5m + expr: histogram_quantile(0.50, sum by (le, job) (rate(rustfs_http_server_request_duration_seconds_bucket[5m]))) + - record: rustfs:http_server_request_duration_seconds:p95_5m + expr: histogram_quantile(0.95, sum by (le, job) (rate(rustfs_http_server_request_duration_seconds_bucket[5m]))) + - record: rustfs:http_server_request_duration_seconds:p99_5m + expr: histogram_quantile(0.99, sum by (le, job) (rate(rustfs_http_server_request_duration_seconds_bucket[5m]))) + + - record: rustfs:http_server_response_body_size_bytes:p50_5m + expr: histogram_quantile(0.50, sum by (le, job) (rate(rustfs_http_server_response_body_size_bytes_bucket[5m]))) + - record: rustfs:http_server_response_body_size_bytes:p95_5m + expr: histogram_quantile(0.95, sum by (le, job) (rate(rustfs_http_server_response_body_size_bytes_bucket[5m]))) + - record: rustfs:http_server_response_body_size_bytes:p99_5m + expr: histogram_quantile(0.99, sum by (le, job) (rate(rustfs_http_server_response_body_size_bytes_bucket[5m]))) + + - record: rustfs:log_cleaner_runs:rate15m + expr: sum by (job) (rate(rustfs_log_cleaner_runs_total[15m])) + - record: rustfs:log_cleaner_failure_ratio:rate5m + expr: sum by (job) (rate(rustfs_log_cleaner_run_failures_total[5m])) / clamp_min(sum by (job) (rate(rustfs_log_cleaner_runs_total[5m])), 1e-9) + - record: rustfs:log_cleaner_rotation_failure_ratio:rate5m + expr: sum by (job) (rate(rustfs_log_cleaner_rotation_failures_total[5m])) / clamp_min(sum by (job) (rate(rustfs_log_cleaner_rotation_total[5m])), 1e-9) + - record: rustfs:log_cleaner_rotation_duration_seconds:p95_5m + expr: histogram_quantile(0.95, sum by (le, job) (rate(rustfs_log_cleaner_rotation_duration_seconds_bucket[5m]))) + - record: rustfs:log_cleaner_compress_duration_seconds:p95_5m + expr: histogram_quantile(0.95, sum by (le, job) (rate(rustfs_log_cleaner_compress_duration_seconds_bucket[5m]))) + + - record: rustfs:scanner_objects_scanned:rate5m + expr: sum by (job) (rate(rustfs_scanner_objects_scanned_total[5m])) + - record: rustfs:scanner_directories_scanned:rate5m + expr: sum by (job) (rate(rustfs_scanner_directories_scanned_total[5m])) + - record: rustfs:scanner_buckets_scanned:rate5m + expr: sum by (job) (rate(rustfs_scanner_buckets_scanned_total[5m])) + - record: rustfs:scanner_cycles_success:rate5m + expr: sum by (job) (rate(rustfs_scanner_cycles_total{result="success"}[5m])) diff --git a/.docker/observability/prometheus.yml b/.docker/observability/prometheus.yml index d0e638f24..1039e2972 100644 --- a/.docker/observability/prometheus.yml +++ b/.docker/observability/prometheus.yml @@ -19,6 +19,9 @@ global: cluster: 'rustfs-dev' # Label to identify the cluster replica: '1' # Replica identifier +rule_files: + - /etc/prometheus/rules/*.yml + scrape_configs: - job_name: 'otel-collector' static_configs: diff --git a/Cargo.lock b/Cargo.lock index 2c3c0c6f1..fabb72abb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8471,6 +8471,7 @@ dependencies = [ name = "rustfs-obs" version = "0.0.5" dependencies = [ + "chrono", "crossbeam-channel", "crossbeam-deque", "crossbeam-utils", @@ -8490,8 +8491,10 @@ dependencies = [ "percent-encoding", "pyroscope", "rustfs-audit", + "rustfs-common", "rustfs-config", "rustfs-ecstore", + "rustfs-iam", "rustfs-io-metrics", "rustfs-notify", "rustfs-utils", diff --git a/crates/common/src/internode_metrics.rs b/crates/common/src/internode_metrics.rs index 025795817..a089e3067 100644 --- a/crates/common/src/internode_metrics.rs +++ b/crates/common/src/internode_metrics.rs @@ -51,7 +51,7 @@ impl InternodeMetrics { return; } self.sent_bytes_total.fetch_add(bytes, Ordering::Relaxed); - counter!("rustfs.internode.sent.bytes.total").increment(bytes); + counter!("rustfs_system_network_internode_sent_bytes_total").increment(bytes); } pub fn record_recv_bytes(&self, bytes: usize) { @@ -60,22 +60,22 @@ impl InternodeMetrics { return; } self.recv_bytes_total.fetch_add(bytes, Ordering::Relaxed); - counter!("rustfs.internode.recv.bytes.total").increment(bytes); + counter!("rustfs_system_network_internode_recv_bytes_total").increment(bytes); } pub fn record_outgoing_request(&self) { self.outgoing_requests_total.fetch_add(1, Ordering::Relaxed); - counter!("rustfs.internode.requests.outgoing.total").increment(1); + counter!("rustfs_system_network_internode_requests_outgoing_total").increment(1); } pub fn record_incoming_request(&self) { self.incoming_requests_total.fetch_add(1, Ordering::Relaxed); - counter!("rustfs.internode.requests.incoming.total").increment(1); + counter!("rustfs_system_network_internode_requests_incoming_total").increment(1); } pub fn record_error(&self) { self.errors_total.fetch_add(1, Ordering::Relaxed); - counter!("rustfs.internode.errors.total").increment(1); + counter!("rustfs_system_network_internode_errors_total").increment(1); } pub fn record_dial_result(&self, duration: Duration, success: bool) { @@ -83,11 +83,11 @@ impl InternodeMetrics { self.dial_total_time_nanos.fetch_add(elapsed_nanos, Ordering::Relaxed); let samples = self.dial_samples_total.fetch_add(1, Ordering::Relaxed) + 1; let total = self.dial_total_time_nanos.load(Ordering::Relaxed); - gauge!("rustfs.internode.dial.avg_time.nanos").set(total as f64 / samples as f64); + gauge!("rustfs_system_network_internode_dial_avg_time_nanos").set(total as f64 / samples as f64); if !success { self.dial_errors_total.fetch_add(1, Ordering::Relaxed); - counter!("rustfs.internode.dial.errors.total").increment(1); + counter!("rustfs_system_network_internode_dial_errors_total").increment(1); } let now_ms = SystemTime::now() diff --git a/crates/ecstore/src/bucket/replication/replication_resyncer.rs b/crates/ecstore/src/bucket/replication/replication_resyncer.rs index adc7b57b2..36056fa76 100644 --- a/crates/ecstore/src/bucket/replication/replication_resyncer.rs +++ b/crates/ecstore/src/bucket/replication/replication_resyncer.rs @@ -2578,6 +2578,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { } }; + let has_tagging_replication = !put_opts.user_tags.is_empty(); if let Some(err) = if is_multipart { drop(gr); let result = replicate_object_with_multipart(MultipartReplicationContext { @@ -2593,6 +2594,9 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { }) .await; record_proxy_request(&bucket, "PutObject", result.is_err()).await; + if has_tagging_replication { + record_proxy_request(&bucket, "PutObjectTagging", result.is_err()).await; + } result.err() } else { gr.stream = wrap_with_bandwidth_monitor(gr.stream, &put_opts, &bucket, &rinfo.arn); @@ -2602,6 +2606,9 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { .await .map_err(|e| std::io::Error::other(e.to_string())); record_proxy_request(&bucket, "PutObject", result.is_err()).await; + if has_tagging_replication { + record_proxy_request(&bucket, "PutObjectTagging", result.is_err()).await; + } result.err() } { rinfo.replication_status = ReplicationStatusType::Failed; @@ -2867,6 +2874,7 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { } }; + let has_tagging_replication = !put_opts.user_tags.is_empty(); if let Some(err) = if is_multipart { drop(gr); let result = replicate_object_with_multipart(MultipartReplicationContext { @@ -2882,6 +2890,9 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { }) .await; record_proxy_request(&bucket, "PutObject", result.is_err()).await; + if has_tagging_replication { + record_proxy_request(&bucket, "PutObjectTagging", result.is_err()).await; + } result.err() } else { gr.stream = wrap_with_bandwidth_monitor(gr.stream, &put_opts, &bucket, &rinfo.arn); @@ -2891,6 +2902,9 @@ impl ReplicateObjectInfoExt for ReplicateObjectInfo { .await .map_err(|e| std::io::Error::other(e.to_string())); record_proxy_request(&bucket, "PutObject", result.is_err()).await; + if has_tagging_replication { + record_proxy_request(&bucket, "PutObjectTagging", result.is_err()).await; + } result.err() } { rinfo.replication_status = ReplicationStatusType::Failed; diff --git a/crates/ecstore/src/bucket/replication/replication_state.rs b/crates/ecstore/src/bucket/replication/replication_state.rs index ed46557ce..2e9938110 100644 --- a/crates/ecstore/src/bucket/replication/replication_state.rs +++ b/crates/ecstore/src/bucket/replication/replication_state.rs @@ -489,8 +489,14 @@ impl QueueCache { pub struct ProxyMetric { pub get_total: i64, pub get_failed: i64, + pub get_tag_total: i64, + pub get_tag_failed: i64, pub put_total: i64, pub put_failed: i64, + pub put_tag_total: i64, + pub put_tag_failed: i64, + pub delete_tag_total: i64, + pub delete_tag_failed: i64, pub head_total: i64, pub head_failed: i64, } @@ -499,8 +505,14 @@ impl ProxyMetric { pub fn add(&mut self, other: &ProxyMetric) { self.get_total += other.get_total; self.get_failed += other.get_failed; + self.get_tag_total += other.get_tag_total; + self.get_tag_failed += other.get_tag_failed; self.put_total += other.put_total; self.put_failed += other.put_failed; + self.put_tag_total += other.put_tag_total; + self.put_tag_failed += other.put_tag_failed; + self.delete_tag_total += other.delete_tag_total; + self.delete_tag_failed += other.delete_tag_failed; self.head_total += other.head_total; self.head_failed += other.head_failed; } @@ -527,18 +539,36 @@ impl ProxyStatsCache { metric.get_failed += 1; } } + "GetObjectTagging" => { + metric.get_tag_total += 1; + if is_err { + metric.get_tag_failed += 1; + } + } "PutObject" => { metric.put_total += 1; if is_err { metric.put_failed += 1; } } + "PutObjectTagging" => { + metric.put_tag_total += 1; + if is_err { + metric.put_tag_failed += 1; + } + } "HeadObject" => { metric.head_total += 1; if is_err { metric.head_failed += 1; } } + "DeleteObjectTagging" => { + metric.delete_tag_total += 1; + if is_err { + metric.delete_tag_failed += 1; + } + } _ => {} } } diff --git a/crates/ecstore/src/erasure_coding/encode.rs b/crates/ecstore/src/erasure_coding/encode.rs index e3d6817cd..8f9b029e2 100644 --- a/crates/ecstore/src/erasure_coding/encode.rs +++ b/crates/ecstore/src/erasure_coding/encode.rs @@ -26,6 +26,20 @@ use tokio::io::AsyncRead; use tokio::sync::mpsc; use tracing::error; +const ENV_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES: &str = "RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES"; +const DEFAULT_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES: usize = 32 * 1024 * 1024; +const DEFAULT_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BLOCKS: usize = 8; + +fn encode_channel_capacity(expanded_block_bytes: usize, max_inflight_bytes: usize) -> usize { + if expanded_block_bytes == 0 { + return 1; + } + + max_inflight_bytes + .saturating_div(expanded_block_bytes) + .clamp(1, DEFAULT_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BLOCKS) +} + pub(crate) struct MultiWriter<'a> { writers: &'a mut [Option], write_quorum: usize, @@ -185,7 +199,14 @@ impl Erasure { where R: AsyncRead + Send + Sync + Unpin + 'static, { - let (tx, mut rx) = mpsc::channel::>(8); + // Bound queued encoded blocks by memory budget to avoid per-request spikes. + let expanded_block_bytes = self.shard_size().saturating_mul(self.total_shard_count()); + let max_inflight_bytes = rustfs_utils::get_env_usize( + ENV_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES, + DEFAULT_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BYTES, + ); + let inflight_blocks = encode_channel_capacity(expanded_block_bytes, max_inflight_bytes); + let (tx, mut rx) = mpsc::channel::>(inflight_blocks); let task = tokio::spawn(async move { let block_size = self.block_size; @@ -310,4 +331,18 @@ mod tests { assert_eq!(written, b"small payload".len()); assert!(!committed.lock().unwrap().is_empty()); } + + #[test] + fn encode_channel_capacity_never_returns_zero() { + assert_eq!(encode_channel_capacity(0, 1024), 1); + assert_eq!(encode_channel_capacity(4096, 0), 1); + assert_eq!(encode_channel_capacity(4096, 1024), 1); + } + + #[test] + fn encode_channel_capacity_respects_budget_and_hard_cap() { + assert_eq!(encode_channel_capacity(4 * 1024 * 1024, 32 * 1024 * 1024), 8); + assert_eq!(encode_channel_capacity(16 * 1024 * 1024, 32 * 1024 * 1024), 2); + assert_eq!(encode_channel_capacity(1, usize::MAX), DEFAULT_RUSTFS_ERASURE_ENCODE_MAX_INFLIGHT_BLOCKS); + } } diff --git a/crates/iam/src/manager.rs b/crates/iam/src/manager.rs index 1a9168a1e..bdc602d60 100644 --- a/crates/iam/src/manager.rs +++ b/crates/iam/src/manager.rs @@ -36,7 +36,7 @@ use rustfs_policy::{ use rustfs_utils::{get_env_opt_str, path::path_join_buf}; use serde::{Deserialize, Serialize}; use serde_json::Value; -use std::sync::atomic::AtomicU8; +use std::sync::atomic::{AtomicU8, AtomicU64}; use std::{ collections::{HashMap, HashSet}, sync::{ @@ -93,6 +93,17 @@ pub struct IamCache { pub roles: HashMap>, pub send_chan: Sender, pub last_timestamp: AtomicI64, + pub sync_failures: AtomicU64, + pub sync_successes: AtomicU64, + pub last_sync_duration_millis: AtomicU64, +} + +#[derive(Debug, Clone, Copy, Default)] +pub struct IamSyncMetricsSnapshot { + pub last_sync_duration_millis: u64, + pub since_last_sync_millis: u64, + pub sync_failures: u64, + pub sync_successes: u64, } impl IamCache @@ -116,6 +127,9 @@ where send_chan: sender, roles: HashMap::new(), last_timestamp: AtomicI64::new(0), + sync_failures: AtomicU64::new(0), + sync_successes: AtomicU64::new(0), + last_sync_duration_millis: AtomicU64::new(0), }); sys.clone().init(receiver).await.unwrap(); @@ -200,11 +214,38 @@ where } async fn load(self: Arc) -> Result<()> { - // debug!("load iam to cache"); - self.api.load_all(&self.cache).await?; - self.last_timestamp - .store(OffsetDateTime::now_utc().unix_timestamp(), Ordering::Relaxed); - Ok(()) + let started_at = std::time::Instant::now(); + match self.api.load_all(&self.cache).await { + Ok(()) => { + self.last_timestamp + .store(OffsetDateTime::now_utc().unix_timestamp(), Ordering::Relaxed); + self.sync_successes.fetch_add(1, Ordering::Relaxed); + self.last_sync_duration_millis + .store(started_at.elapsed().as_millis().min(u128::from(u64::MAX)) as u64, Ordering::Relaxed); + Ok(()) + } + Err(err) => { + self.sync_failures.fetch_add(1, Ordering::Relaxed); + Err(err) + } + } + } + + pub fn sync_metrics_snapshot(&self) -> IamSyncMetricsSnapshot { + let now_secs = OffsetDateTime::now_utc().unix_timestamp(); + let last_sync_secs = self.last_timestamp.load(Ordering::Relaxed); + let since_last_sync_millis = if last_sync_secs > 0 && now_secs >= last_sync_secs { + ((now_secs - last_sync_secs) as u64).saturating_mul(1000) + } else { + 0 + }; + + IamSyncMetricsSnapshot { + last_sync_duration_millis: self.last_sync_duration_millis.load(Ordering::Relaxed), + since_last_sync_millis, + sync_failures: self.sync_failures.load(Ordering::Relaxed), + sync_successes: self.sync_successes.load(Ordering::Relaxed), + } } pub async fn load_user(&self, access_key: &str) -> Result<()> { diff --git a/crates/iam/src/oidc.rs b/crates/iam/src/oidc.rs index 558f0ea7f..d775ed6ba 100644 --- a/crates/iam/src/oidc.rs +++ b/crates/iam/src/oidc.rs @@ -31,11 +31,11 @@ use rustfs_ecstore::config::{Config as ServerConfig, KVS, get_global_server_conf use rustfs_policy::policy::{ClaimLookup, get_claim_case_insensitive}; use serde::{Deserialize, Serialize}; use std::borrow::Cow; -use std::collections::HashMap; +use std::collections::{HashMap, VecDeque}; use std::future::Future; use std::net::IpAddr; use std::pin::Pin; -use std::sync::RwLock; +use std::sync::{LazyLock, Mutex, MutexGuard, RwLock}; use std::time::{Duration as StdDuration, Instant}; use tokio::time::sleep; use tracing::{error, info, warn}; @@ -44,6 +44,127 @@ use url::Url; const OIDC_JWKS_REFRESH_INTERVAL: StdDuration = StdDuration::from_secs(24 * 60 * 60); const OIDC_DISCOVERY_TRANSPORT_RETRIES: usize = 3; const OIDC_DISCOVERY_TRANSPORT_RETRY_DELAY: StdDuration = StdDuration::from_millis(50); +const OIDC_PLUGIN_AUTHN_WINDOW: StdDuration = StdDuration::from_secs(60); + +#[derive(Debug, Clone, Copy, Default)] +pub struct OidcPluginAuthnMetricsSnapshot { + pub failed_requests_minute: u64, + pub last_fail_seconds: u64, + pub last_succ_seconds: u64, + pub succ_avg_rtt_ms_minute: u64, + pub succ_max_rtt_ms_minute: u64, + pub total_requests_minute: u64, +} + +#[derive(Debug, Clone)] +struct OidcPluginAuthnSample { + observed_at: Instant, + succeeded: bool, + rtt_ms: u64, +} + +#[derive(Debug, Default)] +struct OidcPluginAuthnMetrics { + samples: Mutex>, + last_fail_at: Mutex>, + last_succ_at: Mutex>, +} + +fn lock_oidc_plugin_authn_metrics<'a, T>(mutex: &'a Mutex, metric: &'static str) -> MutexGuard<'a, T> { + match mutex.lock() { + Ok(guard) => guard, + Err(err) => { + warn!("recovering poisoned OIDC plugin authn metrics lock: {}", metric); + err.into_inner() + } + } +} + +fn seconds_since(now: Instant, observed_at: Option) -> u64 { + observed_at + .map(|instant| now.duration_since(instant).as_secs()) + .unwrap_or_default() +} + +impl OidcPluginAuthnMetrics { + fn record(&self, rtt_ms: u64, succeeded: bool) { + let now = Instant::now(); + let mut samples = lock_oidc_plugin_authn_metrics(&self.samples, "samples"); + samples.push_back(OidcPluginAuthnSample { + observed_at: now, + succeeded, + rtt_ms, + }); + while samples + .front() + .is_some_and(|sample| now.duration_since(sample.observed_at) > OIDC_PLUGIN_AUTHN_WINDOW) + { + samples.pop_front(); + } + drop(samples); + + if succeeded { + *lock_oidc_plugin_authn_metrics(&self.last_succ_at, "last_succ_at") = Some(now); + } else { + *lock_oidc_plugin_authn_metrics(&self.last_fail_at, "last_fail_at") = Some(now); + } + } + + fn snapshot(&self) -> OidcPluginAuthnMetricsSnapshot { + let now = Instant::now(); + let (total_requests_minute, failed_requests_minute, succ_avg_rtt_ms_minute, succ_max_rtt_ms_minute) = { + let mut samples = lock_oidc_plugin_authn_metrics(&self.samples, "samples"); + while samples + .front() + .is_some_and(|sample| now.duration_since(sample.observed_at) > OIDC_PLUGIN_AUTHN_WINDOW) + { + samples.pop_front(); + } + + let mut failed_requests_minute = 0u64; + let mut successful_requests = 0u64; + let mut successful_rtt_sum = 0u64; + let mut succ_max_rtt_ms_minute = 0u64; + + for sample in samples.iter() { + if sample.succeeded { + successful_requests += 1; + successful_rtt_sum += sample.rtt_ms; + succ_max_rtt_ms_minute = succ_max_rtt_ms_minute.max(sample.rtt_ms); + } else { + failed_requests_minute += 1; + } + } + + let succ_avg_rtt_ms_minute = successful_rtt_sum.checked_div(successful_requests).unwrap_or_default(); + + ( + samples.len() as u64, + failed_requests_minute, + succ_avg_rtt_ms_minute, + succ_max_rtt_ms_minute, + ) + }; + + let last_fail_seconds = seconds_since(now, *lock_oidc_plugin_authn_metrics(&self.last_fail_at, "last_fail_at")); + let last_succ_seconds = seconds_since(now, *lock_oidc_plugin_authn_metrics(&self.last_succ_at, "last_succ_at")); + + OidcPluginAuthnMetricsSnapshot { + failed_requests_minute, + last_fail_seconds, + last_succ_seconds, + succ_avg_rtt_ms_minute, + succ_max_rtt_ms_minute, + total_requests_minute, + } + } +} + +static OIDC_PLUGIN_AUTHN_METRICS: LazyLock = LazyLock::new(OidcPluginAuthnMetrics::default); + +pub fn oidc_plugin_authn_metrics_snapshot() -> OidcPluginAuthnMetricsSnapshot { + OIDC_PLUGIN_AUTHN_METRICS.snapshot() +} // ---- HTTP Client Adapter ---- @@ -120,6 +241,7 @@ impl<'c> AsyncHttpClient<'c> for ReqwestHttpClient { fn call(&'c self, request: http::Request>) -> Self::Future { Box::pin(async move { + let started_at = Instant::now(); let (parts, body) = request.into_parts(); let uri = parts.uri.to_string(); let client = self.client_for_uri(&uri); @@ -128,8 +250,13 @@ impl<'c> AsyncHttpClient<'c> for ReqwestHttpClient { .headers(parts.headers) .body(body) .send() - .await - .map_err(OidcHttpError::Reqwest)?; + .await; + + let elapsed_ms = started_at.elapsed().as_millis().min(u128::from(u64::MAX)) as u64; + let succeeded = response.as_ref().is_ok_and(|resp| resp.status().is_success()); + OIDC_PLUGIN_AUTHN_METRICS.record(elapsed_ms, succeeded); + + let response = response.map_err(OidcHttpError::Reqwest)?; let status = response.status(); let headers = response.headers().clone(); diff --git a/crates/iam/src/sys.rs b/crates/iam/src/sys.rs index bd5604f8f..f50b3821f 100644 --- a/crates/iam/src/sys.rs +++ b/crates/iam/src/sys.rs @@ -16,9 +16,9 @@ use crate::error::Error as IamError; use crate::error::is_err_no_such_account; use crate::error::is_err_no_such_temp_account; use crate::error::{Error, Result}; -use crate::manager::IamCache; use crate::manager::extract_jwt_claims; use crate::manager::get_default_policyes; +use crate::manager::{IamCache, IamSyncMetricsSnapshot}; use crate::store::GroupInfo; use crate::store::MappedPolicy; use crate::store::Store; @@ -186,6 +186,10 @@ impl IamSys { self.store.api.has_watcher() } + pub fn sync_metrics_snapshot(&self) -> IamSyncMetricsSnapshot { + self.store.sync_metrics_snapshot() + } + pub async fn set_policy_plugin_client(client: rustfs_policy::policy::opa::AuthZPlugin) { let policy_plugin_client = get_policy_plugin_client(); let mut guard = policy_plugin_client.write().await; diff --git a/crates/io-metrics/src/adaptive_ttl.rs b/crates/io-metrics/src/adaptive_ttl.rs index 50ddc7ebd..f544515eb 100644 --- a/crates/io-metrics/src/adaptive_ttl.rs +++ b/crates/io-metrics/src/adaptive_ttl.rs @@ -31,14 +31,14 @@ use std::time::{Duration, Instant}; pub fn record_ttl_adjustment(_key: &str, base_ttl: u64, adjusted_ttl: u64) { use metrics::{counter, gauge}; - counter!("rustfs.cache.ttl.adjustments").increment(1); - gauge!("rustfs.cache.ttl.base").set(base_ttl as f64); - gauge!("rustfs.cache.ttl.adjusted").set(adjusted_ttl as f64); + counter!("rustfs_cache_ttl_adjustments").increment(1); + gauge!("rustfs_cache_ttl_base").set(base_ttl as f64); + gauge!("rustfs_cache_ttl_adjusted").set(adjusted_ttl as f64); if adjusted_ttl > base_ttl { - counter!("rustfs.cache.ttl.extensions").increment(1); + counter!("rustfs_cache_ttl_extensions").increment(1); } else if adjusted_ttl < base_ttl { - counter!("rustfs.cache.ttl.reductions").increment(1); + counter!("rustfs_cache_ttl_reductions").increment(1); } } @@ -46,7 +46,7 @@ pub fn record_ttl_adjustment(_key: &str, base_ttl: u64, adjusted_ttl: u64) { #[inline(always)] pub fn record_ttl_expiration() { use metrics::counter; - counter!("rustfs.cache.ttl.expirations").increment(1); + counter!("rustfs_cache_ttl_expirations").increment(1); } /// Record early eviction. @@ -57,7 +57,7 @@ pub fn record_ttl_expiration() { #[inline(always)] pub fn record_early_eviction(reason: &str) { use metrics::counter; - counter!("rustfs.cache.evictions.early", "reason" => reason.to_string()).increment(1); + counter!("rustfs_cache_evictions_early", "reason" => reason.to_string()).increment(1); } /// Record access pattern change. @@ -69,7 +69,7 @@ pub fn record_early_eviction(reason: &str) { #[inline(always)] pub fn record_access_pattern_change(from: &str, to: &str) { use metrics::counter; - counter!("rustfs.cache.access_pattern.changes", "from" => from.to_string(), "to" => to.to_string()).increment(1); + counter!("rustfs_cache_access_pattern_changes", "from" => from.to_string(), "to" => to.to_string()).increment(1); } /// Adaptive TTL statistics. diff --git a/crates/io-metrics/src/backpressure_metrics.rs b/crates/io-metrics/src/backpressure_metrics.rs index 391fbcc65..f519ee846 100644 --- a/crates/io-metrics/src/backpressure_metrics.rs +++ b/crates/io-metrics/src/backpressure_metrics.rs @@ -18,35 +18,35 @@ #[inline(always)] pub fn record_backpressure_state_change(from: &str, to: &str) { use metrics::counter; - counter!("rustfs.backpressure.state.changes", "from" => from.to_string(), "to" => to.to_string()).increment(1); + counter!("rustfs_backpressure_state_changes", "from" => from.to_string(), "to" => to.to_string()).increment(1); } /// Record backpressure rejection. #[inline(always)] pub fn record_backpressure_rejection() { use metrics::counter; - counter!("rustfs.backpressure.rejections").increment(1); + counter!("rustfs_backpressure_rejections").increment(1); } /// Record concurrent operations count. #[inline(always)] pub fn record_concurrent_operations(count: usize) { use metrics::gauge; - gauge!("rustfs.backpressure.concurrent").set(count as f64); + gauge!("rustfs_backpressure_concurrent").set(count as f64); } /// Record backpressure activation. #[inline(always)] pub fn record_backpressure_activation() { use metrics::counter; - counter!("rustfs.backpressure.activations").increment(1); + counter!("rustfs_backpressure_activations").increment(1); } /// Record backpressure deactivation. #[inline(always)] pub fn record_backpressure_deactivation() { use metrics::counter; - counter!("rustfs.backpressure.deactivations").increment(1); + counter!("rustfs_backpressure_deactivations").increment(1); } #[cfg(test)] diff --git a/crates/io-metrics/src/capacity_metrics.rs b/crates/io-metrics/src/capacity_metrics.rs index be659c1dc..f76ea85e7 100644 --- a/crates/io-metrics/src/capacity_metrics.rs +++ b/crates/io-metrics/src/capacity_metrics.rs @@ -20,35 +20,35 @@ use std::time::Duration; /// Record capacity cache hit. #[inline(always)] pub fn record_capacity_cache_hit() { - counter!("rustfs.capacity.cache.hits").increment(1); + counter!("rustfs_capacity_cache_hits").increment(1); } /// Record capacity cache miss. #[inline(always)] pub fn record_capacity_cache_miss() { - counter!("rustfs.capacity.cache.misses").increment(1); + counter!("rustfs_capacity_cache_misses").increment(1); } /// Record how capacity cache was served to the caller. #[inline(always)] pub fn record_capacity_cache_served(state: &'static str) { - counter!("rustfs.capacity.cache.served.total", "state" => state).increment(1); + counter!("rustfs_capacity_cache_served_total", "state" => state).increment(1); } /// Record current capacity gauge. #[inline(always)] pub fn record_capacity_current_bytes(used_bytes: u64) { - gauge!("rustfs.capacity.current").set(used_bytes as f64); + gauge!("rustfs_capacity_current_bytes").set(used_bytes as f64); } /// Record capacity update completion. #[inline(always)] pub fn record_capacity_update_completed(source: &'static str, duration: Duration, used_bytes: u64, is_estimated: bool) { - counter!("rustfs.capacity.update.total", "source" => source).increment(1); - histogram!("rustfs.capacity.update.duration.seconds", "source" => source).record(duration.as_secs_f64()); - histogram!("rustfs.capacity.update.bytes", "source" => source).record(used_bytes as f64); + counter!("rustfs_capacity_update_total", "source" => source).increment(1); + histogram!("rustfs_capacity_update_duration_seconds", "source" => source).record(duration.as_secs_f64()); + histogram!("rustfs_capacity_update_bytes", "source" => source).record(used_bytes as f64); counter!( - "rustfs.capacity.update.estimated.total", + "rustfs_capacity_update_estimated_total", "source" => source, "estimated" => if is_estimated { "true" } else { "false" } ) @@ -58,86 +58,86 @@ pub fn record_capacity_update_completed(source: &'static str, duration: Duration /// Record failed capacity update. #[inline(always)] pub fn record_capacity_update_failed(source: &'static str) { - counter!("rustfs.capacity.update.failures", "source" => source).increment(1); + counter!("rustfs_capacity_update_failures", "source" => source).increment(1); } /// Record a capacity refresh request. #[inline(always)] pub fn record_capacity_refresh_request(mode: &'static str, source: &'static str) { - counter!("rustfs.capacity.refresh.requests.total", "mode" => mode, "source" => source).increment(1); + counter!("rustfs_capacity_refresh_requests_total", "mode" => mode, "source" => source).increment(1); } /// Record a refresh joiner waiting for an inflight refresh. #[inline(always)] pub fn record_capacity_refresh_joiner(source: &'static str) { - counter!("rustfs.capacity.refresh.joiners.total", "source" => source).increment(1); + counter!("rustfs_capacity_refresh_joiners_total", "source" => source).increment(1); } /// Record the number of inflight capacity refreshes. #[inline(always)] pub fn record_capacity_refresh_inflight(count: usize) { - gauge!("rustfs.capacity.refresh.inflight").set(count as f64); + gauge!("rustfs_capacity_refresh_inflight").set(count as f64); } /// Record the final result of a capacity refresh. #[inline(always)] pub fn record_capacity_refresh_result(source: &'static str, result: &'static str, duration: Duration) { - counter!("rustfs.capacity.refresh.result.total", "source" => source, "result" => result).increment(1); - histogram!("rustfs.capacity.refresh.duration.seconds", "source" => source, "result" => result).record(duration.as_secs_f64()); + counter!("rustfs_capacity_refresh_result_total", "source" => source, "result" => result).increment(1); + histogram!("rustfs_capacity_refresh_duration_seconds", "source" => source, "result" => result).record(duration.as_secs_f64()); } /// Record the refresh scope selected for a capacity refresh. #[inline(always)] pub fn record_capacity_refresh_scope(scope: &'static str, disk_count: usize) { - counter!("rustfs.capacity.refresh.scope.total", "scope" => scope).increment(1); - histogram!("rustfs.capacity.refresh.scope.disks", "scope" => scope).record(disk_count as f64); + counter!("rustfs_capacity_refresh_scope_total", "scope" => scope).increment(1); + histogram!("rustfs_capacity_refresh_scope_disks", "scope" => scope).record(disk_count as f64); } /// Record the current number of dirty disks tracked by capacity management. #[inline(always)] pub fn record_capacity_dirty_disk_count(count: usize) { - gauge!("rustfs.capacity.dirty.disks").set(count as f64); + gauge!("rustfs_capacity_dirty_disks").set(count as f64); } /// Record capacity write activity. #[inline(always)] pub fn record_capacity_write_operation(write_frequency: usize) { - counter!("rustfs.capacity.write.operations").increment(1); - gauge!("rustfs.capacity.write.frequency").set(write_frequency as f64); + counter!("rustfs_capacity_write_operations").increment(1); + gauge!("rustfs_capacity_write_frequency").set(write_frequency as f64); } /// Record symlink accounting. #[inline(always)] pub fn record_capacity_symlink(size_bytes: u64) { - counter!("rustfs.capacity.symlinks.encountered").increment(1); - histogram!("rustfs.capacity.symlinks.size.bytes").record(size_bytes as f64); + counter!("rustfs_capacity_symlinks_encountered").increment(1); + histogram!("rustfs_capacity_symlinks_size_bytes").record(size_bytes as f64); } /// Record timeout fallback event. #[inline(always)] pub fn record_capacity_timeout_fallback() { - counter!("rustfs.capacity.timeout.fallback").increment(1); + counter!("rustfs_capacity_timeout_fallback").increment(1); } /// Record stall detection event. #[inline(always)] pub fn record_capacity_stall_detected() { - counter!("rustfs.capacity.timeout.stall").increment(1); + counter!("rustfs_capacity_timeout_stall").increment(1); } /// Record dynamic timeout usage. #[inline(always)] pub fn record_capacity_dynamic_timeout(timeout: Duration) { - counter!("rustfs.capacity.timeout.dynamic").increment(1); - histogram!("rustfs.capacity.timeout.dynamic.seconds").record(timeout.as_secs_f64()); + counter!("rustfs_capacity_timeout_dynamic").increment(1); + histogram!("rustfs_capacity_timeout_dynamic_seconds").record(timeout.as_secs_f64()); } /// Record scan sampling outcome. #[inline(always)] pub fn record_capacity_scan_sampling(sampled_count: usize, estimated: bool) { - histogram!("rustfs.capacity.scan.sampled.count").record(sampled_count as f64); + histogram!("rustfs_capacity_scan_sampled_count").record(sampled_count as f64); counter!( - "rustfs.capacity.scan.estimated.total", + "rustfs_capacity_scan_estimated_total", "estimated" => if estimated { "true" } else { "false" } ) .increment(1); @@ -146,7 +146,7 @@ pub fn record_capacity_scan_sampling(sampled_count: usize, estimated: bool) { /// Record the scan mode used for a capacity result. #[inline(always)] pub fn record_capacity_scan_mode(mode: &'static str) { - counter!("rustfs.capacity.scan.mode.total", "mode" => mode).increment(1); + counter!("rustfs_capacity_scan_mode_total", "mode" => mode).increment(1); } /// Record per-disk capacity scan statistics. @@ -159,16 +159,16 @@ pub fn record_capacity_scan_disk( estimated: bool, partial_errors: bool, ) { - histogram!("rustfs.capacity.scan.disk.duration.seconds", "disk" => disk.to_owned()).record(duration.as_secs_f64()); - histogram!("rustfs.capacity.scan.disk.files", "disk" => disk.to_owned()).record(file_count as f64); - histogram!("rustfs.capacity.scan.disk.sampled", "disk" => disk.to_owned()).record(sampled_count as f64); + histogram!("rustfs_capacity_scan_disk_duration_seconds", "disk" => disk.to_owned()).record(duration.as_secs_f64()); + histogram!("rustfs_capacity_scan_disk_files", "disk" => disk.to_owned()).record(file_count as f64); + histogram!("rustfs_capacity_scan_disk_sampled", "disk" => disk.to_owned()).record(sampled_count as f64); counter!( - "rustfs.capacity.scan.disk.estimated.total", + "rustfs_capacity_scan_disk_estimated_total", "disk" => disk.to_owned(), "estimated" => if estimated { "true" } else { "false" } ) .increment(1); if partial_errors { - counter!("rustfs.capacity.scan.disk.partial_errors.total", "disk" => disk.to_owned()).increment(1); + counter!("rustfs_capacity_scan_disk_partial_errors_total", "disk" => disk.to_owned()).increment(1); } } diff --git a/crates/io-metrics/src/deadlock_metrics.rs b/crates/io-metrics/src/deadlock_metrics.rs index 7d85f80e4..b79d08227 100644 --- a/crates/io-metrics/src/deadlock_metrics.rs +++ b/crates/io-metrics/src/deadlock_metrics.rs @@ -20,52 +20,52 @@ use std::time::Duration; #[inline(always)] pub fn record_deadlock_detected(cycle_length: usize) { use metrics::{counter, histogram}; - counter!("rustfs.deadlock.detected").increment(1); - histogram!("rustfs.deadlock.cycle_length").record(cycle_length as f64); + counter!("rustfs_deadlock_detected_total").increment(1); + histogram!("rustfs_deadlock_cycle_length").record(cycle_length as f64); } /// Record long-held lock. #[inline(always)] pub fn record_long_held_lock(_lock_id: u64, hold_time: Duration) { use metrics::{counter, histogram}; - counter!("rustfs.deadlock.long_held").increment(1); - histogram!("rustfs.deadlock.hold_time.secs").record(hold_time.as_secs_f64()); + counter!("rustfs_deadlock_long_held").increment(1); + histogram!("rustfs_deadlock_hold_time_secs").record(hold_time.as_secs_f64()); } /// Record lock acquisition. #[inline(always)] pub fn record_lock_acquisition(lock_type: &str) { use metrics::counter; - counter!("rustfs.lock.acquisitions", "type" => lock_type.to_string()).increment(1); + counter!("rustfs_lock_acquisitions", "type" => lock_type.to_string()).increment(1); } /// Record lock release. #[inline(always)] pub fn record_lock_release(lock_type: &str, hold_time: Duration) { use metrics::{counter, histogram}; - counter!("rustfs.lock.releases", "type" => lock_type.to_string()).increment(1); - histogram!("rustfs.lock.hold_time.secs", "type" => lock_type.to_string()).record(hold_time.as_secs_f64()); + counter!("rustfs_lock_releases", "type" => lock_type.to_string()).increment(1); + histogram!("rustfs_lock_hold_time_secs", "type" => lock_type.to_string()).record(hold_time.as_secs_f64()); } /// Record lock contention. #[inline(always)] pub fn record_lock_contention(lock_type: &str) { use metrics::counter; - counter!("rustfs.lock.contentions", "type" => lock_type.to_string()).increment(1); + counter!("rustfs_lock_contentions", "type" => lock_type.to_string()).increment(1); } /// Record wait graph edge added. #[inline(always)] pub fn record_wait_edge_added() { use metrics::counter; - counter!("rustfs.deadlock.wait_edges.added").increment(1); + counter!("rustfs_deadlock_wait_edges_added").increment(1); } /// Record wait graph edge removed. #[inline(always)] pub fn record_wait_edge_removed() { use metrics::counter; - counter!("rustfs.deadlock.wait_edges.removed").increment(1); + counter!("rustfs_deadlock_wait_edges_removed").increment(1); } #[cfg(test)] diff --git a/crates/io-metrics/src/io_metrics.rs b/crates/io-metrics/src/io_metrics.rs index 6ef99fbe0..54ed9561b 100644 --- a/crates/io-metrics/src/io_metrics.rs +++ b/crates/io-metrics/src/io_metrics.rs @@ -27,11 +27,11 @@ pub fn record_io_scheduler_decision(buffer_size: usize, load_level: &str, strategy: &str) { use metrics::{counter, gauge, histogram}; - counter!("rustfs.io.scheduler.decisions").increment(1); - gauge!("rustfs.io.scheduler.buffer_size").set(buffer_size as f64); - counter!("rustfs.io.scheduler.load", "level" => load_level.to_string()).increment(1); - counter!("rustfs.io.scheduler.strategy", "type" => strategy.to_string()).increment(1); - histogram!("rustfs.io.scheduler.buffer_size.histogram").record(buffer_size as f64); + counter!("rustfs_io_scheduler_decisions").increment(1); + gauge!("rustfs_io_scheduler_buffer_size").set(buffer_size as f64); + counter!("rustfs_io_scheduler_load", "level" => load_level.to_string()).increment(1); + counter!("rustfs_io_scheduler_strategy", "type" => strategy.to_string()).increment(1); + histogram!("rustfs_io_scheduler_buffer_size_histogram").record(buffer_size as f64); } /// Record I/O priority decision. @@ -44,9 +44,9 @@ pub fn record_io_scheduler_decision(buffer_size: usize, load_level: &str, strate pub fn record_io_priority_decision(priority: &str, size: usize) { use metrics::{counter, histogram}; - counter!("rustfs.io.priority.decisions").increment(1); - counter!("rustfs.io.priority.by_level", "priority" => priority.to_string()).increment(1); - histogram!("rustfs.io.priority.request_size").record(size as f64); + counter!("rustfs_io_priority_decisions").increment(1); + counter!("rustfs_io_priority_by_level", "priority" => priority.to_string()).increment(1); + histogram!("rustfs_io_priority_request_size").record(size as f64); } /// Record load level change. @@ -58,7 +58,7 @@ pub fn record_io_priority_decision(priority: &str, size: usize) { #[inline(always)] pub fn record_load_level_change(from: &str, to: &str) { use metrics::counter; - counter!("rustfs.io.load.changes", "from" => from.to_string(), "to" => to.to_string()).increment(1); + counter!("rustfs_io_load_changes", "from" => from.to_string(), "to" => to.to_string()).increment(1); } /// Record bandwidth observation. @@ -69,8 +69,8 @@ pub fn record_load_level_change(from: &str, to: &str) { #[inline(always)] pub fn record_bandwidth_observation(bps: u64) { use metrics::{gauge, histogram}; - gauge!("rustfs.io.bandwidth.bps").set(bps as f64); - histogram!("rustfs.io.bandwidth.histogram").record(bps as f64); + gauge!("rustfs_io_bandwidth_bps").set(bps as f64); + histogram!("rustfs_io_bandwidth_histogram").record(bps as f64); } /// Record buffer size adjustment. @@ -83,9 +83,9 @@ pub fn record_bandwidth_observation(bps: u64) { #[inline(always)] pub fn record_buffer_size_adjustment(original: usize, adjusted: usize, reason: &str) { use metrics::{counter, gauge}; - counter!("rustfs.io.buffer.adjustments", "reason" => reason.to_string()).increment(1); - gauge!("rustfs.io.buffer.original").set(original as f64); - gauge!("rustfs.io.buffer.adjusted").set(adjusted as f64); + counter!("rustfs_io_buffer_adjustments", "reason" => reason.to_string()).increment(1); + gauge!("rustfs_io_buffer_original").set(original as f64); + gauge!("rustfs_io_buffer_adjusted").set(adjusted as f64); } /// Record queue operation. @@ -98,8 +98,8 @@ pub fn record_buffer_size_adjustment(original: usize, adjusted: usize, reason: & #[inline(always)] pub fn record_queue_operation(operation: &str, priority: &str, queue_size: usize) { use metrics::{counter, gauge}; - counter!("rustfs.io.queue.operations", "operation" => operation.to_string(), "priority" => priority.to_string()).increment(1); - gauge!("rustfs.io.queue.size", "priority" => priority.to_string()).set(queue_size as f64); + counter!("rustfs_io_queue_operations", "operation" => operation.to_string(), "priority" => priority.to_string()).increment(1); + gauge!("rustfs_io_queue_size", "priority" => priority.to_string()).set(queue_size as f64); } /// Record starvation event. @@ -110,7 +110,7 @@ pub fn record_queue_operation(operation: &str, priority: &str, queue_size: usize #[inline(always)] pub fn record_starvation_event(priority: &str) { use metrics::counter; - counter!("rustfs.io.starvation.events", "priority" => priority.to_string()).increment(1); + counter!("rustfs_io_starvation_events", "priority" => priority.to_string()).increment(1); } /// I/O scheduler statistics. diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs index 1c2c5a636..dca5c51e0 100644 --- a/crates/io-metrics/src/lib.rs +++ b/crates/io-metrics/src/lib.rs @@ -217,9 +217,9 @@ pub fn record_get_object_io_state( /// * `duration_ms` - Time taken for the read operation in milliseconds #[inline(always)] pub fn record_zero_copy_read(size_bytes: usize, duration_ms: f64) { - counter!("rustfs.zero_copy.reads.total").increment(1); - histogram!("rustfs.zero_copy.read.size.bytes").record(size_bytes as f64); - histogram!("rustfs.zero_copy.read.duration.ms").record(duration_ms); + counter!("rustfs_zero_copy_reads_total").increment(1); + histogram!("rustfs_zero_copy_read_size_bytes").record(size_bytes as f64); + histogram!("rustfs_zero_copy_read_duration_ms").record(duration_ms); } /// Record memory copies avoided by using zero-copy. @@ -229,7 +229,7 @@ pub fn record_zero_copy_read(size_bytes: usize, duration_ms: f64) { /// * `bytes_saved` - Number of bytes that would have been copied without zero-copy #[inline(always)] pub fn record_memory_copy_saved(bytes_saved: usize) { - counter!("rustfs.zero_copy.memory.saved.bytes").increment(bytes_saved as u64); + counter!("rustfs_zero_copy_memory_saved_bytes_total").increment(bytes_saved as u64); } /// Record a fallback from zero-copy to regular read. @@ -242,7 +242,7 @@ pub fn record_memory_copy_saved(bytes_saved: usize) { /// * `reason` - Reason for the fallback (e.g., "mmap_unavailable", "file_too_large") #[inline(always)] pub fn record_zero_copy_fallback(reason: &str) { - counter!("rustfs.zero_copy.fallback.total", "reason" => reason.to_string()).increment(1); + counter!("rustfs_zero_copy_fallback_total", "reason" => reason.to_string()).increment(1); } // ============================================================================ @@ -258,13 +258,13 @@ pub fn record_zero_copy_fallback(reason: &str) { /// * `from_pool` - Whether buffer was reused from pool #[inline(always)] pub fn record_bytes_pool_acquire(tier: &str, size: usize, from_pool: bool) { - counter!("rustfs.bytes.pool.acquisitions.total", "tier" => tier.to_string()).increment(1); - gauge!("rustfs.bytes.pool.size.bytes", "tier" => tier.to_string()).set(size as f64); + counter!("rustfs_bytes_pool_acquisitions_total", "tier" => tier.to_string()).increment(1); + gauge!("rustfs_bytes_pool_size_bytes", "tier" => tier.to_string()).set(size as f64); if from_pool { - counter!("rustfs.bytes.pool.hits.total", "tier" => tier.to_string()).increment(1); + counter!("rustfs_bytes_pool_hits_total", "tier" => tier.to_string()).increment(1); } else { - counter!("rustfs.bytes.pool.misses.total", "tier" => tier.to_string()).increment(1); + counter!("rustfs_bytes_pool_misses_total", "tier" => tier.to_string()).increment(1); } } @@ -275,7 +275,7 @@ pub fn record_bytes_pool_acquire(tier: &str, size: usize, from_pool: bool) { /// * `tier` - Pool tier ("small", "medium", "large", "xlarge") #[inline(always)] pub fn record_bytes_pool_return(tier: &str) { - counter!("rustfs.bytes.pool.returns.total", "tier" => tier.to_string()).increment(1); + counter!("rustfs_bytes_pool_returns_total", "tier" => tier.to_string()).increment(1); } /// Record current BytesPool allocated bytes. @@ -286,7 +286,7 @@ pub fn record_bytes_pool_return(tier: &str) { /// * `bytes` - Currently allocated bytes #[inline(always)] pub fn record_bytes_pool_allocated(tier: &str, bytes: u64) { - gauge!("rustfs.bytes.pool.allocated.bytes", "tier" => tier.to_string()).set(bytes as f64); + gauge!("rustfs_bytes_pool_allocated_bytes", "tier" => tier.to_string()).set(bytes as f64); } /// Get BytesPool hit rate as a gauge metric. @@ -297,7 +297,7 @@ pub fn record_bytes_pool_allocated(tier: &str, bytes: u64) { /// * `hit_rate` - Hit rate (0.0 - 1.0) #[inline(always)] pub fn record_bytes_pool_hit_rate(tier: &str, hit_rate: f64) { - gauge!("rustfs.bytes.pool.hit.rate", "tier" => tier.to_string()).set(hit_rate * 100.0); + gauge!("rustfs_bytes_pool_hit_rate", "tier" => tier.to_string()).set(hit_rate * 100.0); } /// Record zero-copy write operation. @@ -308,9 +308,9 @@ pub fn record_bytes_pool_hit_rate(tier: &str, hit_rate: f64) { /// * `duration_ms` - Time taken for the write operation in milliseconds #[inline(always)] pub fn record_zero_copy_write(size_bytes: usize, duration_ms: f64) { - counter!("rustfs.zero_copy.write.total").increment(1); - histogram!("rustfs.zero_copy.write.size.bytes").record(size_bytes as f64); - histogram!("rustfs.zero_copy.write.duration.ms").record(duration_ms); + counter!("rustfs_zero_copy_write_total").increment(1); + histogram!("rustfs_zero_copy_write_size_bytes").record(size_bytes as f64); + histogram!("rustfs_zero_copy_write_duration_ms").record(duration_ms); } /// Record zero-copy write fallback. @@ -322,7 +322,7 @@ pub fn record_zero_copy_write(size_bytes: usize, duration_ms: f64) { /// * `reason` - Reason for the fallback #[inline(always)] pub fn record_zero_copy_write_fallback(reason: &str) { - counter!("rustfs.zero_copy.write.fallback.total", "reason" => reason.to_string()).increment(1); + counter!("rustfs_zero_copy_write_fallback_total", "reason" => reason.to_string()).increment(1); } /// Record bytes saved from zero-copy. @@ -332,7 +332,7 @@ pub fn record_zero_copy_write_fallback(reason: &str) { /// * `size_bytes` - Number of bytes saved from zero-copy #[inline(always)] pub fn record_bytes_saved(size_bytes: usize) { - counter!("rustfs.zero_copy.bytes.saved.total").increment(size_bytes as u64); + counter!("rustfs_zero_copy_bytes_saved_total").increment(size_bytes as u64); } // ============================================================================ @@ -350,11 +350,11 @@ pub fn record_bytes_saved(size_bytes: usize) { /// interpreted as the definitive source of truth for data-plane copy mode. #[inline(always)] pub fn record_get_object(duration_ms: f64, size_bytes: i64) { - counter!("rustfs.s3.get_object.total").increment(1); - histogram!("rustfs.s3.get_object.duration.ms").record(duration_ms); + counter!("rustfs_s3_get_object_total").increment(1); + histogram!("rustfs_s3_get_object_duration_ms").record(duration_ms); if size_bytes > 0 { - histogram!("rustfs.s3.get_object.size.bytes").record(size_bytes as f64); + histogram!("rustfs_s3_get_object_size_bytes").record(size_bytes as f64); } } @@ -367,15 +367,15 @@ pub fn record_get_object(duration_ms: f64, size_bytes: i64) { /// * `zero_copy_enabled` - Whether zero-copy was enabled for this operation #[inline(always)] pub fn record_put_object(duration_ms: f64, size_bytes: i64, zero_copy_enabled: bool) { - counter!("rustfs.s3.put_object.total").increment(1); - histogram!("rustfs.s3.put_object.duration.ms").record(duration_ms); + counter!("rustfs_s3_put_object_total").increment(1); + histogram!("rustfs_s3_put_object_duration_ms").record(duration_ms); if size_bytes > 0 { - histogram!("rustfs.s3.put_object.size.bytes").record(size_bytes as f64); + histogram!("rustfs_s3_put_object_size_bytes").record(size_bytes as f64); } if zero_copy_enabled { - counter!("rustfs.s3.put_object.zero_copy.enabled.total").increment(1); + counter!("rustfs_s3_put_object_zero_copy_enabled_total").increment(1); } } @@ -388,12 +388,12 @@ pub fn record_put_object(duration_ms: f64, size_bytes: i64, zero_copy_enabled: b /// * `is_truncated` - Whether the response was truncated #[inline(always)] pub fn record_list_objects(duration_ms: f64, objects_count: u64, is_truncated: bool) { - counter!("rustfs.s3.list_objects.total").increment(1); - histogram!("rustfs.s3.list_objects.duration.ms").record(duration_ms); - histogram!("rustfs.s3.list_objects.count").record(objects_count as f64); + counter!("rustfs_s3_list_objects_total").increment(1); + histogram!("rustfs_s3_list_objects_duration_ms").record(duration_ms); + histogram!("rustfs_s3_list_objects_count").record(objects_count as f64); if is_truncated { - counter!("rustfs.s3.list_objects.truncated.total").increment(1); + counter!("rustfs_s3_list_objects_truncated_total").increment(1); } } @@ -405,11 +405,11 @@ pub fn record_list_objects(duration_ms: f64, objects_count: u64, is_truncated: b /// * `version_deleted` - Whether a specific version was deleted #[inline(always)] pub fn record_delete_object(duration_ms: f64, version_deleted: bool) { - counter!("rustfs.s3.delete_object.total").increment(1); - histogram!("rustfs.s3.delete_object.duration.ms").record(duration_ms); + counter!("rustfs_s3_delete_object_total").increment(1); + histogram!("rustfs_s3_delete_object_duration_ms").record(duration_ms); if version_deleted { - counter!("rustfs.s3.delete_object.version.total").increment(1); + counter!("rustfs_s3_delete_object_version_total").increment(1); } } @@ -427,18 +427,18 @@ pub fn record_delete_object(duration_ms: f64, version_deleted: bool) { /// * `concurrent_requests` - Number of concurrent requests #[inline(always)] pub fn record_io_strategy(storage_media: &str, access_pattern: &str, buffer_size: usize, concurrent_requests: u64) { - counter!("rustfs.io.strategy.total", + counter!("rustfs_io_strategy_total", "storage_media" => storage_media.to_string(), "access_pattern" => access_pattern.to_string(), ) .increment(1); - gauge!("rustfs.io.buffer.size.bytes", + gauge!("rustfs_io_buffer_size_bytes", "storage_media" => storage_media.to_string(), ) .set(buffer_size as f64); - gauge!("rustfs.io.concurrent.requests").set(concurrent_requests as f64); + gauge!("rustfs_io_concurrent_requests").set(concurrent_requests as f64); } /// Record disk permit wait time (load tracking). @@ -448,7 +448,7 @@ pub fn record_io_strategy(storage_media: &str, access_pattern: &str, buffer_size /// * `duration_ms` - Time spent waiting for disk permit #[inline(always)] pub fn record_permit_wait(duration_ms: f64) { - histogram!("rustfs.io.permit.wait.duration.ms").record(duration_ms); + histogram!("rustfs_io_permit_wait_duration_ms").record(duration_ms); } /// Record I/O load level. @@ -459,12 +459,12 @@ pub fn record_permit_wait(duration_ms: f64) { /// * `concurrent_requests` - Number of concurrent requests #[inline(always)] pub fn record_io_load_level(load_level: &str, concurrent_requests: u64) { - counter!("rustfs.io.load.level", + counter!("rustfs_io_load_level", "level" => load_level.to_string(), ) .increment(1); - gauge!("rustfs.io.concurrent.requests").set(concurrent_requests as f64); + gauge!("rustfs_io_concurrent_requests").set(concurrent_requests as f64); } /// Record cache size and entry count. @@ -476,12 +476,12 @@ pub fn record_io_load_level(load_level: &str, concurrent_requests: u64) { /// * `entries` - Number of entries in the cache #[inline(always)] pub fn record_cache_size(tier: &str, size_bytes: usize, entries: u64) { - gauge!("rustfs.cache.size.bytes", + gauge!("rustfs_cache_size_bytes", "tier" => tier.to_string(), ) .set(size_bytes as f64); - gauge!("rustfs.cache.entries", + gauge!("rustfs_cache_entries", "tier" => tier.to_string(), ) .set(entries as f64); @@ -499,13 +499,11 @@ pub fn record_cache_size(tier: &str, size_bytes: usize, entries: u64) { /// * `tier` - Bandwidth tier ("low", "medium", "high", "unknown") #[inline(always)] pub fn record_bandwidth(bytes_per_second: u64, tier: &str) { - gauge!("rustfs.bandwidth.current.bps").set(bytes_per_second as f64); - gauge!("rustfs.bandwidth.current.bps", - "tier" => tier.to_string(), - ) - .set(bytes_per_second as f64); + let tier_label = if tier.is_empty() { "unknown" } else { tier }; + gauge!("rustfs_bandwidth_current_bps", "tier" => "all").set(bytes_per_second as f64); + gauge!("rustfs_bandwidth_current_bps", "tier" => tier_label.to_string()).set(bytes_per_second as f64); - histogram!("rustfs.bandwidth.observed.bps").record(bytes_per_second as f64); + histogram!("rustfs_bandwidth_observed_bps").record(bytes_per_second as f64); } /// Record data transfer for bandwidth calculation. @@ -516,12 +514,12 @@ pub fn record_bandwidth(bytes_per_second: u64, tier: &str) { /// * `duration_ms` - Duration of the transfer in milliseconds #[inline(always)] pub fn record_data_transfer(bytes: u64, duration_ms: f64) { - counter!("rustfs.io.transfer.bytes").increment(bytes); - histogram!("rustfs.io.transfer.duration.ms").record(duration_ms); + counter!("rustfs_io_transfer_bytes_total").increment(bytes); + histogram!("rustfs_io_transfer_duration_ms").record(duration_ms); if duration_ms > 0.0 { let bps = (bytes as f64 * 1000.0) / duration_ms; - histogram!("rustfs.io.transfer.bandwidth.bps").record(bps); + histogram!("rustfs_io_transfer_bandwidth_bps").record(bps); } } @@ -537,12 +535,12 @@ pub fn record_data_transfer(bytes: u64, duration_ms: f64) { /// * `total_bytes` - Total memory in bytes #[inline(always)] pub fn record_memory_usage(used_bytes: u64, total_bytes: u64) { - gauge!("rustfs.memory.used.bytes").set(used_bytes as f64); - gauge!("rustfs.memory.total.bytes").set(total_bytes as f64); + gauge!("rustfs_memory_used_bytes").set(used_bytes as f64); + gauge!("rustfs_memory_total_bytes").set(total_bytes as f64); if total_bytes > 0 { let usage_percent = (used_bytes as f64 / total_bytes as f64) * 100.0; - gauge!("rustfs.memory.usage.percent").set(usage_percent); + gauge!("rustfs_memory_usage_percent").set(usage_percent); } } @@ -553,7 +551,7 @@ pub fn record_memory_usage(used_bytes: u64, total_bytes: u64) { /// * `percent` - CPU usage percentage (0.0 - 100.0) #[inline(always)] pub fn record_cpu_usage(percent: f64) { - gauge!("rustfs.cpu.usage.percent").set(percent); + gauge!("rustfs_cpu_usage_percent").set(percent); } /// Record disk I/O statistics. @@ -566,13 +564,10 @@ pub fn record_cpu_usage(percent: f64) { /// * `write_ops` - Number of write operations #[inline(always)] pub fn record_disk_io(read_bytes: u64, write_bytes: u64, read_ops: u64, write_ops: u64) { - counter!("rustfs.disk.read.bytes").increment(read_bytes); - counter!("rustfs.disk.write.bytes").increment(write_bytes); - counter!("rustfs.disk.read.ops").increment(read_ops); - counter!("rustfs.disk.write.ops").increment(write_ops); - - gauge!("rustfs.disk.read.bytes_total").set(read_bytes as f64); - gauge!("rustfs.disk.write.bytes_total").set(write_bytes as f64); + counter!("rustfs_disk_read_bytes_total").increment(read_bytes); + counter!("rustfs_disk_write_bytes_total").increment(write_bytes); + counter!("rustfs_disk_read_ops_total").increment(read_ops); + counter!("rustfs_disk_write_ops_total").increment(write_ops); } // ============================================================================ @@ -587,7 +582,7 @@ pub fn record_disk_io(read_bytes: u64, write_bytes: u64, read_ops: u64, write_op /// * `error_type` - Error type (e.g., "timeout", "disk_error", "network") #[inline(always)] pub fn record_error(operation: &str, error_type: &str) { - counter!("rustfs.errors.total", + counter!("rustfs_errors_total", "operation" => operation.to_string(), "type" => error_type.to_string(), ) @@ -602,12 +597,12 @@ pub fn record_error(operation: &str, error_type: &str) { /// * `duration_ms` - Duration before timeout #[inline(always)] pub fn record_timeout(operation: &str, duration_ms: f64) { - counter!("rustfs.timeouts.total", + counter!("rustfs_timeouts_total", "operation" => operation.to_string(), ) .increment(1); - histogram!("rustfs.timeouts.duration.ms", + histogram!("rustfs_timeouts_duration_ms", "operation" => operation.to_string(), ) .record(duration_ms); @@ -621,12 +616,12 @@ pub fn record_timeout(operation: &str, duration_ms: f64) { /// * `attempt_number` - Attempt number (1-based) #[inline(always)] pub fn record_retry(operation: &str, attempt_number: u32) { - counter!("rustfs.retries.total", + counter!("rustfs_retries_total", "operation" => operation.to_string(), ) .increment(1); - histogram!("rustfs.retries.attempt", + histogram!("rustfs_retries_attempt", "operation" => operation.to_string(), ) .record(attempt_number as f64); @@ -643,7 +638,7 @@ pub fn record_retry(operation: &str, attempt_number: u32) { /// * `latency_ms` - I/O latency in milliseconds #[inline(always)] pub fn record_io_latency(latency_ms: f64) { - histogram!("rustfs.io.latency.ms").record(latency_ms); + histogram!("rustfs_io_latency_ms").record(latency_ms); } /// Record I/O latency P95 in milliseconds. @@ -653,7 +648,7 @@ pub fn record_io_latency(latency_ms: f64) { /// * `latency_ms` - P95 I/O latency in milliseconds #[inline(always)] pub fn record_io_latency_p95(latency_ms: f64) { - gauge!("rustfs.io.latency.p95.ms").set(latency_ms); + gauge!("rustfs_io_latency_p95_ms").set(latency_ms); } /// Record I/O latency P99 in milliseconds. @@ -663,7 +658,7 @@ pub fn record_io_latency_p95(latency_ms: f64) { /// * `latency_ms` - P99 I/O latency in milliseconds #[inline(always)] pub fn record_io_latency_p99(latency_ms: f64) { - gauge!("rustfs.io.latency.p99.ms").set(latency_ms); + gauge!("rustfs_io_latency_p99_ms").set(latency_ms); } #[cfg(test)] diff --git a/crates/io-metrics/src/lock_metrics.rs b/crates/io-metrics/src/lock_metrics.rs index 173e6f647..da1df38f6 100644 --- a/crates/io-metrics/src/lock_metrics.rs +++ b/crates/io-metrics/src/lock_metrics.rs @@ -20,7 +20,7 @@ use std::time::Duration; #[inline(always)] pub fn record_lock_optimization_enabled(enabled: bool) { use metrics::gauge; - gauge!("rustfs.lock.optimization.enabled").set(if enabled { 1.0 } else { 0.0 }); + gauge!("rustfs_lock_optimization_enabled").set(if enabled { 1.0 } else { 0.0 }); } /// Record spin attempt. @@ -28,9 +28,9 @@ pub fn record_lock_optimization_enabled(enabled: bool) { pub fn record_spin_attempt(success: bool) { use metrics::counter; if success { - counter!("rustfs.lock.spin.successes").increment(1); + counter!("rustfs_lock_spin_successes").increment(1); } else { - counter!("rustfs.lock.spin.failures").increment(1); + counter!("rustfs_lock_spin_failures").increment(1); } } @@ -38,28 +38,28 @@ pub fn record_spin_attempt(success: bool) { #[inline(always)] pub fn record_spin_count_change(new_count: usize) { use metrics::gauge; - gauge!("rustfs.lock.spin.count").set(new_count as f64); + gauge!("rustfs_lock_spin_count").set(new_count as f64); } /// Record lock hold time. #[inline(always)] pub fn record_lock_hold_time(hold_time: Duration) { use metrics::histogram; - histogram!("rustfs.lock.hold_time.secs").record(hold_time.as_secs_f64()); + histogram!("rustfs_lock_hold_time_secs").record(hold_time.as_secs_f64()); } /// Record early release. #[inline(always)] pub fn record_early_release() { use metrics::counter; - counter!("rustfs.lock.early_releases").increment(1); + counter!("rustfs_lock_early_releases").increment(1); } /// Record contention event. #[inline(always)] pub fn record_contention_event() { use metrics::counter; - counter!("rustfs.lock.contentions").increment(1); + counter!("rustfs_lock_contentions").increment(1); } /// Lock statistics summary. diff --git a/crates/io-metrics/src/metric_names.rs b/crates/io-metrics/src/metric_names.rs index e7581ff8f..230e634f1 100644 --- a/crates/io-metrics/src/metric_names.rs +++ b/crates/io-metrics/src/metric_names.rs @@ -49,6 +49,6 @@ pub mod zero_copy { /// Throughput in MB/s pub const THROUGHPUT_MBPS: &str = "rustfs_zero_copy_throughput_mbps"; - /// Memory saved by zero-copy in bytes - pub const MEMORY_SAVED_BYTES: &str = "rustfs_zero_copy_memory_saved_bytes"; + /// Current memory saved estimate by zero-copy in bytes + pub const MEMORY_SAVED_BYTES: &str = "rustfs_zero_copy_memory_saved_bytes_current"; } diff --git a/crates/io-metrics/src/timeout_metrics.rs b/crates/io-metrics/src/timeout_metrics.rs index 1561abf36..436616f46 100644 --- a/crates/io-metrics/src/timeout_metrics.rs +++ b/crates/io-metrics/src/timeout_metrics.rs @@ -34,23 +34,23 @@ pub fn record_operation_duration(operation: &str, duration: Duration) { #[inline(always)] pub fn record_dynamic_timeout(size_bytes: u64, timeout: Duration) { use metrics::{gauge, histogram}; - gauge!("rustfs.timeout.dynamic.size").set(size_bytes as f64); - gauge!("rustfs.timeout.dynamic.secs").set(timeout.as_secs_f64()); - histogram!("rustfs.timeout.dynamic.size.histogram").record(size_bytes as f64); + gauge!("rustfs_timeout_dynamic_size").set(size_bytes as f64); + gauge!("rustfs_timeout_dynamic_secs").set(timeout.as_secs_f64()); + histogram!("rustfs_timeout_dynamic_size_histogram").record(size_bytes as f64); } /// Record operation progress. #[inline(always)] pub fn record_operation_progress(operation: &str, percent: f64) { use metrics::gauge; - gauge!("rustfs.operation.progress", "operation" => operation.to_string()).set(percent); + gauge!("rustfs_operation_progress", "operation" => operation.to_string()).set(percent); } /// Record stalled operation. #[inline(always)] pub fn record_stalled_operation(operation: &str) { use metrics::counter; - counter!("rustfs.operation.stalled", "operation" => operation.to_string()).increment(1); + counter!("rustfs_operation_stalled", "operation" => operation.to_string()).increment(1); } /// Record operation completion. @@ -58,7 +58,7 @@ pub fn record_stalled_operation(operation: &str) { pub fn record_operation_completion(operation: &str, success: bool) { use metrics::counter; let status = if success { "success" } else { "failure" }; - counter!("rustfs.operation.completions", "operation" => operation.to_string(), "status" => status).increment(1); + counter!("rustfs_operation_completions", "operation" => operation.to_string(), "status" => status).increment(1); } /// Timeout statistics summary. diff --git a/crates/obs/Cargo.toml b/crates/obs/Cargo.toml index 1132a33cc..60ae3e9a2 100644 --- a/crates/obs/Cargo.toml +++ b/crates/obs/Cargo.toml @@ -34,11 +34,14 @@ workspace = true [dependencies] rustfs-audit = { workspace = true } +rustfs-common = { workspace = true } rustfs-config = { workspace = true, features = ["constants", "observability"] } rustfs-ecstore = { workspace = true } +rustfs-iam = { workspace = true } rustfs-io-metrics = { workspace = true } rustfs-notify = { workspace = true } rustfs-utils = { workspace = true, features = ["ip"] } +chrono = { workspace = true } flate2 = { workspace = true } glob = { workspace = true } jiff = { workspace = true } diff --git a/crates/obs/README.md b/crates/obs/README.md index 59c079059..570083b49 100644 --- a/crates/obs/README.md +++ b/crates/obs/README.md @@ -171,16 +171,16 @@ The log rotation and cleanup pipeline emits these metrics (via the `metrics` fac | Metric | Type | Description | |---|---|---| -| `rustfs.log_cleaner.deleted_files_total` | counter | Number of files deleted per cleanup pass | -| `rustfs.log_cleaner.freed_bytes_total` | counter | Bytes reclaimed by deletion | -| `rustfs.log_cleaner.compress_duration_seconds` | histogram | Compression stage duration | -| `rustfs.log_cleaner.steal_success_rate` | gauge | Work-stealing success ratio in parallel mode | -| `rustfs.log_cleaner.runs_total` | counter | Successful cleanup loop runs | -| `rustfs.log_cleaner.run_failures_total` | counter | Failed or panicked cleanup loop runs | -| `rustfs.log_cleaner.rotation_total` | counter | Successful file rotations | -| `rustfs.log_cleaner.rotation_failures_total` | counter | Failed file rotations | -| `rustfs.log_cleaner.rotation_duration_seconds` | histogram | Rotation latency | -| `rustfs.log_cleaner.active_file_size_bytes` | gauge | Current active log file size | +| `rustfs_log_cleaner_deleted_files_total` | counter | Number of files deleted per cleanup pass | +| `rustfs_log_cleaner_freed_bytes_total` | counter | Bytes reclaimed by deletion | +| `rustfs_log_cleaner_compress_duration_seconds` | histogram | Compression stage duration | +| `rustfs_log_cleaner_steal_success_rate` | gauge | Work-stealing success ratio in parallel mode | +| `rustfs_log_cleaner_runs_total` | counter | Successful cleanup loop runs | +| `rustfs_log_cleaner_run_failures_total` | counter | Failed or panicked cleanup loop runs | +| `rustfs_log_cleaner_rotation_total` | counter | Successful file rotations | +| `rustfs_log_cleaner_rotation_failures_total` | counter | Failed file rotations | +| `rustfs_log_cleaner_rotation_duration_seconds` | histogram | Rotation latency | +| `rustfs_log_cleaner_active_file_size_bytes` | gauge | Current active log file size | These metrics cover compression, cleanup, and file rotation end-to-end. @@ -195,8 +195,8 @@ These metrics cover compression, cleanup, and file rotation end-to-end. ### Grafana Dashboard JSON Draft (Ready to Import) > Save this as `rustfs-log-cleaner-dashboard.json`, then import from Grafana UI. -> For Prometheus datasources, metric names are usually normalized to underscores, -> so `rustfs.log_cleaner.deleted_files_total` becomes `rustfs_log_cleaner_deleted_files_total`. +> The canonical metric names use underscore notation, for example +> `rustfs_log_cleaner_deleted_files_total`. > > The same panels are now checked in at: > `.docker/observability/grafana/dashboards/rustfs.json` diff --git a/crates/obs/src/cleaner/README.md b/crates/obs/src/cleaner/README.md index 70e32fed7..dfadd6d72 100644 --- a/crates/obs/src/cleaner/README.md +++ b/crates/obs/src/cleaner/README.md @@ -58,14 +58,14 @@ This strategy keeps local cache affinity while still balancing stragglers. The cleaner emits tracing events and runtime metrics: -- `rustfs.log_cleaner.deleted_files_total` (counter) -- `rustfs.log_cleaner.freed_bytes_total` (counter) -- `rustfs.log_cleaner.compress_duration_seconds` (histogram) -- `rustfs.log_cleaner.steal_success_rate` (gauge) -- `rustfs.log_cleaner.rotation_total` (counter) -- `rustfs.log_cleaner.rotation_failures_total` (counter) -- `rustfs.log_cleaner.rotation_duration_seconds` (histogram) -- `rustfs.log_cleaner.active_file_size_bytes` (gauge) +- `rustfs_log_cleaner_deleted_files_total` (counter) +- `rustfs_log_cleaner_freed_bytes_total` (counter) +- `rustfs_log_cleaner_compress_duration_seconds` (histogram) +- `rustfs_log_cleaner_steal_success_rate` (gauge) +- `rustfs_log_cleaner_rotation_total` (counter) +- `rustfs_log_cleaner_rotation_failures_total` (counter) +- `rustfs_log_cleaner_rotation_duration_seconds` (histogram) +- `rustfs_log_cleaner_active_file_size_bytes` (gauge) These values can be wired into dashboards and alert rules for cleanup health. @@ -127,4 +127,3 @@ let _ = cleaner.cleanup(); - Prefer `FileMatchMode::Suffix` when rotations prepend timestamps to the filename. - Prefer `FileMatchMode::Prefix` when rotations append counters or timestamps after a stable base name. - Keep `parallel_workers` modest when `zstd_workers` is greater than `1`, because each compression task may already use internal codec threads. - diff --git a/crates/obs/src/metrics/collectors/cluster_iam.rs b/crates/obs/src/metrics/collectors/cluster_iam.rs index 50cdf3a48..a33330125 100644 --- a/crates/obs/src/metrics/collectors/cluster_iam.rs +++ b/crates/obs/src/metrics/collectors/cluster_iam.rs @@ -25,6 +25,10 @@ use crate::metrics::schema::cluster_iam::*; /// IAM statistics. #[derive(Debug, Clone, Default)] pub struct IamStats { + /// Last successful IAM data sync duration in milliseconds + pub last_sync_duration_millis: u64, + /// Failed requests count in the last full minute + pub plugin_authn_service_failed_requests_minute: u64, /// Time in seconds since last failed authn service request pub plugin_authn_service_last_fail_seconds: u64, /// Time in seconds since last successful authn service request @@ -48,6 +52,11 @@ pub struct IamStats { /// Returns a vector of Prometheus metrics for IAM statistics. pub fn collect_iam_metrics(stats: &IamStats) -> Vec { vec![ + PrometheusMetric::from_descriptor(&LAST_SYNC_DURATION_MILLIS_MD, stats.last_sync_duration_millis as f64), + PrometheusMetric::from_descriptor( + &PLUGIN_AUTHN_SERVICE_FAILED_REQUESTS_MINUTE_MD, + stats.plugin_authn_service_failed_requests_minute as f64, + ), PrometheusMetric::from_descriptor( &PLUGIN_AUTHN_SERVICE_LAST_FAIL_SECONDS_MD, stats.plugin_authn_service_last_fail_seconds as f64, @@ -82,6 +91,8 @@ mod tests { #[test] fn test_collect_iam_metrics() { let stats = IamStats { + last_sync_duration_millis: 250, + plugin_authn_service_failed_requests_minute: 7, plugin_authn_service_last_fail_seconds: 3600, plugin_authn_service_last_succ_seconds: 10, plugin_authn_service_succ_avg_rtt_ms_minute: 50, @@ -95,7 +106,7 @@ mod tests { let metrics = collect_iam_metrics(&stats); report_metrics(&metrics); - assert_eq!(metrics.len(), 8); + assert_eq!(metrics.len(), 10); let sync_successes_name = SYNC_SUCCESSES_MD.get_full_metric_name(); let sync_successes = metrics.iter().find(|m| m.name == sync_successes_name); @@ -108,7 +119,7 @@ mod tests { let stats = IamStats::default(); let metrics = collect_iam_metrics(&stats); - assert_eq!(metrics.len(), 8); + assert_eq!(metrics.len(), 10); for metric in &metrics { assert_eq!(metric.value, 0.0); assert!(metric.labels.is_empty()); diff --git a/crates/obs/src/metrics/scheduler.rs b/crates/obs/src/metrics/scheduler.rs index 703af23d4..35fa36beb 100644 --- a/crates/obs/src/metrics/scheduler.rs +++ b/crates/obs/src/metrics/scheduler.rs @@ -36,13 +36,20 @@ use crate::metrics::collectors::{ collect_bucket_metrics, collect_bucket_replication_bandwidth_metrics, collect_bucket_replication_metrics, + collect_bucket_usage_metrics, + collect_cluster_config_metrics, collect_cluster_health_metrics, collect_cluster_metrics, + collect_cluster_usage_metrics, collect_cpu_metrics, collect_drive_count_metrics, collect_drive_detailed_metrics, + collect_erasure_set_metrics, collect_host_network_metrics, + collect_iam_metrics, + collect_ilm_metrics, collect_memory_metrics, + collect_network_metrics, collect_node_metrics, collect_notification_metrics, collect_notification_target_metrics, @@ -53,6 +60,7 @@ use crate::metrics::collectors::{ collect_process_metrics, collect_replication_metrics, collect_resource_metrics, + collect_scanner_metrics, }; use crate::metrics::config::{ DEFAULT_AUDIT_METRICS_INTERVAL, DEFAULT_BUCKET_METRICS_INTERVAL, DEFAULT_BUCKET_REPLICATION_BANDWIDTH_METRICS_INTERVAL, @@ -64,8 +72,10 @@ use crate::metrics::config::{ use crate::metrics::report::{PrometheusMetric, report_metrics}; use crate::metrics::stats_collector::{ ProcessMetricBundle, collect_bucket_replication_bandwidth_stats, collect_bucket_replication_detail_stats, - collect_bucket_stats, collect_cluster_and_health_stats, collect_disk_and_system_drive_stats, collect_host_network_stats, - collect_process_metric_bundle, collect_replication_stats, collect_system_cpu_and_memory_stats_with, + collect_bucket_stats, collect_cluster_and_health_stats, collect_cluster_config_stats, collect_cluster_usage_metric_stats, + collect_disk_and_system_drive_stats, collect_erasure_set_stats, collect_host_network_stats, collect_iam_stats, + collect_ilm_metric_stats, collect_internode_network_stats, collect_process_metric_bundle, collect_replication_stats, + collect_scanner_metric_stats, collect_system_cpu_and_memory_stats_with, }; use rustfs_audit::audit_target_metrics; use rustfs_notify::{notification_metrics_snapshot, notification_target_metrics}; @@ -176,6 +186,46 @@ pub fn init_metrics_runtime(token: CancellationToken) { } }); + // Spawn task for supplementary cluster metrics that are defined in schema/collector + // but filled by later task-specific runtime sources. + let token_clone = token.clone(); + tokio::spawn(async move { + let mut interval = tokio::time::interval(cluster_interval); + loop { + tokio::select! { + _ = interval.tick() => { + let mut metrics = Vec::new(); + + if let Some(stats) = collect_cluster_config_stats().await { + metrics.extend(collect_cluster_config_metrics(&stats)); + } + + let erasure_sets = collect_erasure_set_stats().await; + if !erasure_sets.is_empty() { + metrics.extend(collect_erasure_set_metrics(&erasure_sets)); + } + + if let Some(stats) = collect_iam_stats().await { + metrics.extend(collect_iam_metrics(&stats)); + } + + if let Some((cluster_usage, bucket_usage)) = collect_cluster_usage_metric_stats().await { + metrics.extend(collect_cluster_usage_metrics(&cluster_usage)); + metrics.extend(collect_bucket_usage_metrics(&bucket_usage)); + } + + if !metrics.is_empty() { + report_metrics(&metrics); + } + } + _ = token_clone.cancelled() => { + warn!("Metrics collection for supplementary cluster stats cancelled."); + return; + } + } + } + }); + // Spawn task for bucket metrics let token_clone = token.clone(); tokio::spawn(async move { @@ -302,6 +352,35 @@ pub fn init_metrics_runtime(token: CancellationToken) { } }); + // Spawn task for background workflow metrics such as ILM and scanner. + let token_clone = token.clone(); + tokio::spawn(async move { + let mut interval = tokio::time::interval(cluster_interval); + loop { + tokio::select! { + _ = interval.tick() => { + let mut metrics = Vec::new(); + + if let Some(stats) = collect_ilm_metric_stats().await { + metrics.extend(collect_ilm_metrics(&stats)); + } + + if let Some(stats) = collect_scanner_metric_stats().await { + metrics.extend(collect_scanner_metrics(&stats)); + } + + if !metrics.is_empty() { + report_metrics(&metrics); + } + } + _ = token_clone.cancelled() => { + warn!("Metrics collection for background workflow stats cancelled."); + return; + } + } + } + }); + // Spawn task for system monitoring metrics (migrated from rustfs-obs::system) let system_interval = get_env_opt_u64(ENV_SYSTEM_METRICS_INTERVAL) .or_else(|| get_env_opt_u64(LEGACY_SYSTEM_METRICS_INTERVAL).map(|ms| ms / 1000)) // Convert ms to seconds @@ -310,7 +389,7 @@ pub fn init_metrics_runtime(token: CancellationToken) { .map(Duration::from_secs) .unwrap_or(DEFAULT_SYSTEM_METRICS_INTERVAL); - let token_clone = token; + let token_clone = token.clone(); tokio::spawn(async move { let labels = current_process_metric_labels(); let mut host_system = System::new_all(); @@ -378,6 +457,28 @@ pub fn init_metrics_runtime(token: CancellationToken) { } } }); + + // Spawn task for internode/system network metrics. + let token_clone = token; + tokio::spawn(async move { + let mut interval = tokio::time::interval(system_interval); + loop { + tokio::select! { + _ = interval.tick() => { + if let Some(stats) = collect_internode_network_stats() { + let metrics = collect_network_metrics(&stats); + if !metrics.is_empty() { + report_metrics(&metrics); + } + } + } + _ = token_clone.cancelled() => { + warn!("Metrics collection for internode network stats cancelled."); + return; + } + } + } + }); } /// Backward-compatible alias kept during migration. diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs index 0f735edbf..9dff01d14 100644 --- a/crates/obs/src/metrics/stats_collector.rs +++ b/crates/obs/src/metrics/stats_collector.rs @@ -21,10 +21,14 @@ //! and convert them to the Stats structs used by collectors. use crate::metrics::collectors::{ - BucketReplicationBandwidthStats, BucketReplicationStats, BucketReplicationTargetStats, BucketStats, ClusterHealthStats, - ClusterStats, CpuStats, DiskStats, DriveCountStats, DriveDetailedStats, HostNetworkStats, MemoryStats, ProcessStats, - ProcessStatusType, ReplicationStats, ResourceStats, + BucketReplicationBandwidthStats, BucketReplicationStats, BucketReplicationTargetStats, BucketStats, BucketUsageStats, + ClusterConfigStats, ClusterHealthStats, ClusterStats, ClusterUsageStats, CpuStats, DiskStats, DriveCountStats, + DriveDetailedStats, ErasureSetStats, HostNetworkStats, IamStats, IlmStats, MemoryStats, NetworkStats, ProcessStats, + ProcessStatusType, ReplicationStats, ResourceStats, ScannerStats, }; +use chrono::Utc; +use rustfs_common::{internode_metrics::global_internode_metrics, metrics::global_metrics}; +use rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_ops::{GLOBAL_ExpiryState, GLOBAL_TransitionState}; use rustfs_ecstore::bucket::metadata_sys::get_quota_config; use rustfs_ecstore::bucket::replication::GLOBAL_REPLICATION_STATS; use rustfs_ecstore::data_usage::load_data_usage_from_backend; @@ -32,7 +36,9 @@ use rustfs_ecstore::global::get_global_bucket_monitor; use rustfs_ecstore::pools::{get_total_usable_capacity, get_total_usable_capacity_free}; use rustfs_ecstore::store_api::{BucketOperations, BucketOptions}; use rustfs_ecstore::{StorageAPI, new_object_layer_fn}; +use rustfs_iam::{get_global_iam_sys, oidc::oidc_plugin_authn_metrics_snapshot}; use rustfs_io_metrics::{ProcessStatusSnapshot, snapshot_process_resource_and_system}; +use std::collections::HashMap; use std::time::Duration; use sysinfo::{Networks, System}; use tracing::{instrument, warn}; @@ -42,6 +48,15 @@ const DRIVE_STATE_ONLINE: &str = "online"; const DRIVE_STATE_UNFORMATTED: &str = "unformatted"; const DRIVE_RUNTIME_STATE_RETURNING: &str = "returning"; +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct ErasureSetQuorumShape { + data_shards: u32, + read_quorum: u32, + write_quorum: u32, + read_tolerance: u32, + write_tolerance: u32, +} + fn disk_is_online_for_metrics(state: &str, runtime_state: Option<&str>) -> bool { let state_is_acceptable = state.eq_ignore_ascii_case(DRIVE_STATE_OK) || state.eq_ignore_ascii_case(DRIVE_STATE_ONLINE) @@ -56,6 +71,30 @@ fn disk_is_online_for_metrics(state: &str, runtime_state: Option<&str>) -> bool state_is_acceptable } +fn derive_erasure_set_quorum_shape(set_drive_count: usize, parity: usize) -> ErasureSetQuorumShape { + let data_shards = set_drive_count.saturating_sub(parity); + let read_quorum = data_shards.max(1); + let mut write_quorum = read_quorum; + if data_shards == parity { + write_quorum += 1; + } + + ErasureSetQuorumShape { + data_shards: data_shards as u32, + read_quorum: read_quorum as u32, + write_quorum: write_quorum as u32, + read_tolerance: parity as u32, + write_tolerance: set_drive_count.saturating_sub(write_quorum) as u32, + } +} + +fn apply_erasure_set_health(entry: &mut ErasureSetStats) { + let online = entry.online_drives_count; + entry.read_health = u8::from(online >= entry.read_quorum); + entry.write_health = u8::from(online >= entry.write_quorum); + entry.health = u8::from(entry.write_health == 1); +} + #[derive(Debug, Clone, Default)] pub struct ProcessMetricBundle { pub resource: ResourceStats, @@ -285,14 +324,12 @@ pub async fn collect_bucket_replication_detail_stats() -> Vec HostNetworkStats { } } +/// Collect internode network metrics from the global internode metrics snapshot. +/// +/// The returned values come directly from `global_internode_metrics().snapshot()` +/// and currently include only the counters and dial timing data tracked by the +/// internode metrics runtime. +pub fn collect_internode_network_stats() -> Option { + let snapshot = global_internode_metrics().snapshot(); + + Some(NetworkStats { + internode_errors_total: snapshot.errors_total, + internode_dial_errors_total: snapshot.dial_errors_total, + internode_dial_avg_time_nanos: snapshot.dial_avg_time_nanos, + internode_sent_bytes_total: snapshot.sent_bytes_total, + internode_recv_bytes_total: snapshot.recv_bytes_total, + }) +} + +/// Collect cluster config metrics from backend parity configuration. +pub async fn collect_cluster_config_stats() -> Option { + let store = new_object_layer_fn()?; + let backend = store.backend_info().await; + + Some(ClusterConfigStats { + rrs_parity: backend.rr_sc_parity.unwrap_or_default() as u32, + standard_parity: backend.standard_sc_parity.unwrap_or_default() as u32, + }) +} + +/// Collect cluster erasure set metrics from storage and backend topology info. +pub async fn collect_erasure_set_stats() -> Vec { + let Some(store) = new_object_layer_fn() else { + return Vec::new(); + }; + + let storage_info = store.storage_info().await; + let backend = store.backend_info().await; + let mut grouped: HashMap<(usize, usize), ErasureSetStats> = HashMap::new(); + + for disk in &storage_info.disks { + let pool_idx = disk.pool_index.max(0) as usize; + let set_idx = disk.set_index.max(0) as usize; + let set_drive_count = backend.drives_per_set.get(pool_idx).copied().unwrap_or_default(); + let parity = backend + .standard_sc_parities + .get(pool_idx) + .copied() + .or(backend.standard_sc_parity) + .unwrap_or(set_drive_count / 2); + let quorum_shape = derive_erasure_set_quorum_shape(set_drive_count, parity); + + let entry = grouped.entry((pool_idx, set_idx)).or_insert_with(|| ErasureSetStats { + pool_id: pool_idx as u32, + set_id: set_idx as u32, + size: set_drive_count as u32, + parity: parity as u32, + data_shards: quorum_shape.data_shards, + read_quorum: quorum_shape.read_quorum, + write_quorum: quorum_shape.write_quorum, + online_drives_count: 0, + healing_drives_count: 0, + health: 0, + read_tolerance: quorum_shape.read_tolerance, + write_tolerance: quorum_shape.write_tolerance, + read_health: 0, + write_health: 0, + }); + + if disk_is_online_for_metrics(disk.state.as_str(), disk.runtime_state.as_deref()) { + entry.online_drives_count += 1; + } + if disk.healing { + entry.healing_drives_count += 1; + } + } + + for entry in grouped.values_mut() { + apply_erasure_set_health(entry); + } + + let mut stats = grouped.into_values().collect::>(); + stats.sort_by_key(|stat| (stat.pool_id, stat.set_id)); + stats +} + +pub async fn collect_iam_stats() -> Option { + let iam_sys = get_global_iam_sys()?; + let sync = iam_sys.sync_metrics_snapshot(); + let oidc = oidc_plugin_authn_metrics_snapshot(); + + Some(IamStats { + last_sync_duration_millis: sync.last_sync_duration_millis, + plugin_authn_service_failed_requests_minute: oidc.failed_requests_minute, + plugin_authn_service_last_fail_seconds: oidc.last_fail_seconds, + plugin_authn_service_last_succ_seconds: oidc.last_succ_seconds, + plugin_authn_service_succ_avg_rtt_ms_minute: oidc.succ_avg_rtt_ms_minute, + plugin_authn_service_succ_max_rtt_ms_minute: oidc.succ_max_rtt_ms_minute, + plugin_authn_service_total_requests_minute: oidc.total_requests_minute, + since_last_sync_millis: sync.since_last_sync_millis, + sync_failures: sync.sync_failures, + sync_successes: sync.sync_successes, + }) +} + +/// Collect cluster and per-bucket usage metrics from backend usage snapshots. +/// +/// This reads persisted usage data via `load_data_usage_from_backend()` and +/// builds cluster totals plus per-bucket distributions from the returned +/// histograms. It does not trigger an inline object-data rescan. +pub async fn collect_cluster_usage_metric_stats() -> Option<(ClusterUsageStats, Vec)> { + let store = new_object_layer_fn()?; + let data_usage = load_data_usage_from_backend(store.clone()).await.ok()?; + let mut buckets = Vec::with_capacity(data_usage.buckets_usage.len()); + + for (bucket_name, usage) in &data_usage.buckets_usage { + if bucket_name.starts_with('.') { + continue; + } + + let quota_bytes = match get_quota_config(bucket_name).await { + Ok((quota, _)) => quota.get_quota_limit().unwrap_or(0), + Err(_) => 0, + }; + + buckets.push(BucketUsageStats { + bucket: bucket_name.clone(), + total_bytes: usage.size, + objects_count: usage.objects_count, + versions_count: usage.versions_count, + delete_markers_count: usage.delete_markers_count, + quota_bytes, + object_size_distribution: usage + .object_size_histogram + .iter() + .map(|(range, count)| (range.clone(), *count)) + .collect(), + version_count_distribution: usage + .object_versions_histogram + .iter() + .map(|(range, count)| (range.clone(), *count)) + .collect(), + }); + } + + buckets.sort_by(|a, b| a.bucket.cmp(&b.bucket)); + + Some(( + ClusterUsageStats { + total_bytes: data_usage.objects_total_size, + objects_count: data_usage.objects_total_count, + versions_count: data_usage.versions_total_count, + delete_markers_count: data_usage.delete_markers_total_count, + object_size_distribution: data_usage + .buckets_usage + .values() + .flat_map(|usage| usage.object_size_histogram.iter()) + .fold(HashMap::::new(), |mut acc, (range, count)| { + *acc.entry(range.clone()).or_default() += *count; + acc + }) + .into_iter() + .collect(), + versions_distribution: data_usage + .buckets_usage + .values() + .flat_map(|usage| usage.object_versions_histogram.iter()) + .fold(HashMap::::new(), |mut acc, (range, count)| { + *acc.entry(range.clone()).or_default() += *count; + acc + }) + .into_iter() + .collect(), + }, + buckets, + )) +} + +/// Collect ILM metrics from the current lifecycle runtime state. +pub async fn collect_ilm_metric_stats() -> Option { + let expiry_pending_tasks = GLOBAL_ExpiryState.read().await.pending_tasks().await as u64; + let transition_active_tasks = GLOBAL_TransitionState.active_tasks().max(0) as u64; + let transition_pending_tasks = GLOBAL_TransitionState.pending_tasks() as u64; + let transition_missed_immediate_tasks = GLOBAL_TransitionState.missed_immediate_tasks().max(0) as u64; + let metrics = global_metrics().report().await; + let versions_scanned = metrics.life_time_ilm.values().copied().sum(); + + Some(IlmStats { + expiry_pending_tasks, + transition_active_tasks, + transition_pending_tasks, + transition_missed_immediate_tasks, + versions_scanned, + }) +} + +/// Collect scanner metrics from a runtime source. +/// +/// Task 5 maps scanner runtime snapshots from `global_metrics()` into the +/// rustfs-obs scanner collector shape. +pub async fn collect_scanner_metric_stats() -> Option { + let metrics = global_metrics().report().await; + let bucket_scans_finished = metrics.life_time_ops.get("scan_bucket_drive").copied().unwrap_or_default(); + let directories_scanned = metrics.life_time_ops.get("scan_folder").copied().unwrap_or_default(); + let objects_scanned = metrics.life_time_ops.get("scan_object").copied().unwrap_or_default(); + let versions_scanned = metrics.life_time_ilm.values().copied().sum(); + let reference_time = metrics.cycles_completed_at.last().copied().unwrap_or(metrics.current_started); + let last_activity_seconds = Utc::now().signed_duration_since(reference_time).num_seconds().max(0) as u64; + + Some(ScannerStats { + bucket_scans_finished, + // `global_metrics()` currently tracks completed bucket-drive scans, not a + // separate started counter. Mirror the finished count until Task 5/Task 10 + // expands the scanner runtime source shape. + bucket_scans_started: bucket_scans_finished, + directories_scanned, + objects_scanned, + versions_scanned, + last_activity_seconds, + }) +} + #[cfg(test)] mod tests { use super::*; @@ -598,4 +855,57 @@ mod tests { fn disk_is_online_for_metrics_rejects_offline_runtime_state() { assert!(!disk_is_online_for_metrics(DRIVE_STATE_OK, Some("offline"))); } + + #[test] + fn derive_erasure_set_quorum_shape_handles_standard_layout() { + let shape = derive_erasure_set_quorum_shape(16, 4); + + assert_eq!( + shape, + ErasureSetQuorumShape { + data_shards: 12, + read_quorum: 12, + write_quorum: 12, + read_tolerance: 4, + write_tolerance: 4, + } + ); + } + + #[test] + fn derive_erasure_set_quorum_shape_handles_equal_data_and_parity() { + let shape = derive_erasure_set_quorum_shape(4, 2); + + assert_eq!( + shape, + ErasureSetQuorumShape { + data_shards: 2, + read_quorum: 2, + write_quorum: 3, + read_tolerance: 2, + write_tolerance: 1, + } + ); + } + + #[test] + fn apply_erasure_set_health_marks_read_and_write_health_from_online_count() { + let mut stats = ErasureSetStats { + read_quorum: 3, + write_quorum: 4, + online_drives_count: 3, + ..Default::default() + }; + + apply_erasure_set_health(&mut stats); + assert_eq!(stats.read_health, 1); + assert_eq!(stats.write_health, 0); + assert_eq!(stats.health, 0); + + stats.online_drives_count = 4; + apply_erasure_set_health(&mut stats); + assert_eq!(stats.read_health, 1); + assert_eq!(stats.write_health, 1); + assert_eq!(stats.health, 1); + } } diff --git a/crates/obs/src/telemetry/local.rs b/crates/obs/src/telemetry/local.rs index b73d3ac23..3c972ad50 100644 --- a/crates/obs/src/telemetry/local.rs +++ b/crates/obs/src/telemetry/local.rs @@ -148,7 +148,7 @@ fn init_stdout_only(_config: &OtelConfig, logger_level: &str, is_production: boo .init(); set_observability_metric_enabled(false); - counter!("rustfs.start.total").increment(1); + counter!("rustfs_start_total").increment(1); info!("Init stdout logging (level: {})", logger_level); OtelGuard { diff --git a/crates/obs/src/telemetry/otel.rs b/crates/obs/src/telemetry/otel.rs index d575da2fd..3793ada1d 100644 --- a/crates/obs/src/telemetry/otel.rs +++ b/crates/obs/src/telemetry/otel.rs @@ -304,7 +304,7 @@ pub(super) fn init_observability_http( .with(metrics_layer) .init(); - counter!("rustfs.start.total").increment(1); + counter!("rustfs_start_total").increment(1); info!( "Init observability (HTTP): trace='{}', metric='{}', log='{}'", trace_ep, metric_ep, log_ep diff --git a/crates/s3select-api/src/query/execution.rs b/crates/s3select-api/src/query/execution.rs index 1ed532cce..9a1148fba 100644 --- a/crates/s3select-api/src/query/execution.rs +++ b/crates/s3select-api/src/query/execution.rs @@ -52,7 +52,7 @@ impl Drop for PhaseTimer { if !std::thread::panicking() { metrics::histogram!( - "rustfs.s3select.phase.duration.seconds", + "rustfs_s3select_phase_duration_seconds", "phase" => self.phase_name ) .record(duration.as_secs_f64()); @@ -171,7 +171,7 @@ impl QueryStateMachine { fn record_phase_timestamp(&self, phase: &'static str, event: &'static str) { let elapsed = self.start.elapsed(); metrics::histogram!( - "rustfs.s3select.phase.timestamp.seconds", + "rustfs_s3select_phase_timestamp_seconds", "phase" => phase, "event" => event ) diff --git a/rustfs/src/admin/handlers/site_replication.rs b/rustfs/src/admin/handlers/site_replication.rs index a27fb5477..3e662932e 100644 --- a/rustfs/src/admin/handlers/site_replication.rs +++ b/rustfs/src/admin/handlers/site_replication.rs @@ -1282,8 +1282,8 @@ async fn build_metrics_summary(local_peer: &PeerInfo) -> SRMetricsSummary { head_total: non_negative_u64(node.proxied.head_total), get_failed_total: non_negative_u64(node.proxied.get_failed), head_failed_total: non_negative_u64(node.proxied.head_failed), - put_tag_total: non_negative_u64(node.proxied.put_total), - put_tag_failed_total: non_negative_u64(node.proxied.put_failed), + put_tag_total: non_negative_u64(node.proxied.put_tag_total), + put_tag_failed_total: non_negative_u64(node.proxied.put_tag_failed), ..Default::default() }, metrics, diff --git a/rustfs/src/app/object_usecase.rs b/rustfs/src/app/object_usecase.rs index 6e7de2b37..02d70e870 100644 --- a/rustfs/src/app/object_usecase.rs +++ b/rustfs/src/app/object_usecase.rs @@ -1651,8 +1651,8 @@ impl DefaultObjectUsecase { if enable_zero_copy { // Record zero-copy write attempt - counter!("rustfs.zero_copy.write.attempts.total").increment(1); - histogram!("rustfs.zero_copy.write.size.bytes").record(size as f64); + counter!("rustfs_zero_copy_write_attempts_total").increment(1); + histogram!("rustfs_zero_copy_write_size_bytes").record(size as f64); debug!("Zero-copy write enabled for {} byte object (bucket={}, key={})", size, bucket, key); } diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index ac638ec28..6a54998cb 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -38,7 +38,7 @@ use hyper_util::{ server::graceful::GracefulShutdown, service::TowerToHyperService, }; -use metrics::{counter, histogram}; +use metrics::{counter, gauge, histogram}; use opentelemetry::global; use opentelemetry::trace::TraceContextExt; use rustfs_common::GlobalReadiness; @@ -54,6 +54,7 @@ use socket2::{SockRef, TcpKeepalive}; use std::io::{Error, Result}; use std::net::SocketAddr; use std::sync::Arc; +use std::sync::atomic::{AtomicU64, Ordering}; use std::time::Duration; use tokio::net::{TcpListener, TcpStream}; use tonic::{Request, Status}; @@ -66,12 +67,18 @@ use tower_http::trace::TraceLayer; use tracing::{Span, debug, error, info, instrument, warn}; use tracing_opentelemetry::OpenTelemetrySpanExt; -const LABEL_REQUEST_METHOD: &str = "request_method"; -const METRIC_API_REQUESTS_TOTAL: &str = "rustfs_api_requests_total"; -const METRIC_API_REQUESTS_FAILURE_TOTAL: &str = "rustfs_api_requests_failure_total"; -const METRIC_REQUEST_BODY_BYTES_TOTAL: &str = "rustfs_request_body_bytes_total"; -const METRIC_REQUEST_LATENCY_MS: &str = "rustfs_request_latency_ms"; -const METRIC_REQUEST_BODY_LEN: &str = "rustfs_request_body_len"; +const LABEL_HTTP_METHOD: &str = "method"; +const LABEL_HTTP_STATUS_CLASS: &str = "status_class"; +const METRIC_HTTP_SERVER_REQUESTS_TOTAL: &str = "rustfs_http_server_requests_total"; +const METRIC_HTTP_SERVER_FAILURES_TOTAL: &str = "rustfs_http_server_failures_total"; +const METRIC_HTTP_SERVER_ACTIVE_REQUESTS: &str = "rustfs_http_server_active_requests"; +const METRIC_HTTP_SERVER_REQUEST_DURATION_SECONDS: &str = "rustfs_http_server_request_duration_seconds"; +const METRIC_HTTP_SERVER_REQUEST_BODY_BYTES_TOTAL: &str = "rustfs_http_server_request_body_bytes_total"; +const METRIC_HTTP_SERVER_REQUEST_BODY_SIZE_BYTES: &str = "rustfs_http_server_request_body_size_bytes"; +const METRIC_HTTP_SERVER_RESPONSE_BODY_BYTES_TOTAL: &str = "rustfs_http_server_response_body_bytes_total"; +const METRIC_HTTP_SERVER_RESPONSE_BODY_SIZE_BYTES: &str = "rustfs_http_server_response_body_size_bytes"; + +static ACTIVE_HTTP_REQUESTS: AtomicU64 = AtomicU64::new(0); #[inline] fn request_method_label(method: &Method) -> &'static str { @@ -89,6 +96,32 @@ fn request_method_label(method: &Method) -> &'static str { } } +#[inline] +fn status_class_label(status: http::StatusCode) -> &'static str { + match status.as_u16() / 100 { + 1 => "1xx", + 2 => "2xx", + 3 => "3xx", + 4 => "4xx", + 5 => "5xx", + _ => "unknown", + } +} + +#[inline] +fn record_active_http_requests(delta: i64) { + let next = if delta >= 0 { + ACTIVE_HTTP_REQUESTS.fetch_add(delta as u64, Ordering::Relaxed) + delta as u64 + } else { + let decrement = (-delta) as u64; + ACTIVE_HTTP_REQUESTS + .fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| Some(current.saturating_sub(decrement))) + .unwrap_or_else(|current| current) + .saturating_sub(decrement) + }; + gauge!(METRIC_HTTP_SERVER_ACTIVE_REQUESTS).set(next as f64); +} + pub async fn start_http_server( config: &config::Config, readiness: Arc, @@ -739,28 +772,55 @@ fn process_connection( .on_request(|request: &HttpRequest<_>, span: &Span| { let _enter = span.enter(); debug!("http started method: {}, url path: {}", request.method(), request.uri().path()); + let method = request_method_label(request.method()); + record_active_http_requests(1); counter!( - METRIC_API_REQUESTS_TOTAL, - LABEL_REQUEST_METHOD => request_method_label(request.method()) + METRIC_HTTP_SERVER_REQUESTS_TOTAL, + LABEL_HTTP_METHOD => method ) .increment(1); - // Aggregate request body size for throughput monitoring (lightweight) + if let Some(cl) = request.headers().get("content-length") && let Some(len) = cl.to_str().ok().and_then(|s| s.parse::().ok()) { - counter!(METRIC_REQUEST_BODY_BYTES_TOTAL, "direction" => "request").increment(len); + counter!(METRIC_HTTP_SERVER_REQUEST_BODY_BYTES_TOTAL).increment(len); + histogram!( + METRIC_HTTP_SERVER_REQUEST_BODY_SIZE_BYTES, + LABEL_HTTP_METHOD => method + ) + .record(len as f64); } }) .on_response(|response: &Response<_>, latency: Duration, span: &Span| { span.record("status_code", tracing::field::display(response.status())); let _enter = span.enter(); - histogram!(METRIC_REQUEST_LATENCY_MS).record(latency.as_millis() as f64); + let status_class = status_class_label(response.status()); + record_active_http_requests(-1); + histogram!( + METRIC_HTTP_SERVER_REQUEST_DURATION_SECONDS, + LABEL_HTTP_STATUS_CLASS => status_class + ) + .record(latency.as_secs_f64()); + if response.status().is_client_error() || response.status().is_server_error() { + counter!( + METRIC_HTTP_SERVER_FAILURES_TOTAL, + LABEL_HTTP_STATUS_CLASS => status_class + ) + .increment(1); + } + if let Some(cl) = response.headers().get("content-length") + && let Some(len) = cl.to_str().ok().and_then(|s| s.parse::().ok()) + { + histogram!( + METRIC_HTTP_SERVER_RESPONSE_BODY_SIZE_BYTES, + LABEL_HTTP_STATUS_CLASS => status_class + ) + .record(len as f64); + } debug!("http response generated in {:?}", latency) }) .on_body_chunk(|chunk: &Bytes, latency: Duration, span: &Span| { - // Always track aggregate body bytes (lightweight counter, no debug logging) - counter!(METRIC_REQUEST_BODY_BYTES_TOTAL, "direction" => "response").increment(chunk.len() as u64); - histogram!(METRIC_REQUEST_BODY_LEN, "direction" => "response").record(chunk.len() as f64); + counter!(METRIC_HTTP_SERVER_RESPONSE_BODY_BYTES_TOTAL).increment(chunk.len() as u64); #[cfg(feature = "tracing-chunk-debug")] { let _enter = span.enter(); @@ -784,7 +844,12 @@ fn process_connection( }) .on_failure(|_error, latency: Duration, span: &Span| { let _enter = span.enter(); - counter!(METRIC_API_REQUESTS_FAILURE_TOTAL).increment(1); + record_active_http_requests(-1); + counter!( + METRIC_HTTP_SERVER_FAILURES_TOTAL, + LABEL_HTTP_STATUS_CLASS => "transport" + ) + .increment(1); debug!("http request failure error: {:?} in {:?}", _error, latency) }), ) @@ -1115,11 +1180,14 @@ mod tests { #[test] fn test_http_metric_names_and_labels_use_snake_case() { let metric_names = [ - METRIC_API_REQUESTS_TOTAL, - METRIC_API_REQUESTS_FAILURE_TOTAL, - METRIC_REQUEST_BODY_BYTES_TOTAL, - METRIC_REQUEST_LATENCY_MS, - METRIC_REQUEST_BODY_LEN, + METRIC_HTTP_SERVER_REQUESTS_TOTAL, + METRIC_HTTP_SERVER_FAILURES_TOTAL, + METRIC_HTTP_SERVER_ACTIVE_REQUESTS, + METRIC_HTTP_SERVER_REQUEST_DURATION_SECONDS, + METRIC_HTTP_SERVER_REQUEST_BODY_BYTES_TOTAL, + METRIC_HTTP_SERVER_REQUEST_BODY_SIZE_BYTES, + METRIC_HTTP_SERVER_RESPONSE_BODY_BYTES_TOTAL, + METRIC_HTTP_SERVER_RESPONSE_BODY_SIZE_BYTES, ]; for metric_name in metric_names { @@ -1127,7 +1195,8 @@ mod tests { assert!(!metric_name.contains('.')); } - assert_eq!(LABEL_REQUEST_METHOD, "request_method"); + assert_eq!(LABEL_HTTP_METHOD, "method"); + assert_eq!(LABEL_HTTP_STATUS_CLASS, "status_class"); } #[test] diff --git a/rustfs/src/storage/access.rs b/rustfs/src/storage/access.rs index 38610eaef..c691442a5 100644 --- a/rustfs/src/storage/access.rs +++ b/rustfs/src/storage/access.rs @@ -249,7 +249,7 @@ async fn get_or_fetch_object_tag_conditions( return Ok(cached.values.clone()); } - counter!("rustfs.object_tag_conditions.fetched", "op" => action_tag_metric_label(&action)).increment(1); + counter!("rustfs_object_tag_conditions_fetched_total", "op" => action_tag_metric_label(&action)).increment(1); let fetched = auth_fs() .get_object_tag_conditions_for_policy(bucket, object, version_id) .await?; @@ -268,7 +268,7 @@ async fn maybe_merge_object_tag_conditions( needs_tag: bool, ) -> S3Result<()> { if !needs_tag || bucket.is_empty() || object.is_empty() { - counter!("rustfs.object_tag_conditions.skipped", "op" => action_tag_metric_label(&action)).increment(1); + counter!("rustfs_object_tag_conditions_skipped_total", "op" => action_tag_metric_label(&action)).increment(1); return Ok(()); } diff --git a/rustfs/src/storage/backpressure.rs b/rustfs/src/storage/backpressure.rs index 2bb92445d..f6ea4be1a 100644 --- a/rustfs/src/storage/backpressure.rs +++ b/rustfs/src/storage/backpressure.rs @@ -280,7 +280,7 @@ impl BackpressurePipe { if usage >= threshold && !self.state.load(Ordering::Relaxed) { self.state.store(true, Ordering::Relaxed); - counter!("rustfs.backpressure.events.total", "state" => "high_watermark").increment(1); + counter!("rustfs_backpressure_events_total", "state" => "high_watermark").increment(1); warn!( buffer_usage = usage, @@ -300,7 +300,7 @@ impl BackpressurePipe { if usage <= threshold && self.state.load(Ordering::Relaxed) { self.state.store(false, Ordering::Relaxed); - counter!("rustfs.backpressure.events.total", "state" => "normal").increment(1); + counter!("rustfs_backpressure_events_total", "state" => "normal").increment(1); debug!( buffer_usage = usage, @@ -406,14 +406,14 @@ impl BackpressureMonitor { if usage >= high { if !self.in_high_watermark.swap(true, Ordering::Relaxed) { - counter!("rustfs.backpressure.events.total", "state" => "high_watermark").increment(1); + counter!("rustfs_backpressure_events_total", "state" => "high_watermark").increment(1); debug!(usage_percent = self.usage_percent() as u32, "Backpressure: entered high watermark"); } BackpressureState::HighWatermark } else if usage <= low { if self.in_high_watermark.swap(false, Ordering::Relaxed) { - counter!("rustfs.backpressure.events.total", "state" => "normal").increment(1); + counter!("rustfs_backpressure_events_total", "state" => "normal").increment(1); debug!(usage_percent = self.usage_percent() as u32, "Backpressure: returned to normal"); } diff --git a/rustfs/src/storage/concurrency/io_schedule.rs b/rustfs/src/storage/concurrency/io_schedule.rs index b5e58233a..3d7a271c5 100644 --- a/rustfs/src/storage/concurrency/io_schedule.rs +++ b/rustfs/src/storage/concurrency/io_schedule.rs @@ -1277,7 +1277,7 @@ pub fn get_concurrency_aware_buffer_size(file_size: i64, base_buffer_size: usize // Record concurrent request metrics { use metrics::gauge; - gauge!("rustfs.concurrent.get.requests").set(concurrent_requests as f64); + gauge!("rustfs_concurrent_get_requests").set(concurrent_requests as f64); } // For low concurrency, use the base buffer size for maximum throughput diff --git a/rustfs/src/storage/deadlock_detector.rs b/rustfs/src/storage/deadlock_detector.rs index bf86edbe0..cde095f9e 100644 --- a/rustfs/src/storage/deadlock_detector.rs +++ b/rustfs/src/storage/deadlock_detector.rs @@ -464,7 +464,7 @@ impl DeadlockDetector { if let Some(cycle) = Self::find_cycle(&wait_graph) { deadlocks_detected.fetch_add(1, Ordering::Relaxed); - counter!("rustfs.deadlock.detected.total").increment(1); + counter!("rustfs_deadlock_detected_total").increment(1); // Log detailed deadlock information error!( diff --git a/rustfs/src/storage/ecfs.rs b/rustfs/src/storage/ecfs.rs index fe5521cef..94bcceae2 100644 --- a/rustfs/src/storage/ecfs.rs +++ b/rustfs/src/storage/ecfs.rs @@ -31,6 +31,7 @@ use rustfs_ecstore::{ }, metadata_sys, object_lock::objectlock_sys::check_retention_for_modification, + replication::{GLOBAL_REPLICATION_STATS, ReplicationConfigurationExt}, tagging::{decode_tags, decode_tags_to_map, encode_tags}, utils::serialize, versioning::VersioningApi, @@ -72,6 +73,22 @@ impl FS { Self {} } + async fn replication_tagging_enabled(bucket: &str, object: &str) -> bool { + metadata_sys::get_replication_config(bucket) + .await + .map(|(cfg, _)| cfg.has_active_rules(object, true)) + .unwrap_or(false) + } + + async fn record_replication_tagging_metric(bucket: &str, object: &str, api: &str, is_err: bool) { + if !Self::replication_tagging_enabled(bucket, object).await { + return; + } + if let Some(stats) = GLOBAL_REPLICATION_STATS.get() { + stats.inc_proxy(bucket, api, is_err).await; + } + } + pub async fn get_object_tag_conditions_for_policy( &self, bucket: &str, @@ -374,7 +391,9 @@ impl S3 for FS { ..Default::default() }; - store.delete_object_tags(&bucket, &object, &opts).await.map_err(|e| { + let delete_tags_result = store.delete_object_tags(&bucket, &object, &opts).await; + Self::record_replication_tagging_metric(&bucket, &object, "DeleteObjectTagging", delete_tags_result.is_err()).await; + delete_tags_result.map_err(|e| { error!("Failed to delete object tags: {}", e); ApiError::from(e) })?; @@ -393,7 +412,7 @@ impl S3 for FS { } }; - counter!("rustfs.delete_object_tagging.success").increment(1); + counter!("rustfs_delete_object_tagging_success").increment(1); let event_version_id = version_id .as_deref() @@ -413,7 +432,7 @@ impl S3 for FS { let result = Ok(S3Response::new(DeleteObjectTaggingOutput { version_id })); let _ = helper.complete(&result); let duration = start_time.elapsed(); - histogram!("rustfs.object_tagging.operation.duration.seconds", "operation" => "delete").record(duration.as_secs_f64()); + histogram!("rustfs_object_tagging_operation_duration_seconds", "operation" => "delete").record(duration.as_secs_f64()); result } @@ -801,7 +820,9 @@ impl S3 for FS { ..Default::default() }; - let tags = store.get_object_tags(bucket, object, &opts).await.map_err(|e| { + let tags_result = store.get_object_tags(bucket, object, &opts).await; + Self::record_replication_tagging_metric(bucket, object, "GetObjectTagging", tags_result.is_err()).await; + let tags = tags_result.map_err(|e| { if is_err_object_not_found(&e) { error!("Object not found: {}", e); return s3_error!(NoSuchKey); @@ -813,9 +834,9 @@ impl S3 for FS { let tag_set = decode_tags(tags.as_str()); debug!("Decoded tag set: {:?}", tag_set); - counter!("rustfs.get_object_tagging.success").increment(1); + counter!("rustfs_get_object_tagging_success").increment(1); let duration = start_time.elapsed(); - histogram!("rustfs.object_tagging.operation.duration.seconds", "operation" => "get").record(duration.as_secs_f64()); + histogram!("rustfs_object_tagging_operation_duration_seconds", "operation" => "get").record(duration.as_secs_f64()); Ok(S3Response::new(GetObjectTaggingOutput { tag_set, version_id: req.input.version_id.clone(), @@ -1363,9 +1384,11 @@ impl S3 for FS { ..Default::default() }; - store.put_object_tags(&bucket, &object, &tags, &opts).await.map_err(|e| { + let put_tags_result = store.put_object_tags(&bucket, &object, &tags, &opts).await; + Self::record_replication_tagging_metric(&bucket, &object, "PutObjectTagging", put_tags_result.is_err()).await; + put_tags_result.map_err(|e| { error!("Failed to put object tags: {}", e); - counter!("rustfs.put_object_tagging.failure").increment(1); + counter!("rustfs_put_object_tagging_failure").increment(1); ApiError::from(e) })?; @@ -1383,7 +1406,7 @@ impl S3 for FS { } }; - counter!("rustfs.put_object_tagging.success").increment(1); + counter!("rustfs_put_object_tagging_success").increment(1); let event_version_id = req .input @@ -1407,7 +1430,7 @@ impl S3 for FS { })); let _ = helper.complete(&result); let duration = start_time.elapsed(); - histogram!("rustfs.object_tagging.operation.duration.seconds", "operation" => "put").record(duration.as_secs_f64()); + histogram!("rustfs_object_tagging_operation_duration_seconds", "operation" => "put").record(duration.as_secs_f64()); result } diff --git a/rustfs/src/storage/ecfs_extend.rs b/rustfs/src/storage/ecfs_extend.rs index 821be3435..c761f3e0c 100644 --- a/rustfs/src/storage/ecfs_extend.rs +++ b/rustfs/src/storage/ecfs_extend.rs @@ -190,12 +190,12 @@ pub(crate) fn get_buffer_size_opt_in(file_size: i64) -> usize { // Optional performance metrics collection for monitoring and optimization { use metrics::histogram; - histogram!("rustfs.buffer.size.bytes").record(buffer_size as f64); - counter!("rustfs.buffer.size.selections").increment(1); + histogram!("rustfs_buffer_size_bytes").record(buffer_size as f64); + counter!("rustfs_buffer_size_selections_total").increment(1); - if file_size >= 0 { + if file_size > 0 { let ratio = buffer_size as f64 / file_size as f64; - histogram!("rustfs.buffer.to.file.ratio").record(ratio); + histogram!("rustfs_buffer_to_file_ratio").record(ratio); } } diff --git a/rustfs/src/storage/lock_optimizer.rs b/rustfs/src/storage/lock_optimizer.rs index e5bbfc345..c31b2e0df 100644 --- a/rustfs/src/storage/lock_optimizer.rs +++ b/rustfs/src/storage/lock_optimizer.rs @@ -223,7 +223,7 @@ impl OptimizedLockGuard { self.stats.record_early_release(hold_time); - histogram!("rustfs.lock.hold.duration.seconds").record(hold_time.as_secs_f64()); + histogram!("rustfs_lock_hold_duration_seconds").record(hold_time.as_secs_f64()); debug!( resource = %self.resource, @@ -247,7 +247,7 @@ impl Drop for OptimizedLockGuard { self.stats.record_early_release(hold_time); - histogram!("rustfs.lock.hold.duration.seconds").record(hold_time.as_secs_f64()); + histogram!("rustfs_lock_hold_duration_seconds").record(hold_time.as_secs_f64()); debug!( resource = %self.resource, diff --git a/rustfs/src/storage/request_context.rs b/rustfs/src/storage/request_context.rs index a2a088cf8..b37a35fc4 100644 --- a/rustfs/src/storage/request_context.rs +++ b/rustfs/src/storage/request_context.rs @@ -90,7 +90,7 @@ impl RequestContext { pub fn fallback() -> Self { let trace_ctx = current_trace_context_ids(); let id = build_fallback_request_id(trace_ctx.as_ref()); - counter!("rustfs.log.chain.fallback_request_id.total", "source" => "request_context_fallback").increment(1); + counter!("rustfs_log_chain_fallback_request_id_total", "source" => "request_context_fallback").increment(1); Self { request_id: id.clone(), x_amz_request_id: id, @@ -138,7 +138,7 @@ pub fn extract_request_id_from_headers(headers: &HeaderMap) -> String { .unwrap_or_else(generate_fallback_request_id); if !headers.contains_key(REQUEST_ID_HEADER) && !headers.contains_key(AMZ_REQUEST_ID) { - counter!("rustfs.log.chain.fallback_request_id.total", "source" => "headers_missing").increment(1); + counter!("rustfs_log_chain_fallback_request_id_total", "source" => "headers_missing").increment(1); } request_id diff --git a/scripts/run.sh b/scripts/run.sh index 2000c444a..de7435ae0 100755 --- a/scripts/run.sh +++ b/scripts/run.sh @@ -43,7 +43,14 @@ fi # export RUSTFS_ERASURE_SET_DRIVE_COUNT=5 -# export RUSTFS_STORAGE_CLASS_INLINE_BLOCK="512 KB"√ +# export RUSTFS_STORAGE_CLASS_INLINE_BLOCK="512 KB" + +# This script provisions multiple local export directories on the same disk. +# Default the bypass only for this local layout, while still allowing callers +# to override it explicitly through the environment. +if [ -z "${RUSTFS_UNSAFE_BYPASS_DISK_CHECK+x}" ] && [ -z "${MINIO_CI+x}" ]; then + export RUSTFS_UNSAFE_BYPASS_DISK_CHECK=true +fi export RUSTFS_VOLUMES="./target/volume/test{1...4}" # export RUSTFS_VOLUMES="./target/volume/test"