From b3b0d892bbfdf0c57fe77d6f670f7a74a926dd8b Mon Sep 17 00:00:00 2001 From: cxymds Date: Wed, 9 Sep 2026 19:00:20 +0800 Subject: [PATCH] fix(metrics): prevent duplicate drive reporting across nodes (#7582) Co-authored-by: houseme --- ...se.cluster.local-build.profiling-amd64.yml | 4 + .../docker-compose.cluster.local-build.yml | 4 + .../compose/docker-compose.observability.yaml | 4 + .docker/observability/README.md | 4 + .../docker-compose-example-for-rustfs.yml | 1 + .../grafana/dashboards/rustfs.json | 146 +++-- .../prometheus-rules/rustfs-storage.yml | 575 ++++++++++++++++++ .../tests/storage-rules.test.yml | 189 ++++++ crates/e2e_test/README.md | 2 + crates/e2e_test/src/lib.rs | 3 + .../src/storage_metric_ownership_test.rs | 335 ++++++++++ .../src/metrics/collectors/cluster_drive.rs | 150 +++++ crates/obs/src/metrics/collectors/mod.rs | 1 + .../src/metrics/collectors/system_drive.rs | 102 +++- crates/obs/src/metrics/mod.rs | 1 + crates/obs/src/metrics/report.rs | 2 +- crates/obs/src/metrics/scheduler.rs | 340 ++--------- crates/obs/src/metrics/schema/system_drive.rs | 32 +- crates/obs/src/metrics/stats_collector.rs | 129 +++- crates/obs/src/metrics/storage_snapshot.rs | 407 +++++++++++++ docker-compose.yml | 1 + docs/README.md | 3 + docs/operations/storage-metrics.md | 142 +++++ 23 files changed, 2188 insertions(+), 389 deletions(-) create mode 100644 .docker/observability/prometheus-rules/rustfs-storage.yml create mode 100644 .docker/observability/tests/storage-rules.test.yml create mode 100644 crates/e2e_test/src/storage_metric_ownership_test.rs create mode 100644 crates/obs/src/metrics/collectors/cluster_drive.rs create mode 100644 crates/obs/src/metrics/storage_snapshot.rs create mode 100644 docs/operations/storage-metrics.md diff --git a/.docker/compose/docker-compose.cluster.local-build.profiling-amd64.yml b/.docker/compose/docker-compose.cluster.local-build.profiling-amd64.yml index ecaf98874..7657e60b0 100644 --- a/.docker/compose/docker-compose.cluster.local-build.profiling-amd64.yml +++ b/.docker/compose/docker-compose.cluster.local-build.profiling-amd64.yml @@ -43,6 +43,7 @@ services: - RUSTFS_ACCESS_KEY=${RUSTFS_ACCESS_KEY:-rustfs-cluster-admin} - RUSTFS_SECRET_KEY=${RUSTFS_SECRET_KEY:-rustfs-cluster-secret} - RUSTFS_OBS_ENDPOINT=${RUSTFS_OBS_ENDPOINT:-http://host.docker.internal:4318} + - OTEL_RESOURCE_ATTRIBUTES=${OTEL_RESOURCE_ATTRIBUTES:-rustfs.cluster.id=rustfs-dev} # `info` is enough for startup logs/metrics. Use `debug` if Tempo/Jaeger # should show richer nested spans during request-path verification. - RUSTFS_OBS_LOGGER_LEVEL=${RUSTFS_OBS_LOGGER_LEVEL:-info} @@ -89,6 +90,7 @@ services: - RUSTFS_ACCESS_KEY=${RUSTFS_ACCESS_KEY:-rustfs-cluster-admin} - RUSTFS_SECRET_KEY=${RUSTFS_SECRET_KEY:-rustfs-cluster-secret} - RUSTFS_OBS_ENDPOINT=${RUSTFS_OBS_ENDPOINT:-http://host.docker.internal:4318} + - OTEL_RESOURCE_ATTRIBUTES=${OTEL_RESOURCE_ATTRIBUTES:-rustfs.cluster.id=rustfs-dev} # `info` is enough for startup logs/metrics. Use `debug` if Tempo/Jaeger # should show richer nested spans during request-path verification. - RUSTFS_OBS_LOGGER_LEVEL=${RUSTFS_OBS_LOGGER_LEVEL:-info} @@ -135,6 +137,7 @@ services: - RUSTFS_ACCESS_KEY=${RUSTFS_ACCESS_KEY:-rustfs-cluster-admin} - RUSTFS_SECRET_KEY=${RUSTFS_SECRET_KEY:-rustfs-cluster-secret} - RUSTFS_OBS_ENDPOINT=${RUSTFS_OBS_ENDPOINT:-http://host.docker.internal:4318} + - OTEL_RESOURCE_ATTRIBUTES=${OTEL_RESOURCE_ATTRIBUTES:-rustfs.cluster.id=rustfs-dev} # `info` is enough for startup logs/metrics. Use `debug` if Tempo/Jaeger # should show richer nested spans during request-path verification. - RUSTFS_OBS_LOGGER_LEVEL=${RUSTFS_OBS_LOGGER_LEVEL:-info} @@ -181,6 +184,7 @@ services: - RUSTFS_ACCESS_KEY=${RUSTFS_ACCESS_KEY:-rustfs-cluster-admin} - RUSTFS_SECRET_KEY=${RUSTFS_SECRET_KEY:-rustfs-cluster-secret} - RUSTFS_OBS_ENDPOINT=${RUSTFS_OBS_ENDPOINT:-http://host.docker.internal:4318} + - OTEL_RESOURCE_ATTRIBUTES=${OTEL_RESOURCE_ATTRIBUTES:-rustfs.cluster.id=rustfs-dev} # `info` is enough for startup logs/metrics. Use `debug` if Tempo/Jaeger # should show richer nested spans during request-path verification. - RUSTFS_OBS_LOGGER_LEVEL=${RUSTFS_OBS_LOGGER_LEVEL:-info} diff --git a/.docker/compose/docker-compose.cluster.local-build.yml b/.docker/compose/docker-compose.cluster.local-build.yml index 80af50c2b..76dcce105 100644 --- a/.docker/compose/docker-compose.cluster.local-build.yml +++ b/.docker/compose/docker-compose.cluster.local-build.yml @@ -26,6 +26,7 @@ services: - RUSTFS_ACCESS_KEY=${RUSTFS_ACCESS_KEY:-rustfsadmin-local} - RUSTFS_SECRET_KEY=${RUSTFS_SECRET_KEY:-rustfssecret-local} - RUSTFS_OBS_ENDPOINT=${RUSTFS_OBS_ENDPOINT:-http://host.docker.internal:4318} + - OTEL_RESOURCE_ATTRIBUTES=${OTEL_RESOURCE_ATTRIBUTES:-rustfs.cluster.id=rustfs-dev} - RUSTFS_OBS_LOGGER_LEVEL=${RUSTFS_OBS_LOGGER_LEVEL:-info} - RUSTFS_OBS_USE_STDOUT=${RUSTFS_OBS_USE_STDOUT:-false} - RUSTFS_OBS_LOG_STDOUT_ENABLED=${RUSTFS_OBS_LOG_STDOUT_ENABLED:-false} @@ -72,6 +73,7 @@ services: - RUSTFS_ACCESS_KEY=${RUSTFS_ACCESS_KEY:-rustfsadmin-local} - RUSTFS_SECRET_KEY=${RUSTFS_SECRET_KEY:-rustfssecret-local} - RUSTFS_OBS_ENDPOINT=${RUSTFS_OBS_ENDPOINT:-http://host.docker.internal:4318} + - OTEL_RESOURCE_ATTRIBUTES=${OTEL_RESOURCE_ATTRIBUTES:-rustfs.cluster.id=rustfs-dev} - RUSTFS_OBS_LOGGER_LEVEL=${RUSTFS_OBS_LOGGER_LEVEL:-info} - RUSTFS_OBS_USE_STDOUT=${RUSTFS_OBS_USE_STDOUT:-false} - RUSTFS_OBS_LOG_STDOUT_ENABLED=${RUSTFS_OBS_LOG_STDOUT_ENABLED:-false} @@ -118,6 +120,7 @@ services: - RUSTFS_ACCESS_KEY=${RUSTFS_ACCESS_KEY:-rustfsadmin-local} - RUSTFS_SECRET_KEY=${RUSTFS_SECRET_KEY:-rustfssecret-local} - RUSTFS_OBS_ENDPOINT=${RUSTFS_OBS_ENDPOINT:-http://host.docker.internal:4318} + - OTEL_RESOURCE_ATTRIBUTES=${OTEL_RESOURCE_ATTRIBUTES:-rustfs.cluster.id=rustfs-dev} - RUSTFS_OBS_LOGGER_LEVEL=${RUSTFS_OBS_LOGGER_LEVEL:-info} - RUSTFS_OBS_USE_STDOUT=${RUSTFS_OBS_USE_STDOUT:-false} - RUSTFS_OBS_LOG_STDOUT_ENABLED=${RUSTFS_OBS_LOG_STDOUT_ENABLED:-false} @@ -164,6 +167,7 @@ services: - RUSTFS_ACCESS_KEY=${RUSTFS_ACCESS_KEY:-rustfsadmin-local} - RUSTFS_SECRET_KEY=${RUSTFS_SECRET_KEY:-rustfssecret-local} - RUSTFS_OBS_ENDPOINT=${RUSTFS_OBS_ENDPOINT:-http://host.docker.internal:4318} + - OTEL_RESOURCE_ATTRIBUTES=${OTEL_RESOURCE_ATTRIBUTES:-rustfs.cluster.id=rustfs-dev} - RUSTFS_OBS_LOGGER_LEVEL=${RUSTFS_OBS_LOGGER_LEVEL:-info} - RUSTFS_OBS_USE_STDOUT=${RUSTFS_OBS_USE_STDOUT:-false} - RUSTFS_OBS_LOG_STDOUT_ENABLED=${RUSTFS_OBS_LOG_STDOUT_ENABLED:-false} diff --git a/.docker/compose/docker-compose.observability.yaml b/.docker/compose/docker-compose.observability.yaml index 03b4d765d..a52065ce6 100644 --- a/.docker/compose/docker-compose.observability.yaml +++ b/.docker/compose/docker-compose.observability.yaml @@ -201,6 +201,7 @@ services: - RUSTFS_ADDRESS=:9000 - RUSTFS_CONSOLE_ENABLE=true - RUSTFS_OBS_ENDPOINT=http://otel-collector:4318 + - OTEL_RESOURCE_ATTRIBUTES=${OTEL_RESOURCE_ATTRIBUTES:-rustfs.cluster.id=rustfs-dev} - RUSTFS_OBS_LOGGER_LEVEL=debug platform: linux/amd64 ports: @@ -220,6 +221,7 @@ services: - RUSTFS_ADDRESS=:9000 - RUSTFS_CONSOLE_ENABLE=true - RUSTFS_OBS_ENDPOINT=http://otel-collector:4318 + - OTEL_RESOURCE_ATTRIBUTES=${OTEL_RESOURCE_ATTRIBUTES:-rustfs.cluster.id=rustfs-dev} - RUSTFS_OBS_LOGGER_LEVEL=debug platform: linux/amd64 ports: @@ -239,6 +241,7 @@ services: - RUSTFS_ADDRESS=:9000 - RUSTFS_CONSOLE_ENABLE=true - RUSTFS_OBS_ENDPOINT=http://otel-collector:4318 + - OTEL_RESOURCE_ATTRIBUTES=${OTEL_RESOURCE_ATTRIBUTES:-rustfs.cluster.id=rustfs-dev} - RUSTFS_OBS_LOGGER_LEVEL=debug platform: linux/amd64 ports: @@ -258,6 +261,7 @@ services: - RUSTFS_ADDRESS=:9000 - RUSTFS_CONSOLE_ENABLE=true - RUSTFS_OBS_ENDPOINT=http://otel-collector:4318 + - OTEL_RESOURCE_ATTRIBUTES=${OTEL_RESOURCE_ATTRIBUTES:-rustfs.cluster.id=rustfs-dev} - RUSTFS_OBS_LOGGER_LEVEL=debug platform: linux/amd64 ports: diff --git a/.docker/observability/README.md b/.docker/observability/README.md index 8781c1b4c..d9b61b521 100644 --- a/.docker/observability/README.md +++ b/.docker/observability/README.md @@ -47,6 +47,10 @@ Three pre-built Grafana dashboards are included for monitoring RustFS GET perfor | **GET Resource Impact** | `grafana-get-resource-impact.json` | Monitors resource usage: concurrent requests, IO queue utilization, disk permit wait, RSS trend | | **Object Data Cache** | `grafana-object-data-cache.json` | Monitors the GET body cache (`rustfs_object_data_cache_*`): hit ratio, lookup/plan/fill outcomes, fill duration quantiles, hit vs fill throughput, entries/weighted bytes, inflight fills, memory-pressure skips, invalidations, and size-class breakdowns | +### Storage Metrics + +Storage panels require `prometheus-rules/rustfs-storage.yml` and the cluster resource attribute. See the [storage metrics guide](../../docs/operations/storage-metrics.md) for ownership, observer selection, freshness, and rolling upgrades. + ### Prometheus Alert Rules The file `prometheus-rules/rustfs-get-optimization-alerts.yaml` contains pre-configured alerting rules: diff --git a/.docker/observability/docker-compose-example-for-rustfs.yml b/.docker/observability/docker-compose-example-for-rustfs.yml index 1d8ce46bb..f2b92c96c 100644 --- a/.docker/observability/docker-compose-example-for-rustfs.yml +++ b/.docker/observability/docker-compose-example-for-rustfs.yml @@ -32,6 +32,7 @@ services: - RUSTFS_SECRET_KEY=rustfsadmin - RUSTFS_OBS_LOGGER_LEVEL=info - RUSTFS_OBS_ENDPOINT=http://otel-collector:4318 + - OTEL_RESOURCE_ATTRIBUTES=${OTEL_RESOURCE_ATTRIBUTES:-rustfs.cluster.id=rustfs-dev} - RUSTFS_OBS_PROFILING_ENDPOINT=http://pyroscope:4040 volumes: - rustfs-data:/data/rustfs diff --git a/.docker/observability/grafana/dashboards/rustfs.json b/.docker/observability/grafana/dashboards/rustfs.json index 5a72306c3..15fbdc34b 100644 --- a/.docker/observability/grafana/dashboards/rustfs.json +++ b/.docker/observability/grafana/dashboards/rustfs.json @@ -223,14 +223,15 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "sum(rustfs_cluster_buckets_total{job=~\"$job\"})", + "expr": "rustfs:storage:current{source_metric=\"rustfs_cluster_buckets_total\",collection_scope=\"cluster\",rustfs_cluster_id=\"$storage_cluster\",job=~\"$job\",observer=\"$storage_observer\"}", "legendFormat": "__auto", "range": true, "refId": "A" } ], "title": "Total Buckets", - "type": "stat" + "type": "stat", + "description": "Storage snapshot rules are required. Local details come from each drive owner. Global values use the selected fresh cluster observer. Missing or expired observations show no data; select another observer if needed." }, { "datasource": { @@ -289,14 +290,15 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "sum(rustfs_cluster_objects_total{job=~\"$job\"})", + "expr": "rustfs:storage:current{source_metric=\"rustfs_cluster_objects_total\",collection_scope=\"cluster\",rustfs_cluster_id=\"$storage_cluster\",job=~\"$job\",observer=\"$storage_observer\"}", "legendFormat": "__auto", "range": true, "refId": "A" } ], "title": "Total Objects", - "type": "stat" + "type": "stat", + "description": "Storage snapshot rules are required. Local details come from each drive owner. Global values use the selected fresh cluster observer. Missing or expired observations show no data; select another observer if needed." }, { "datasource": { @@ -427,7 +429,7 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "sum(rustfs_cluster_capacity_used_bytes{job=~\"$job\"})", + "expr": "rustfs:storage:current{source_metric=\"rustfs_cluster_capacity_used_bytes\",collection_scope=\"cluster\",rustfs_cluster_id=\"$storage_cluster\",job=~\"$job\",observer=\"$storage_observer\"}", "legendFormat": "Used", "range": true, "refId": "A" @@ -438,7 +440,7 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "sum(rustfs_cluster_capacity_raw_total_bytes{job=~\"$job\"})", + "expr": "rustfs:storage:current{source_metric=\"rustfs_cluster_capacity_raw_total_bytes\",collection_scope=\"cluster\",rustfs_cluster_id=\"$storage_cluster\",job=~\"$job\",observer=\"$storage_observer\"}", "hide": false, "legendFormat": "Total", "range": true, @@ -450,7 +452,7 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "sum(rustfs_cluster_capacity_used_bytes{job=~\"$job\"}) / sum(rustfs_cluster_capacity_raw_total_bytes{job=~\"$job\"})", + "expr": "rustfs:storage:current{source_metric=\"rustfs_cluster_capacity_used_bytes\",collection_scope=\"cluster\",rustfs_cluster_id=\"$storage_cluster\",job=~\"$job\",observer=\"$storage_observer\"} / ignoring(source_metric) rustfs:storage:current{source_metric=\"rustfs_cluster_capacity_raw_total_bytes\",collection_scope=\"cluster\",rustfs_cluster_id=\"$storage_cluster\",job=~\"$job\",observer=\"$storage_observer\"}", "hide": false, "instant": false, "legendFormat": "Percent", @@ -459,7 +461,8 @@ } ], "title": "Capacity", - "type": "stat" + "type": "stat", + "description": "Storage snapshot rules are required. Local details come from each drive owner. Global values use the selected fresh cluster observer. Missing or expired observations show no data; select another observer if needed." }, { "datasource": { @@ -525,7 +528,7 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "sum(rustfs_cluster_capacity_stale_drives{job=~\"$job\"})", + "expr": "rustfs:storage:current{source_metric=\"rustfs_cluster_capacity_stale_drives\",collection_scope=\"cluster\",rustfs_cluster_id=\"$storage_cluster\",job=~\"$job\",observer=\"$storage_observer\"}", "legendFormat": "Stale Drives", "range": true, "refId": "A" @@ -536,14 +539,15 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "sum(rustfs_cluster_capacity_missing_drives{job=~\"$job\"})", + "expr": "rustfs:storage:current{source_metric=\"rustfs_cluster_capacity_missing_drives\",collection_scope=\"cluster\",rustfs_cluster_id=\"$storage_cluster\",job=~\"$job\",observer=\"$storage_observer\"}", "legendFormat": "Missing Drives", "range": true, "refId": "B" } ], "title": "Capacity Observation", - "type": "stat" + "type": "stat", + "description": "Storage snapshot rules are required. Local details come from each drive owner. Global values use the selected fresh cluster observer. Missing or expired observations show no data; select another observer if needed." }, { "datasource": { @@ -1989,8 +1993,8 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "sum by (drive) (rustfs_system_drive_used_bytes{job=~\"$job\", drive=~\"$drive\"})", - "legendFormat": "{{drive}} (bytes)", + "expr": "rustfs:storage:current{source_metric=\"rustfs_system_drive_used_bytes\",collection_scope=\"local\",rustfs_cluster_id=\"$storage_cluster\",job=~\"$job\",server=~\"$server\",drive=~\"$drive\"}", + "legendFormat": "{{server}} | {{drive}} (bytes)", "range": true, "refId": "A" }, @@ -2000,16 +2004,17 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "sum by (drive) (rustfs_system_drive_used_bytes{job=~\"$job\", drive=~\"$drive\"}) / sum by (drive)(rustfs_system_drive_total_bytes{job=~\"$job\", drive=~\"$drive\"})", + "expr": "rustfs:storage:current{source_metric=\"rustfs_system_drive_used_bytes\",collection_scope=\"local\",rustfs_cluster_id=\"$storage_cluster\",job=~\"$job\",server=~\"$server\",drive=~\"$drive\"} / ignoring(source_metric) rustfs:storage:current{source_metric=\"rustfs_system_drive_total_bytes\",collection_scope=\"local\",rustfs_cluster_id=\"$storage_cluster\",job=~\"$job\",server=~\"$server\",drive=~\"$drive\"}", "hide": false, "instant": false, - "legendFormat": "{{drive}} (percent)", + "legendFormat": "{{server}} | {{drive}} (percent)", "range": true, "refId": "B" } ], "title": "System Drive Usage", - "type": "timeseries" + "type": "timeseries", + "description": "Storage snapshot rules are required. Local details come from each drive owner. Global values use the selected fresh cluster observer. Missing or expired observations show no data; select another observer if needed." }, { "datasource": { @@ -2097,14 +2102,15 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "max by (drive) (rustfs_system_drive_capacity_observation_age_seconds{job=~\"$job\", drive=~\"$drive\"})", - "legendFormat": "{{drive}}", + "expr": "rustfs:storage:current{source_metric=\"rustfs_system_drive_capacity_observation_age_seconds\",collection_scope=\"local\",rustfs_cluster_id=\"$storage_cluster\",job=~\"$job\",server=~\"$server\",drive=~\"$drive\"}", + "legendFormat": "{{server}} | {{drive}}", "range": true, "refId": "A" } ], "title": "Drive Capacity Observation Age", - "type": "timeseries" + "type": "timeseries", + "description": "Storage snapshot rules are required. Local details come from each drive owner. Global values use the selected fresh cluster observer. Missing or expired observations show no data; select another observer if needed." }, { "datasource": { @@ -2190,8 +2196,8 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "max by (drive) (rustfs_system_drive_capacity_observation_state{job=~\"$job\", drive=~\"$drive\", state=\"stale\"})", - "legendFormat": "{{drive}} stale", + "expr": "rustfs:storage:current{source_metric=\"rustfs_system_drive_capacity_observation_state\",collection_scope=\"local\",rustfs_cluster_id=\"$storage_cluster\",job=~\"$job\",server=~\"$server\",drive=~\"$drive\",state=\"stale\"}", + "legendFormat": "{{server}} | {{drive}} stale", "range": true, "refId": "A" }, @@ -2201,14 +2207,15 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "max by (drive) (rustfs_system_drive_capacity_observation_state{job=~\"$job\", drive=~\"$drive\", state=\"missing\"})", - "legendFormat": "{{drive}} missing", + "expr": "rustfs:storage:current{source_metric=\"rustfs_system_drive_capacity_observation_state\",collection_scope=\"local\",rustfs_cluster_id=\"$storage_cluster\",job=~\"$job\",server=~\"$server\",drive=~\"$drive\",state=\"missing\"}", + "legendFormat": "{{server}} | {{drive}} missing", "range": true, "refId": "B" } ], "title": "Drive Capacity Observation State", - "type": "timeseries" + "type": "timeseries", + "description": "Storage snapshot rules are required. Local details come from each drive owner. Global values use the selected fresh cluster observer. Missing or expired observations show no data; select another observer if needed." }, { "datasource": { @@ -4551,7 +4558,7 @@ "index": 0, "text": "INACTIVE" }, - "to": 1e-9 + "to": 1e-09 }, "type": "range" } @@ -6563,7 +6570,7 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "rustfs_cluster_health_drives_online_count{job=~\"$job\"}", + "expr": "rustfs:storage:current{source_metric=\"rustfs_cluster_health_drives_online_count\",collection_scope=\"cluster\",rustfs_cluster_id=\"$storage_cluster\",job=~\"$job\",observer=\"$storage_observer\"}", "legendFormat": "online - {{job}}", "range": true, "refId": "A" @@ -6574,7 +6581,7 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "rustfs_cluster_health_drives_offline_count{job=~\"$job\"}", + "expr": "rustfs:storage:current{source_metric=\"rustfs_cluster_health_drives_offline_count\",collection_scope=\"cluster\",rustfs_cluster_id=\"$storage_cluster\",job=~\"$job\",observer=\"$storage_observer\"}", "legendFormat": "offline - {{job}}", "range": true, "refId": "B" @@ -6585,14 +6592,15 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "rustfs_cluster_health_drives_count{job=~\"$job\"}", + "expr": "rustfs:storage:current{source_metric=\"rustfs_cluster_health_drives_count\",collection_scope=\"cluster\",rustfs_cluster_id=\"$storage_cluster\",job=~\"$job\",observer=\"$storage_observer\"}", "legendFormat": "total - {{job}}", "range": true, "refId": "C" } ], "title": "Cluster Drive Health Counts", - "type": "timeseries" + "type": "timeseries", + "description": "Storage snapshot rules are required. Local details come from each drive owner. Global values use the selected fresh cluster observer. Missing or expired observations show no data; select another observer if needed." }, { "collapsed": false, @@ -8554,14 +8562,15 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "{__name__=~\"rustfs_system_drive_.*\",job=~\"$job\",drive=~\"$drive\"}", - "legendFormat": "{{__name__}} | {{drive}}", + "expr": "rustfs:storage:current{source_metric=~\"rustfs_system_drive_.*\",collection_scope=\"local\",rustfs_cluster_id=\"$storage_cluster\",job=~\"$job\",server=~\"$server\",drive=~\"$drive\"}", + "legendFormat": "{{server}} | {{source_metric}} | {{drive}}", "range": true, "refId": "A" } ], "title": "System Drive (All)", - "type": "timeseries" + "type": "timeseries", + "description": "Storage snapshot rules are required. Local details come from each drive owner. Global values use the selected fresh cluster observer. Missing or expired observations show no data; select another observer if needed." }, { "datasource": { @@ -8942,14 +8951,15 @@ "uid": "${datasource}" }, "editorMode": "code", - "expr": "{__name__=~\"rustfs_cluster_erasure_set_.*\",job=~\"$job\"}", - "legendFormat": "{{__name__}}", + "expr": "rustfs:storage:current{source_metric=~\"rustfs_cluster_erasure_set_.*\",collection_scope=\"cluster\",rustfs_cluster_id=\"$storage_cluster\",job=~\"$job\",observer=\"$storage_observer\"}", + "legendFormat": "{{source_metric}}", "range": true, "refId": "A" } ], "title": "Cluster Erasure Set (All)", - "type": "timeseries" + "type": "timeseries", + "description": "Storage snapshot rules are required. Local details come from each drive owner. Global values use the selected fresh cluster observer. Missing or expired observations show no data; select another observer if needed." }, { "datasource": { @@ -11716,7 +11726,7 @@ "editorMode": "code", "range": true, "refId": "A", - "expr": "max by (server, drive, pool_index, set_index, drive_index, state) (rustfs_system_drive_runtime_state{job=~\"$job\",server=~\"$server\",drive=~\"$drive\"})", + "expr": "rustfs:storage:current{source_metric=\"rustfs_cluster_drive_runtime_state\",collection_scope=\"cluster\",rustfs_cluster_id=\"$storage_cluster\",job=~\"$job\",observer=\"$storage_observer\",drive=~\"$drive\"}", "legendFormat": "{{server}} | {{drive}} | p{{pool_index}}/s{{set_index}}/d{{drive_index}} | {{state}}" }, { @@ -11727,12 +11737,13 @@ "editorMode": "code", "range": true, "refId": "B", - "expr": "max by (server, drive, pool_index, set_index, drive_index) (rustfs_system_drive_offline_duration_seconds{job=~\"$job\",server=~\"$server\",drive=~\"$drive\"})", + "expr": "rustfs:storage:current{source_metric=\"rustfs_cluster_drive_offline_duration_seconds\",collection_scope=\"cluster\",rustfs_cluster_id=\"$storage_cluster\",job=~\"$job\",observer=\"$storage_observer\",drive=~\"$drive\"}", "legendFormat": "{{server}} | {{drive}} | offline seconds" } ], - "title": "Drive Runtime State and Offline Duration", - "type": "timeseries" + "title": "Observed Cluster Drive State and Offline Duration", + "type": "timeseries", + "description": "Storage snapshot rules are required. Local details come from each drive owner. Global values use the selected fresh cluster observer. Missing or expired observations show no data; select another observer if needed." }, { "datasource": { @@ -11825,12 +11836,13 @@ "editorMode": "code", "range": true, "refId": "A", - "expr": "sum by (server, drive, pool_index, set_index, drive_index, api) (rate(rustfs_system_drive_api_calls_total{job=~\"$job\",server=~\"$server\",drive=~\"$drive\",api=~\"$drive_api\"}[$__rate_interval]))", + "expr": "rate(rustfs_system_drive_api_calls_total{collection_scope=\"local\",rustfs_cluster_id=\"$storage_cluster\",job=~\"$job\",server=~\"$server\",drive=~\"$drive\",api=~\"$drive_api\"}[$__rate_interval]) and ignoring(source_metric) rustfs:storage:current{source_metric=\"rustfs_system_drive_api_calls_total\",collection_scope=\"local\",rustfs_cluster_id=\"$storage_cluster\",job=~\"$job\",server=~\"$server\",drive=~\"$drive\",api=~\"$drive_api\"}", "legendFormat": "{{server}} | {{drive}} | p{{pool_index}}/s{{set_index}}/d{{drive_index}} | {{api}}" } ], "title": "Drive API Calls by Operation", - "type": "timeseries" + "type": "timeseries", + "description": "Storage snapshot rules are required. Local details come from each drive owner. Global values use the selected fresh cluster observer. Missing or expired observations show no data; select another observer if needed." }, { "datasource": { @@ -12376,6 +12388,50 @@ "sort": 1, "type": "query" }, + { + "current": {}, + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "definition": "label_values(rustfs:storage_snapshot:fresh, rustfs_cluster_id)", + "includeAll": false, + "label": "Storage cluster", + "multi": false, + "name": "storage_cluster", + "options": [], + "query": { + "qryType": 1, + "query": "label_values(rustfs:storage_snapshot:fresh, rustfs_cluster_id)", + "refId": "PrometheusVariableQueryEditor-storage_cluster" + }, + "refresh": 2, + "regex": "", + "sort": 1, + "type": "query" + }, + { + "current": {}, + "datasource": { + "type": "prometheus", + "uid": "${datasource}" + }, + "definition": "query_result(rustfs:storage_snapshot:fresh{collection_scope=\"cluster\",rustfs_cluster_id=\"$storage_cluster\",job=~\"$job\"})", + "includeAll": false, + "label": "Cluster observer", + "multi": false, + "name": "storage_observer", + "options": [], + "query": { + "qryType": 3, + "query": "query_result(rustfs:storage_snapshot:fresh{collection_scope=\"cluster\",rustfs_cluster_id=\"$storage_cluster\",job=~\"$job\"})", + "refId": "PrometheusVariableQueryEditor-storage_observer" + }, + "refresh": 2, + "regex": "/observer=\"([^\"]+)\"/", + "sort": 1, + "type": "query" + }, { "allValue": ".*", "current": { @@ -12386,7 +12442,7 @@ "type": "prometheus", "uid": "${datasource}" }, - "definition": "label_values(rustfs_system_drive_api_calls_total,api)", + "definition": "label_values(rustfs:storage:current{source_metric=\"rustfs_system_drive_api_calls_total\",collection_scope=\"local\",rustfs_cluster_id=\"$storage_cluster\"}, api)", "includeAll": true, "label": "Drive API", "multi": true, @@ -12394,7 +12450,7 @@ "options": [], "query": { "qryType": 1, - "query": "label_values(rustfs_system_drive_api_calls_total,api)", + "query": "label_values(rustfs:storage:current{source_metric=\"rustfs_system_drive_api_calls_total\",collection_scope=\"local\",rustfs_cluster_id=\"$storage_cluster\"}, api)", "refId": "PrometheusVariableQueryEditor-drive_api" }, "refresh": 2, @@ -12455,7 +12511,7 @@ "text": "All", "value": "$__all" }, - "definition": "label_values(rustfs_system_drive_used_bytes,drive)", + "definition": "label_values(rustfs:storage:current{source_metric=\"rustfs_system_drive_used_bytes\",collection_scope=\"local\",rustfs_cluster_id=\"$storage_cluster\"}, drive)", "includeAll": true, "label": "Drive", "multi": true, @@ -12463,7 +12519,7 @@ "options": [], "query": { "qryType": 1, - "query": "label_values(rustfs_system_drive_used_bytes,drive)", + "query": "label_values(rustfs:storage:current{source_metric=\"rustfs_system_drive_used_bytes\",collection_scope=\"local\",rustfs_cluster_id=\"$storage_cluster\"}, drive)", "refId": "PrometheusVariableQueryEditor-VariableQuery" }, "refresh": 2, diff --git a/.docker/observability/prometheus-rules/rustfs-storage.yml b/.docker/observability/prometheus-rules/rustfs-storage.yml new file mode 100644 index 000000000..45f5d956c --- /dev/null +++ b/.docker/observability/prometheus-rules/rustfs-storage.yml @@ -0,0 +1,575 @@ +# Preserve OTLP timestamps in the Collector. Apply timestamp() directly to +# each raw selector, before label rewriting; functions such as label_replace() +# replace the evaluation timestamp and would make a cached value look new. +# Compare publication times at Prometheus millisecond precision so points +# published and exported within the same millisecond are not withheld. +groups: + - name: rustfs-storage-snapshots + interval: 15s + rules: + - record: rustfs:storage_snapshot:fresh + expr: | + rustfs_storage_snapshot_last_success_timestamp_seconds{rustfs_cluster_id!="",collection_scope=~"local|cluster"} + and ((time() - rustfs_storage_snapshot_last_success_timestamp_seconds{rustfs_cluster_id!="",collection_scope=~"local|cluster"}) <= rustfs_storage_snapshot_max_age_seconds{rustfs_cluster_id!="",collection_scope=~"local|cluster"}) + and ((time() - rustfs_storage_snapshot_last_success_timestamp_seconds{rustfs_cluster_id!="",collection_scope=~"local|cluster"}) >= 0) + and (rustfs_storage_snapshot_max_age_seconds{rustfs_cluster_id!="",collection_scope=~"local|cluster"} > 0) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_buckets_total + expr: | + rustfs_cluster_buckets_total{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_buckets_total{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_capacity_free_bytes + expr: | + rustfs_cluster_capacity_free_bytes{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_capacity_free_bytes{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_capacity_missing_drives + expr: | + rustfs_cluster_capacity_missing_drives{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_capacity_missing_drives{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_capacity_raw_total_bytes + expr: | + rustfs_cluster_capacity_raw_total_bytes{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_capacity_raw_total_bytes{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_capacity_stale_drives + expr: | + rustfs_cluster_capacity_stale_drives{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_capacity_stale_drives{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_capacity_usable_total_bytes + expr: | + rustfs_cluster_capacity_usable_total_bytes{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_capacity_usable_total_bytes{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_capacity_used_bytes + expr: | + rustfs_cluster_capacity_used_bytes{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_capacity_used_bytes{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_drive_capacity_observation_age_seconds + expr: | + rustfs_cluster_drive_capacity_observation_age_seconds{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_drive_capacity_observation_age_seconds{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_drive_capacity_observation_state + expr: | + rustfs_cluster_drive_capacity_observation_state{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_drive_capacity_observation_state{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_drive_free_bytes + expr: | + rustfs_cluster_drive_free_bytes{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_drive_free_bytes{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_drive_offline_duration_seconds + expr: | + rustfs_cluster_drive_offline_duration_seconds{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_drive_offline_duration_seconds{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_drive_present + expr: | + rustfs_cluster_drive_present{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_drive_present{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_drive_runtime_state + expr: | + rustfs_cluster_drive_runtime_state{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_drive_runtime_state{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_drive_total_bytes + expr: | + rustfs_cluster_drive_total_bytes{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_drive_total_bytes{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_drive_used_bytes + expr: | + rustfs_cluster_drive_used_bytes{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_drive_used_bytes{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_erasure_set_data_shards + expr: | + rustfs_cluster_erasure_set_data_shards{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_erasure_set_data_shards{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_erasure_set_healing_drives_count + expr: | + rustfs_cluster_erasure_set_healing_drives_count{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_erasure_set_healing_drives_count{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_erasure_set_health + expr: | + rustfs_cluster_erasure_set_health{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_erasure_set_health{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_erasure_set_online_drives_count + expr: | + rustfs_cluster_erasure_set_online_drives_count{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_erasure_set_online_drives_count{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_erasure_set_overall_health + expr: | + rustfs_cluster_erasure_set_overall_health{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_erasure_set_overall_health{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_erasure_set_overall_write_quorum + expr: | + rustfs_cluster_erasure_set_overall_write_quorum{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_erasure_set_overall_write_quorum{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_erasure_set_parity + expr: | + rustfs_cluster_erasure_set_parity{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_erasure_set_parity{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_erasure_set_read_health + expr: | + rustfs_cluster_erasure_set_read_health{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_erasure_set_read_health{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_erasure_set_read_quorum + expr: | + rustfs_cluster_erasure_set_read_quorum{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_erasure_set_read_quorum{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_erasure_set_read_tolerance + expr: | + rustfs_cluster_erasure_set_read_tolerance{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_erasure_set_read_tolerance{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_erasure_set_size + expr: | + rustfs_cluster_erasure_set_size{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_erasure_set_size{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_erasure_set_write_health + expr: | + rustfs_cluster_erasure_set_write_health{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_erasure_set_write_health{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_erasure_set_write_quorum + expr: | + rustfs_cluster_erasure_set_write_quorum{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_erasure_set_write_quorum{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_erasure_set_write_tolerance + expr: | + rustfs_cluster_erasure_set_write_tolerance{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_erasure_set_write_tolerance{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_health_drives_count + expr: | + rustfs_cluster_health_drives_count{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_health_drives_count{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_health_drives_offline_count + expr: | + rustfs_cluster_health_drives_offline_count{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_health_drives_offline_count{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_health_drives_online_count + expr: | + rustfs_cluster_health_drives_online_count{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_health_drives_online_count{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_cluster_objects_total + expr: | + rustfs_cluster_objects_total{rustfs_cluster_id!="",collection_scope="cluster"} + and (timestamp(rustfs_cluster_objects_total{rustfs_cluster_id!="",collection_scope="cluster"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_node_disk_free_bytes + expr: | + rustfs_node_disk_free_bytes{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_node_disk_free_bytes{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_node_disk_total_bytes + expr: | + rustfs_node_disk_total_bytes{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_node_disk_total_bytes{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_node_disk_used_bytes + expr: | + rustfs_node_disk_used_bytes{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_node_disk_used_bytes{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_api_calls_total + expr: | + rustfs_system_drive_api_calls_total{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_api_calls_total{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_api_latency_by_api_micros + expr: | + rustfs_system_drive_api_latency_by_api_micros{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_api_latency_by_api_micros{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_api_latency_micros + expr: | + rustfs_system_drive_api_latency_micros{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_api_latency_micros{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_availability_errors_total + expr: | + rustfs_system_drive_availability_errors_total{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_availability_errors_total{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_capacity_observation_age_seconds + expr: | + rustfs_system_drive_capacity_observation_age_seconds{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_capacity_observation_age_seconds{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_capacity_observation_state + expr: | + rustfs_system_drive_capacity_observation_state{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_capacity_observation_state{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_count + expr: | + rustfs_system_drive_count{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_count{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_deletes_total + expr: | + rustfs_system_drive_deletes_total{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_deletes_total{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_free_bytes + expr: | + rustfs_system_drive_free_bytes{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_free_bytes{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_free_inodes + expr: | + rustfs_system_drive_free_inodes{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_free_inodes{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_healing + expr: | + rustfs_system_drive_healing{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_healing{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_health + expr: | + rustfs_system_drive_health{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_health{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_info + expr: | + rustfs_system_drive_info{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_info{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_io_errors_total + expr: | + rustfs_system_drive_io_errors_total{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_io_errors_total{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_offline_count + expr: | + rustfs_system_drive_offline_count{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_offline_count{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_offline_duration_seconds + expr: | + rustfs_system_drive_offline_duration_seconds{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_offline_duration_seconds{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_online_count + expr: | + rustfs_system_drive_online_count{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_online_count{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_perc_util + expr: | + rustfs_system_drive_perc_util{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_perc_util{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_present + expr: | + rustfs_system_drive_present{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_present{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_reads_await + expr: | + rustfs_system_drive_reads_await{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_reads_await{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_reads_kb_per_sec + expr: | + rustfs_system_drive_reads_kb_per_sec{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_reads_kb_per_sec{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_reads_per_sec + expr: | + rustfs_system_drive_reads_per_sec{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_reads_per_sec{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_runtime_state + expr: | + rustfs_system_drive_runtime_state{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_runtime_state{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_scanning + expr: | + rustfs_system_drive_scanning{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_scanning{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_timeout_errors_total + expr: | + rustfs_system_drive_timeout_errors_total{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_timeout_errors_total{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_total_bytes + expr: | + rustfs_system_drive_total_bytes{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_total_bytes{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_total_inodes + expr: | + rustfs_system_drive_total_inodes{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_total_inodes{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_used_bytes + expr: | + rustfs_system_drive_used_bytes{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_used_bytes{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_used_inodes + expr: | + rustfs_system_drive_used_inodes{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_used_inodes{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_waiting_io + expr: | + rustfs_system_drive_waiting_io{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_waiting_io{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_writes_await + expr: | + rustfs_system_drive_writes_await{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_writes_await{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_writes_kb_per_sec + expr: | + rustfs_system_drive_writes_kb_per_sec{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_writes_kb_per_sec{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_writes_per_sec + expr: | + rustfs_system_drive_writes_per_sec{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_writes_per_sec{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) + - record: rustfs:storage:current + labels: + source_metric: rustfs_system_drive_writes_total + expr: | + rustfs_system_drive_writes_total{rustfs_cluster_id!="",collection_scope="local"} + and (timestamp(rustfs_system_drive_writes_total{rustfs_cluster_id!="",collection_scope="local"}) + >= on (rustfs_cluster_id, observer, collection_scope, job, instance) group_left() + (floor(rustfs:storage_snapshot:fresh * 1000) / 1000)) diff --git a/.docker/observability/tests/storage-rules.test.yml b/.docker/observability/tests/storage-rules.test.yml new file mode 100644 index 000000000..d70f68087 --- /dev/null +++ b/.docker/observability/tests/storage-rules.test.yml @@ -0,0 +1,189 @@ +rule_files: +- ../prometheus-rules/rustfs-storage.yml +evaluation_interval: 15s +tests: +- name: owners, pools, clusters and observer views remain distinct in mixed versions + interval: 1m + input_series: + - series: rustfs_storage_snapshot_last_success_timestamp_seconds{rustfs_cluster_id="a",observer="n0",collection_scope="local",job="rustfs",instance="collector"} + values: 0+60x8 + - series: rustfs_storage_snapshot_max_age_seconds{rustfs_cluster_id="a",observer="n0",collection_scope="local",job="rustfs",instance="collector"} + values: 180x8 + - series: rustfs_system_drive_total_bytes{rustfs_cluster_id="a",observer="n0",collection_scope="local",job="rustfs",instance="collector",server="n0",drive="/data",pool_index="0",set_index="0",drive_index="0"} + values: 100x8 + - series: rustfs_storage_snapshot_last_success_timestamp_seconds{rustfs_cluster_id="a",observer="n1",collection_scope="local",job="rustfs",instance="collector"} + values: 0+60x8 + - series: rustfs_storage_snapshot_max_age_seconds{rustfs_cluster_id="a",observer="n1",collection_scope="local",job="rustfs",instance="collector"} + values: 180x8 + - series: rustfs_system_drive_total_bytes{rustfs_cluster_id="a",observer="n1",collection_scope="local",job="rustfs",instance="collector",server="n1",drive="/data",pool_index="0",set_index="0",drive_index="1"} + values: 100x8 + - series: rustfs_storage_snapshot_last_success_timestamp_seconds{rustfs_cluster_id="a",observer="n2",collection_scope="local",job="rustfs",instance="collector"} + values: 0+60x8 + - series: rustfs_storage_snapshot_max_age_seconds{rustfs_cluster_id="a",observer="n2",collection_scope="local",job="rustfs",instance="collector"} + values: 180x8 + - series: rustfs_system_drive_total_bytes{rustfs_cluster_id="a",observer="n2",collection_scope="local",job="rustfs",instance="collector",server="n2",drive="/data",pool_index="1",set_index="0",drive_index="0"} + values: 100x8 + - series: rustfs_storage_snapshot_last_success_timestamp_seconds{rustfs_cluster_id="a",observer="n3",collection_scope="local",job="rustfs",instance="collector"} + values: 0+60x8 + - series: rustfs_storage_snapshot_max_age_seconds{rustfs_cluster_id="a",observer="n3",collection_scope="local",job="rustfs",instance="collector"} + values: 180x8 + - series: rustfs_system_drive_total_bytes{rustfs_cluster_id="a",observer="n3",collection_scope="local",job="rustfs",instance="collector",server="n3",drive="/data",pool_index="1",set_index="0",drive_index="1"} + values: 100x8 + - series: rustfs_storage_snapshot_last_success_timestamp_seconds{rustfs_cluster_id="b",observer="n0",collection_scope="local",job="rustfs",instance="collector"} + values: 0+60x8 + - series: rustfs_storage_snapshot_max_age_seconds{rustfs_cluster_id="b",observer="n0",collection_scope="local",job="rustfs",instance="collector"} + values: 180x8 + - series: rustfs_system_drive_total_bytes{rustfs_cluster_id="b",observer="n0",collection_scope="local",job="rustfs",instance="collector",server="n0",drive="/data",pool_index="0",set_index="0",drive_index="0"} + values: 1000x8 + - series: rustfs_storage_snapshot_last_success_timestamp_seconds{rustfs_cluster_id="b",observer="n1",collection_scope="local",job="rustfs",instance="collector"} + values: 0+60x8 + - series: rustfs_storage_snapshot_max_age_seconds{rustfs_cluster_id="b",observer="n1",collection_scope="local",job="rustfs",instance="collector"} + values: 180x8 + - series: rustfs_system_drive_total_bytes{rustfs_cluster_id="b",observer="n1",collection_scope="local",job="rustfs",instance="collector",server="n1",drive="/data",pool_index="0",set_index="0",drive_index="1"} + values: 1000x8 + - series: rustfs_storage_snapshot_last_success_timestamp_seconds{rustfs_cluster_id="b",observer="n2",collection_scope="local",job="rustfs",instance="collector"} + values: 0+60x8 + - series: rustfs_storage_snapshot_max_age_seconds{rustfs_cluster_id="b",observer="n2",collection_scope="local",job="rustfs",instance="collector"} + values: 180x8 + - series: rustfs_system_drive_total_bytes{rustfs_cluster_id="b",observer="n2",collection_scope="local",job="rustfs",instance="collector",server="n2",drive="/data",pool_index="1",set_index="0",drive_index="0"} + values: 1000x8 + - series: rustfs_storage_snapshot_last_success_timestamp_seconds{rustfs_cluster_id="b",observer="n3",collection_scope="local",job="rustfs",instance="collector"} + values: 0+60x8 + - series: rustfs_storage_snapshot_max_age_seconds{rustfs_cluster_id="b",observer="n3",collection_scope="local",job="rustfs",instance="collector"} + values: 180x8 + - series: rustfs_system_drive_total_bytes{rustfs_cluster_id="b",observer="n3",collection_scope="local",job="rustfs",instance="collector",server="n3",drive="/data",pool_index="1",set_index="0",drive_index="1"} + values: 1000x8 + - series: rustfs_storage_snapshot_last_success_timestamp_seconds{rustfs_cluster_id="a",observer="n0",collection_scope="cluster",job="rustfs",instance="collector"} + values: 0+60x8 + - series: rustfs_storage_snapshot_max_age_seconds{rustfs_cluster_id="a",observer="n0",collection_scope="cluster",job="rustfs",instance="collector"} + values: 180x8 + - series: rustfs_cluster_capacity_raw_total_bytes{rustfs_cluster_id="a",observer="n0",collection_scope="cluster",job="rustfs",instance="collector"} + values: 400x8 + - series: rustfs_storage_snapshot_last_success_timestamp_seconds{rustfs_cluster_id="a",observer="n1",collection_scope="cluster",job="rustfs",instance="collector"} + values: 0+60x8 + - series: rustfs_storage_snapshot_max_age_seconds{rustfs_cluster_id="a",observer="n1",collection_scope="cluster",job="rustfs",instance="collector"} + values: 180x8 + - series: rustfs_cluster_capacity_raw_total_bytes{rustfs_cluster_id="a",observer="n1",collection_scope="cluster",job="rustfs",instance="collector"} + values: 300x8 + - series: rustfs_system_drive_total_bytes{rustfs_cluster_id="a",observer="old",server="n0",drive="/data",job="rustfs",instance="collector"} + values: 999x8 + promql_expr_test: + - expr: sum by (rustfs_cluster_id) (rustfs:storage:current{source_metric="rustfs_system_drive_total_bytes",collection_scope="local"}) + eval_time: 2m + exp_samples: + - labels: '{rustfs_cluster_id="a"}' + value: 400 + - labels: '{rustfs_cluster_id="b"}' + value: 4000 + - expr: count by (rustfs_cluster_id,pool_index) (rustfs:storage:current{source_metric="rustfs_system_drive_total_bytes"}) + eval_time: 2m + exp_samples: + - labels: '{rustfs_cluster_id="a",pool_index="0"}' + value: 2 + - labels: '{rustfs_cluster_id="a",pool_index="1"}' + value: 2 + - labels: '{rustfs_cluster_id="b",pool_index="0"}' + value: 2 + - labels: '{rustfs_cluster_id="b",pool_index="1"}' + value: 2 + - expr: sum(rustfs:storage:current{source_metric="rustfs_cluster_capacity_raw_total_bytes",rustfs_cluster_id="a",observer="n0"}) + eval_time: 2m + exp_samples: + - labels: '{}' + value: 400 + - expr: sum(rustfs:storage:current{source_metric="rustfs_cluster_capacity_raw_total_bytes",rustfs_cluster_id="a",observer="n1"}) + eval_time: 2m + exp_samples: + - labels: '{}' + value: 300 +- name: removed values and identities cannot rejoin new snapshots; stalled sources expire + interval: 1m + input_series: + - series: rustfs_storage_snapshot_last_success_timestamp_seconds{rustfs_cluster_id="a",observer="n0",collection_scope="local",job="rustfs",instance="collector"} + values: 0+60x8 + - series: rustfs_storage_snapshot_max_age_seconds{rustfs_cluster_id="a",observer="n0",collection_scope="local",job="rustfs",instance="collector"} + values: 180x8 + - series: rustfs_storage_snapshot_last_success_timestamp_seconds{rustfs_cluster_id="a",observer="n1",collection_scope="local",job="rustfs",instance="collector"} + values: 0 60 _ _ _ _ _ _ _ + - series: rustfs_storage_snapshot_max_age_seconds{rustfs_cluster_id="a",observer="n1",collection_scope="local",job="rustfs",instance="collector"} + values: 180x8 + - series: rustfs_system_drive_info{rustfs_cluster_id="a",observer="n0",collection_scope="local",job="rustfs",instance="collector",server="n0",drive="/data",disk_id="old"} + values: 1 1 _ _ _ _ _ _ _ + - series: rustfs_system_drive_info{rustfs_cluster_id="a",observer="n0",collection_scope="local",job="rustfs",instance="collector",server="n0",drive="/data",disk_id="new"} + values: _ _ 1 1 1 1 1 1 1 + - series: rustfs_system_drive_waiting_io{rustfs_cluster_id="a",observer="n0",collection_scope="local",job="rustfs",instance="collector",server="n0",drive="/data"} + values: 7 7 _ _ _ _ _ _ _ + - series: rustfs_system_drive_total_bytes{rustfs_cluster_id="a",observer="n1",collection_scope="local",job="rustfs",instance="collector",server="n1",drive="/data"} + values: 100 100 _ _ _ _ _ _ _ + promql_expr_test: + - expr: rustfs:storage:current{source_metric="rustfs_system_drive_info",disk_id="old"} + eval_time: 2m + exp_samples: [] + - expr: rustfs:storage:current{source_metric="rustfs_system_drive_waiting_io"} + eval_time: 2m + exp_samples: [] + - expr: count(rustfs:storage:current{source_metric="rustfs_system_drive_info",disk_id="new"}) + eval_time: 2m + exp_samples: + - labels: '{}' + value: 1 + - expr: rustfs:storage:current{source_metric="rustfs_system_drive_total_bytes"} + eval_time: 5m + exp_samples: [] + - expr: rustfs:storage_snapshot:fresh{observer="n1"} + eval_time: 5m + exp_samples: [] +- name: counters reset independently; filter versions before rate and sum rates across owners + interval: 1m + input_series: + - series: rustfs_storage_snapshot_last_success_timestamp_seconds{rustfs_cluster_id="a",observer="n0",collection_scope="local",job="rustfs",instance="collector"} + values: 0+60x8 + - series: rustfs_storage_snapshot_max_age_seconds{rustfs_cluster_id="a",observer="n0",collection_scope="local",job="rustfs",instance="collector"} + values: 180x8 + - series: rustfs_system_drive_api_calls_total{rustfs_cluster_id="a",observer="n0",collection_scope="local",job="rustfs",instance="collector",server="n0",drive="/data",disk_id="n0-disk",api="read_all"} + values: 0 60 120 30 90 150 210 + - series: rustfs_storage_snapshot_last_success_timestamp_seconds{rustfs_cluster_id="a",observer="n1",collection_scope="local",job="rustfs",instance="collector"} + values: 0+60x8 + - series: rustfs_storage_snapshot_max_age_seconds{rustfs_cluster_id="a",observer="n1",collection_scope="local",job="rustfs",instance="collector"} + values: 180x8 + - series: rustfs_system_drive_api_calls_total{rustfs_cluster_id="a",observer="n1",collection_scope="local",job="rustfs",instance="collector",server="n1",drive="/data",disk_id="n1-disk",api="read_all"} + values: 0+120x6 + - series: rustfs_system_drive_api_calls_total{rustfs_cluster_id="a",server="n0",drive="/data",observer="old"} + values: 0+999x6 + promql_expr_test: + - expr: sum(resets(rustfs_system_drive_api_calls_total{collection_scope="local"}[5m])) + eval_time: 5m + exp_samples: + - labels: '{}' + value: 1 + - expr: sum(rate(rustfs_system_drive_api_calls_total{collection_scope="local"}[2m]) and ignoring(source_metric) rustfs:storage:current{source_metric="rustfs_system_drive_api_calls_total"}) + eval_time: 5m + exp_samples: + - labels: '{}' + value: 3 +- name: completed slow collection has no validity budget + interval: 1m + input_series: + - series: rustfs_storage_snapshot_last_success_timestamp_seconds{rustfs_cluster_id="a",observer="slow",collection_scope="cluster",job="rustfs",instance="collector"} + values: 0+60x3 + - series: rustfs_storage_snapshot_max_age_seconds{rustfs_cluster_id="a",observer="slow",collection_scope="cluster",job="rustfs",instance="collector"} + values: '0x3' + promql_expr_test: + - expr: rustfs:storage_snapshot:fresh{observer="slow"} + eval_time: 2m + exp_samples: [] +- name: submillisecond publication and millisecond OTLP samples share a cutoff + interval: 1m + input_series: + - series: rustfs_storage_snapshot_last_success_timestamp_seconds{rustfs_cluster_id="a",observer="submillisecond",collection_scope="local",job="rustfs",instance="collector"} + values: 0.0009+60x3 + - series: rustfs_storage_snapshot_max_age_seconds{rustfs_cluster_id="a",observer="submillisecond",collection_scope="local",job="rustfs",instance="collector"} + values: 180x3 + - series: rustfs_system_drive_total_bytes{rustfs_cluster_id="a",observer="submillisecond",collection_scope="local",job="rustfs",instance="collector",server="submillisecond",drive="/data"} + values: 100x3 + promql_expr_test: + - expr: count(rustfs:storage:current{source_metric="rustfs_system_drive_total_bytes",observer="submillisecond"}) + eval_time: 2m15s + exp_samples: + - labels: '{}' + value: 1 diff --git a/crates/e2e_test/README.md b/crates/e2e_test/README.md index 9ba6ae346..28da958bc 100644 --- a/crates/e2e_test/README.md +++ b/crates/e2e_test/README.md @@ -30,6 +30,8 @@ Registered in [`src/lib.rs`](src/lib.rs). Grouped by concern: | **chaos / reliability** | [`src/chaos.rs`](src/chaos.rs), `reliability_disk_fault_test`, `heal_erasure_disk_rebuild_test`, `server_startup_failfast_test` | Disk offline/replace/corrupt, EC rebuild, heal, fail-fast startup | | **upgrade compatibility** | `upgrade_compatibility_test` | Pinned previous-release writes followed by current-build reads on the same data directory | +The external-tool `storage_metric_ownership_test` validates the OTLP/Collector/Prometheus path, including a rolling upgrade and node failures. See the [storage metrics guide](../../docs/operations/storage-metrics.md) for its required binaries and focused command. + ## How to run All commands assume repo root. `cargo test` triggers an on-demand build of the diff --git a/crates/e2e_test/src/lib.rs b/crates/e2e_test/src/lib.rs index 72f5121fb..a9d35da00 100644 --- a/crates/e2e_test/src/lib.rs +++ b/crates/e2e_test/src/lib.rs @@ -12,6 +12,9 @@ // See the License for the specific language governing permissions and // limitations under the License. +#[cfg(test)] +mod storage_metric_ownership_test; + mod reliant; mod storage_api; diff --git a/crates/e2e_test/src/storage_metric_ownership_test.rs b/crates/e2e_test/src/storage_metric_ownership_test.rs new file mode 100644 index 000000000..f89bd6920 --- /dev/null +++ b/crates/e2e_test/src/storage_metric_ownership_test.rs @@ -0,0 +1,335 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Native OTLP -> Collector -> Prometheus contract, including a rolling upgrade. + +use crate::common::RustFSTestClusterEnvironment; +use aws_sdk_s3::primitives::ByteStream; +use serde_json::Value; +use std::fs::{self, File}; +use std::net::TcpListener; +use std::path::{Path, PathBuf}; +use std::process::{Child, Command, Stdio}; +use std::time::{Duration, Instant}; + +type TestResult = Result>; + +struct ToolProcess(Child); +impl Drop for ToolProcess { + fn drop(&mut self) { + let _ = self.0.kill(); + let _ = self.0.wait(); + } +} + +fn required_binary(name: &str) -> TestResult { + let path = PathBuf::from(std::env::var(name).map_err(|_| format!("{name} must name a pinned executable"))?); + if !path.is_file() { + return Err(format!("{name} does not name a file: {}", path.display()).into()); + } + Ok(path) +} + +fn free_port() -> TestResult { + Ok(TcpListener::bind("127.0.0.1:0")?.local_addr()?.port()) +} + +fn start_tool(binary: &Path, args: &[String], log: &Path) -> TestResult { + let log = File::create(log)?; + Ok(ToolProcess( + Command::new(binary) + .args(args) + .env("NO_PROXY", "127.0.0.1,localhost") + .env_remove("HTTP_PROXY") + .env_remove("HTTPS_PROXY") + .stdout(Stdio::from(log.try_clone()?)) + .stderr(Stdio::from(log)) + .spawn()?, + )) +} + +async fn query(client: &reqwest::Client, base: &str, expression: &str) -> TestResult { + let mut url = reqwest::Url::parse(&format!("{base}/api/v1/query"))?; + url.query_pairs_mut().append_pair("query", expression); + let result: Value = client.get(url).send().await?.error_for_status()?.json().await?; + if result["status"] != "success" { + return Err(format!("PromQL failed: {result}").into()); + } + Ok(result["data"]["result"].clone()) +} + +async fn await_count(client: &reqwest::Client, base: &str, selector: &str, expected: u64) -> TestResult { + let deadline = Instant::now() + Duration::from_secs(120); + loop { + let result = query(client, base, &format!("count({selector}) or vector(0)")).await; + if let Ok(rows) = &result { + if rows[0]["value"][1].as_str().and_then(|value| value.parse::().ok()) == Some(expected) { + println!("PASS count={expected}: {selector}"); + return Ok(()); + } + } + if Instant::now() >= deadline { + return Err(format!("expected {expected} for {selector}; last result: {result:?}").into()); + } + tokio::time::sleep(Duration::from_millis(250)).await; + } +} + +async fn validate_dashboard_queries(client: &reqwest::Client, base: &str, observer: &str) -> TestResult { + let path = PathBuf::from(env!("CARGO_MANIFEST_DIR")).join("../../.docker/observability/grafana/dashboards/rustfs.json"); + let dashboard: Value = serde_json::from_str(&fs::read_to_string(path)?)?; + for name in ["storage_cluster", "storage_observer"] { + let variable = dashboard["templating"]["list"] + .as_array() + .ok_or("dashboard variables")? + .iter() + .find(|variable| variable["name"] == name) + .ok_or("storage selection variable")?; + assert_eq!(variable["multi"], false, "storage views must select one {name}"); + assert_eq!(variable["includeAll"], false, "storage views must select one {name}"); + } + let mut pending = dashboard["panels"] + .as_array() + .ok_or("dashboard panels")? + .iter() + .collect::>(); + let mut checked = 0; + while let Some(panel) = pending.pop() { + if let Some(children) = panel["panels"].as_array() { + pending.extend(children); + } + for target in panel["targets"].as_array().into_iter().flatten() { + let Some(expression) = target["expr"].as_str() else { continue }; + if !expression.contains("rustfs:storage:current") { + continue; + } + if expression.contains("collection_scope=\"cluster\"") { + assert!( + expression.contains("observer=\"$storage_observer\""), + "global views must select one observer: {expression}" + ); + } + let mut expression = expression.to_string(); + for (name, value) in [ + ("$__rate_interval", "5m"), + ("$storage_cluster", "metrics-e2e"), + ("$storage_observer", observer), + ("$drive_api", ".*"), + ("$server", ".*"), + ("$drive", ".*"), + ("$job", "rustfs"), + ] { + expression = expression.replace(name, value); + } + query(client, base, &expression).await?; + checked += 1; + } + } + assert!(checked > 0, "the storage dashboard queries must be exercised"); + println!("PASS: {checked} storage dashboard queries against the live pipeline"); + Ok(()) +} + +#[tokio::test(flavor = "multi_thread", worker_threads = 4)] +#[ignore = "external tools: pinned Collector, Prometheus, previous release and current RustFS binaries"] +async fn storage_metric_ownership_pipeline() -> TestResult { + let baseline = required_binary("RUSTFS_METRICS_BASELINE_BINARY")?; + let current = required_binary("CARGO_BIN_EXE_rustfs")?; + let collector = required_binary("RUSTFS_OTELCOL_BINARY")?; + let prometheus = required_binary("RUSTFS_PROMETHEUS_BINARY")?; + let temp = tempfile::Builder::new().prefix("rustfs-storage-metrics-").tempdir()?; + let work = if let Ok(path) = std::env::var("RUSTFS_METRICS_E2E_ARTIFACTS") { + let path = PathBuf::from(path); + fs::create_dir_all(&path)?; + tempfile::Builder::new().prefix("storage-run-").tempdir_in(path)?.keep() + } else { + temp.path().to_path_buf() + }; + println!("Metrics pipeline logs: {}", work.display()); + let otlp = free_port()?; + let scrape = free_port()?; + let prom = free_port()?; + let collector_config = work.join("collector.yaml"); + fs::write( + &collector_config, + format!( + r#"receivers: + otlp: + protocols: + http: + endpoint: 127.0.0.1:{otlp} +exporters: + prometheus: + endpoint: 127.0.0.1:{scrape} + send_timestamps: true + metric_expiration: 5m + resource_to_telemetry_conversion: + enabled: true +service: + telemetry: + metrics: + level: none + logs: + level: warn + pipelines: + metrics: + receivers: [otlp] + exporters: [prometheus] +"# + ), + )?; + let rules = PathBuf::from(env!("CARGO_MANIFEST_DIR")) + .join("../../.docker/observability/prometheus-rules/rustfs-storage.yml") + .canonicalize()?; + let prom_config = work.join("prometheus.yaml"); + fs::write( + &prom_config, + format!( + r#"global: + scrape_interval: 1s + evaluation_interval: 1s +rule_files: + - '{}' +scrape_configs: + - job_name: rustfs + static_configs: + - targets: ['127.0.0.1:{scrape}'] +"#, + rules.display() + ), + )?; + // Use the shipped expressions, with only the test evaluation interval shortened. + let test_rules = work.join("storage-rules.yaml"); + fs::write(&test_rules, fs::read_to_string(&rules)?.replace("interval: 15s", "interval: 1s"))?; + fs::write( + &prom_config, + fs::read_to_string(&prom_config)?.replace(&rules.display().to_string(), &test_rules.display().to_string()), + )?; + let _collector = start_tool( + &collector, + &[format!("--config={}", collector_config.display())], + &work.join("collector.log"), + )?; + let _prometheus = start_tool( + &prometheus, + &[ + format!("--config.file={}", prom_config.display()), + format!("--web.listen-address=127.0.0.1:{prom}"), + format!("--storage.tsdb.path={}", work.join("prometheus-data").display()), + ], + &work.join("prometheus.log"), + )?; + let http = reqwest::Client::builder() + .no_proxy() + .timeout(Duration::from_secs(5)) + .build()?; + let prom_url = format!("http://127.0.0.1:{prom}"); + await_count(&http, &prom_url, "up{job=\"rustfs\"} == 1", 1).await?; + + let mut cluster = RustFSTestClusterEnvironment::new(4).await?; + cluster.set_env("NO_PROXY", "127.0.0.1,localhost"); + cluster.set_env("RUSTFS_OBS_METRIC_ENDPOINT", format!("http://127.0.0.1:{otlp}/v1/metrics")); + cluster.set_env("OTEL_RESOURCE_ATTRIBUTES", "rustfs.cluster.id=metrics-e2e"); + cluster.set_env("RUSTFS_OBS_METER_INTERVAL", "2"); + cluster.set_env("RUSTFS_OBS_METRICS_EXPORT_ENABLED", "true"); + cluster.set_env("RUSTFS_OBS_LOGS_EXPORT_ENABLED", "false"); + cluster.set_env("RUSTFS_OBS_TRACES_EXPORT_ENABLED", "false"); + cluster.set_env("RUSTFS_METRICS_NODE_INTERVAL", "2"); + cluster.set_env("RUSTFS_METRICS_CLUSTER_INTERVAL", "5"); + for index in 0..4 { + // Four localhost processes otherwise share the startup resource IP. + // Distinct test host IDs model the four hosts in a distributed deployment. + cluster.set_node_env( + index, + "OTEL_RESOURCE_ATTRIBUTES", + format!("rustfs.cluster.id=metrics-e2e,host.id=metrics-node-{index}"), + )?; + cluster.set_node_capture_log_path(index, work.join(format!("node-{index}.log")).display().to_string())?; + } + cluster.start_with_binary(&baseline).await?; + // Reproduce the original four observers x four global drives before fixing it. + await_count(&http, &prom_url, "rustfs_system_drive_total_bytes{collection_scope=\"\",drive!=\"\"}", 16).await?; + let local = "rustfs:storage:current{source_metric=\"rustfs_system_drive_total_bytes\",collection_scope=\"local\",rustfs_cluster_id=\"metrics-e2e\"}"; + for index in 0..4 { + cluster.stop_node_gracefully(index).await?; + cluster.start_node_from_binary(index, ¤t).await?; + await_count(&http, &prom_url, local, u64::try_from(index + 1)?).await?; + } + let rows = query(&http, &prom_url, local).await?; + for row in rows.as_array().ok_or("expected a metric vector")? { + assert_eq!( + row["metric"]["observer"], row["metric"]["server"], + "a node must only export its own detailed drives" + ); + } + let observer = cluster.nodes[0].address.clone(); + let inventory = format!( + "rustfs:storage:current{{source_metric=\"rustfs_cluster_drive_present\",collection_scope=\"cluster\",rustfs_cluster_id=\"metrics-e2e\",observer=\"{observer}\"}}" + ); + await_count(&http, &prom_url, &inventory, 4).await?; + cluster.create_test_bucket("metrics-ownership").await?; + let client = cluster.create_s3_client(0)?; + for index in 0..8 { + client + .put_object() + .bucket("metrics-ownership") + .key(format!("object-{index}")) + .body(ByteStream::from(vec![7_u8; 4096])) + .send() + .await?; + } + await_count( + &http, + &prom_url, + "count by (server) (rustfs:storage:current{source_metric=\"rustfs_system_drive_api_calls_total\",collection_scope=\"local\"})", + 4, + ).await?; + let counters = query( + &http, + &prom_url, + "rustfs:storage:current{source_metric=\"rustfs_system_drive_api_calls_total\",collection_scope=\"local\"}", + ) + .await?; + assert!( + !counters.as_array().ok_or("expected counters")?.is_empty(), + "exercise actual storage counters" + ); + for row in counters.as_array().ok_or("expected counters")? { + assert_eq!(row["metric"]["observer"], row["metric"]["server"]); + assert!( + !row["metric"]["disk_id"].as_str().unwrap_or_default().is_empty(), + "counters must carry physical disk identity" + ); + } + validate_dashboard_queries(&http, &prom_url, &observer).await?; + for index in (1..4).rev() { + cluster.stop_node(index)?; + // The Collector stays alive; cached samples must not keep stopped owners fresh. + await_count(&http, &prom_url, local, u64::try_from(index)?).await?; + await_count(&http, &prom_url, &inventory, 4).await?; + let unavailable = format!( + "rustfs:storage:current{{source_metric=\"rustfs_cluster_drive_runtime_state\",collection_scope=\"cluster\",rustfs_cluster_id=\"metrics-e2e\",observer=\"{observer}\",state=~\"offline|unknown|suspect\"}} == 1" + ); + await_count(&http, &prom_url, &unavailable, u64::try_from(4 - index)?).await?; + } + cluster.stop(); + await_count(&http, &prom_url, local, 0).await?; + cluster.start_with_binary(¤t).await?; + await_count(&http, &prom_url, local, 4).await?; + await_count(&http, &prom_url, &inventory, 4).await?; + let restored = client.get_object().bucket("metrics-ownership").key("object-0").send().await?; + assert_eq!(restored.body.collect().await?.into_bytes().as_ref(), vec![7_u8; 4096].as_slice()); + println!("PASS: baseline duplication; four rolling upgrades; owner identity; counters; 4 -> 3 -> 2 -> 1 -> 0 -> 4 recovery"); + Ok(()) +} diff --git a/crates/obs/src/metrics/collectors/cluster_drive.rs b/crates/obs/src/metrics/collectors/cluster_drive.rs new file mode 100644 index 000000000..c2c039f43 --- /dev/null +++ b/crates/obs/src/metrics/collectors/cluster_drive.rs @@ -0,0 +1,150 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! One observer's global drive inventory. Runtime counters belong to local drives only. + +use crate::metrics::report::PrometheusMetric; +use crate::metrics::schema::MetricType; + +#[derive(Debug, Clone, Default)] +pub(crate) struct ClusterDriveStats { + pub server: String, + pub drive: String, + pub pool_index: String, + pub set_index: String, + pub drive_index: String, + pub disk_id: String, + pub runtime_state: String, + pub offline_duration_seconds: Option, + pub capacity_state: &'static str, + pub capacity_age_seconds: u64, + pub total_bytes: u64, + pub used_bytes: u64, + pub free_bytes: u64, +} + +pub(crate) fn collect_cluster_drive_metrics(stats: &[ClusterDriveStats]) -> Vec { + let mut metrics = Vec::with_capacity(stats.len() * 13); + for drive in stats { + let metric = |name, help, value| { + PrometheusMetric::new(name, MetricType::Gauge, help, value) + .with_label_owned("server", drive.server.clone()) + .with_label_owned("drive", drive.drive.clone()) + .with_label_owned("pool_index", drive.pool_index.clone()) + .with_label_owned("set_index", drive.set_index.clone()) + .with_label_owned("drive_index", drive.drive_index.clone()) + .with_label_owned("disk_id", drive.disk_id.clone()) + }; + metrics.push(metric( + "rustfs_cluster_drive_present", + "Configured drive slot in this observer's inventory", + 1.0, + )); + for state in ["online", "offline", "returning", "suspect", "unknown"] { + let observed = match drive.runtime_state.as_str() { + "online" | "offline" | "returning" | "suspect" => drive.runtime_state.as_str(), + _ => "unknown", + }; + metrics.push( + metric( + "rustfs_cluster_drive_runtime_state", + "Observed drive runtime state (one active state)", + f64::from(state == observed), + ) + .with_label("state", state), + ); + } + if let Some(seconds) = drive.offline_duration_seconds { + metrics.push(metric( + "rustfs_cluster_drive_offline_duration_seconds", + "Observed duration in seconds the drive has been offline", + seconds as f64, + )); + } + for state in ["live", "stale", "missing"] { + metrics.push( + metric( + "rustfs_cluster_drive_capacity_observation_state", + "Provenance of the observed drive capacity", + f64::from(state == drive.capacity_state), + ) + .with_label("state", state), + ); + } + metrics.push(metric( + "rustfs_cluster_drive_capacity_observation_age_seconds", + "Age in seconds of the drive capacity observation at collection", + drive.capacity_age_seconds as f64, + )); + // A missing capacity observation is unknown, not a zero-capacity drive. + if drive.capacity_state != "missing" { + metrics.push(metric( + "rustfs_cluster_drive_total_bytes", + "Observed total drive capacity in bytes", + drive.total_bytes as f64, + )); + metrics.push(metric( + "rustfs_cluster_drive_used_bytes", + "Observed used drive capacity in bytes", + drive.used_bytes as f64, + )); + metrics.push(metric( + "rustfs_cluster_drive_free_bytes", + "Observed free drive capacity in bytes", + drive.free_bytes as f64, + )); + } + } + metrics +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn unknown_slots_remain_visible_without_inventing_capacity_or_counters() { + let metrics = collect_cluster_drive_metrics(&[ClusterDriveStats { + server: "unreachable:9000".into(), + drive: "/data".into(), + pool_index: "1".into(), + set_index: "0".into(), + drive_index: "3".into(), + capacity_state: "missing", + ..Default::default() + }]); + assert!( + metrics + .iter() + .any(|metric| metric.name == "rustfs_cluster_drive_present" && metric.value == 1.0) + ); + assert!( + metrics + .iter() + .any(|metric| metric.name == "rustfs_cluster_drive_runtime_state" + && metric.value == 1.0 + && metric.labels.iter().any(|(key, value)| *key == "state" && value == "unknown")) + ); + assert!( + !metrics + .iter() + .any(|metric| metric.name.ends_with("_bytes") || metric.metric_type == MetricType::Counter) + ); + assert!( + metrics + .iter() + .all(|metric| metric.labels.iter().any(|(key, value)| *key == "pool_index" && value == "1")) + ); + } +} diff --git a/crates/obs/src/metrics/collectors/mod.rs b/crates/obs/src/metrics/collectors/mod.rs index cb4078669..5043202f4 100644 --- a/crates/obs/src/metrics/collectors/mod.rs +++ b/crates/obs/src/metrics/collectors/mod.rs @@ -17,6 +17,7 @@ pub mod bucket; pub mod bucket_replication; pub mod cluster; pub mod cluster_config; +pub(crate) mod cluster_drive; pub mod cluster_erasure_set; pub mod cluster_health; pub mod cluster_iam; diff --git a/crates/obs/src/metrics/collectors/system_drive.rs b/crates/obs/src/metrics/collectors/system_drive.rs index c38797bfa..a48186b4d 100644 --- a/crates/obs/src/metrics/collectors/system_drive.rs +++ b/crates/obs/src/metrics/collectors/system_drive.rs @@ -193,31 +193,35 @@ pub(crate) fn collect_drive_runtime_detailed_metrics(stats: &[DriveRuntimeDetail let mut metrics = Vec::with_capacity(metric_capacity); for stat in stats { + let first_metric = metrics.len(); let server_label = stat.stats.server.as_str(); let drive_label = stat.stats.drive.as_str(); let topology_labels = topology_labels(stat); - push_drive_metric( - &mut metrics, - &DRIVE_TOTAL_BYTES_MD, - stat.stats.total_bytes as f64, - server_label, - drive_label, - ); - push_drive_metric( - &mut metrics, - &DRIVE_USED_BYTES_MD, - stat.stats.used_bytes as f64, - server_label, - drive_label, - ); - push_drive_metric( - &mut metrics, - &DRIVE_FREE_BYTES_MD, - stat.stats.free_bytes as f64, - server_label, - drive_label, - ); + push_drive_metric(&mut metrics, &DRIVE_PRESENT_MD, 1.0, server_label, drive_label); + if stat.stats.capacity_observation_state != "missing" { + push_drive_metric( + &mut metrics, + &DRIVE_TOTAL_BYTES_MD, + stat.stats.total_bytes as f64, + server_label, + drive_label, + ); + push_drive_metric( + &mut metrics, + &DRIVE_USED_BYTES_MD, + stat.stats.used_bytes as f64, + server_label, + drive_label, + ); + push_drive_metric( + &mut metrics, + &DRIVE_FREE_BYTES_MD, + stat.stats.free_bytes as f64, + server_label, + drive_label, + ); + } push_drive_metric( &mut metrics, &DRIVE_CAPACITY_OBSERVATION_AGE_SECONDS_MD, @@ -352,6 +356,13 @@ pub(crate) fn collect_drive_runtime_detailed_metrics(stats: &[DriveRuntimeDetail ); } } + for metric in &mut metrics[first_metric..] { + if metric.metric_type == crate::metrics::schema::MetricType::Counter { + metric + .labels + .push((DISK_ID_LABEL, Cow::Owned(stat.disk_id.clone().unwrap_or_default()))); + } + } } metrics @@ -409,6 +420,46 @@ pub fn collect_process_disk_metrics( #[cfg(test)] mod tests { use super::*; + + #[test] + fn counters_follow_physical_disk_identity_and_unknown_slots_stay_present() { + let mut drive = DriveRuntimeDetailedStats { + disk_id: Some("old-disk".into()), + pool_index: Some("0".into()), + set_index: Some("0".into()), + drive_index: Some("1".into()), + api_calls: vec![("read_all".into(), 7)], + stats: DriveDetailedStats { + server: "node1:9000".into(), + drive: "/data".into(), + writes_total: Some(7), + ..Default::default() + }, + ..Default::default() + }; + for id in ["old-disk", "new-disk"] { + drive.disk_id = Some(id.into()); + let metrics = collect_drive_runtime_detailed_metrics(&[drive.clone()]); + let counters = metrics + .iter() + .filter(|metric| metric.metric_type == crate::metrics::schema::MetricType::Counter) + .collect::>(); + assert_eq!(counters.len(), 2); + assert!( + counters + .iter() + .all(|metric| metric.labels.iter().any(|(key, value)| *key == DISK_ID_LABEL && value == id)) + ); + } + drive.disk_id = None; + let metrics = collect_drive_runtime_detailed_metrics(&[drive]); + assert!( + metrics + .iter() + .any(|metric| metric.name == "rustfs_system_drive_present" && metric.value == 1.0) + ); + assert!(!metrics.iter().any(|metric| metric.name == "rustfs_system_drive_info")); + } use crate::metrics::report::report_metrics; use crate::metrics::schema::system_process::{PROCESS_EXECUTABLE_NAME_LABEL, PROCESS_PID_LABEL}; use std::collections::BTreeSet; @@ -474,7 +525,7 @@ mod tests { let metrics = collect_drive_runtime_detailed_metrics(&stats); report_metrics(&metrics); - assert_eq!(metrics.len(), 36); + assert_eq!(metrics.len(), 37); // Verify total bytes metric let total_bytes_name = DRIVE_TOTAL_BYTES_MD.get_full_metric_name(); @@ -513,10 +564,11 @@ mod tests { SET_INDEX_LABEL, DRIVE_INDEX_LABEL, API_LABEL, + DISK_ID_LABEL, ], ); - assert_metric_label_keys(&metrics, &DRIVE_WRITES_TOTAL_MD, 11.0, &[SERVER_LABEL, DRIVE_LABEL]); - assert_metric_label_keys(&metrics, &DRIVE_DELETES_TOTAL_MD, 4.0, &[SERVER_LABEL, DRIVE_LABEL]); + assert_metric_label_keys(&metrics, &DRIVE_WRITES_TOTAL_MD, 11.0, &[SERVER_LABEL, DRIVE_LABEL, DISK_ID_LABEL]); + assert_metric_label_keys(&metrics, &DRIVE_DELETES_TOTAL_MD, 4.0, &[SERVER_LABEL, DRIVE_LABEL, DISK_ID_LABEL]); } #[test] @@ -551,7 +603,7 @@ mod tests { let metrics = collect_drive_detailed_metrics(&stats); - assert_eq!(metrics.len(), 8); + assert_eq!(metrics.len(), 9); assert!( metrics .iter() diff --git a/crates/obs/src/metrics/mod.rs b/crates/obs/src/metrics/mod.rs index 009cc94f2..d854c1bdc 100644 --- a/crates/obs/src/metrics/mod.rs +++ b/crates/obs/src/metrics/mod.rs @@ -20,6 +20,7 @@ pub mod scheduler; pub mod schema; pub mod stats_collector; mod storage_api; +mod storage_snapshot; pub use collectors::*; pub use config::*; diff --git a/crates/obs/src/metrics/report.rs b/crates/obs/src/metrics/report.rs index b3a421f55..e8d2f0f45 100644 --- a/crates/obs/src/metrics/report.rs +++ b/crates/obs/src/metrics/report.rs @@ -69,7 +69,7 @@ fn describe_metric_once(name: &'static str, metric_type: MetricType, help: &Cow< } } -fn counter_value_from_f64(value: f64) -> Option { +pub(crate) fn counter_value_from_f64(value: f64) -> Option { if !value.is_finite() || value < 0.0 { return None; } diff --git a/crates/obs/src/metrics/scheduler.rs b/crates/obs/src/metrics/scheduler.rs index 8d527c8df..e82ee8e31 100644 --- a/crates/obs/src/metrics/scheduler.rs +++ b/crates/obs/src/metrics/scheduler.rs @@ -23,13 +23,13 @@ //! - Process disk I/O metrics //! - Host network I/O metrics +use crate::metrics::collectors::cluster_drive::collect_cluster_drive_metrics; use crate::metrics::collectors::{ AuditTargetRuntimeStats, AuditTargetStats, BucketReplicationBacklogStats, BucketReplicationBandwidthStats, BucketReplicationRuntimeStats, - DriveRuntimeDetailedStats, NotificationStats, NotificationTargetRuntimeStats, NotificationTargetStats, @@ -107,7 +107,6 @@ use crate::metrics::schema::bucket_replication::{ BUCKET_REPL_TARGET_SENT_COUNT_MD, BUCKET_REPL_TARGET_TOTAL_FAILED_BYTES_MD, BUCKET_REPL_TARGET_TOTAL_FAILED_COUNT_MD, OPERATION_L, RANGE_L, RESULT_L, TARGET_ARN_L, }; -use crate::metrics::schema::cluster::{CLUSTER_BUCKETS_TOTAL_MD, CLUSTER_OBJECTS_TOTAL_MD}; use crate::metrics::schema::cluster_usage::{ BUCKET_LABEL as USAGE_BUCKET_LABEL, RANGE_LABEL as USAGE_RANGE_LABEL, USAGE_BUCKET_DELETE_MARKERS_COUNT_MD, USAGE_BUCKET_OBJECT_SIZE_DISTRIBUTION_MD, USAGE_BUCKET_OBJECT_VERSION_COUNT_DISTRIBUTION_MD, USAGE_BUCKET_OBJECTS_TOTAL_MD, @@ -139,25 +138,23 @@ use crate::metrics::schema::scanner::{ RESULT_LABEL as SCANNER_RESULT_LABEL, SCANNER_ACTIVE_BUCKET_DRIVE_SCAN_AGE_SECONDS_MD, SCANNER_ACTIVE_BUCKET_DRIVE_SCANS_MD, SCANNER_BUCKET_DRIVE_RESULT_TOTAL_MD, SCANNER_CYCLE_BUCKET_DRIVE_RESULT_MD, SOURCE_LABEL as SCANNER_SOURCE_LABEL, }; -use crate::metrics::schema::system_drive::{ - API_LABEL as DRIVE_API_LABEL, DISK_ID_LABEL, DRIVE_API_CALLS_MD, DRIVE_API_LATENCY_BY_API_MD, DRIVE_DELETES_TOTAL_MD, - DRIVE_HEALING_MD, DRIVE_INDEX_LABEL, DRIVE_INFO_MD, DRIVE_LABEL, DRIVE_OFFLINE_DURATION_SECONDS_MD, DRIVE_RUNTIME_STATE_MD, - DRIVE_SCANNING_MD, DRIVE_WRITES_TOTAL_MD, POOL_INDEX_LABEL, SET_INDEX_LABEL, STATE_LABEL as DRIVE_STATE_LABEL, -}; use crate::metrics::schema::system_process::{PROCESS_EXECUTABLE_NAME_LABEL, PROCESS_PID_LABEL}; use crate::metrics::stats_collector::{ ProcessMetricBundle, collect_api_request_stats, collect_bucket_replication_bandwidth_stats, - collect_bucket_replication_stats_bundle, collect_bucket_stats, collect_cluster_and_health_stats, - collect_cluster_config_stats, collect_cluster_usage_metric_stats, collect_compression_cluster_stats, - collect_disk_and_system_drive_runtime_stats, collect_erasure_set_stats, collect_host_network_stats, collect_iam_stats, - collect_ilm_runtime_metric_stats, collect_internode_network_stats, collect_on_demand_migration_backfill_stats, - collect_on_demand_migration_stats, collect_process_metric_bundle_with, collect_replication_stats, - collect_scanner_runtime_metric_stats, collect_system_cpu_and_memory_stats_with, collect_tier_request_metric_stats, + collect_bucket_replication_stats_bundle, collect_bucket_stats, collect_cluster_config_stats, + collect_cluster_storage_snapshot, collect_cluster_usage_metric_stats, collect_compression_cluster_stats, + collect_disk_and_system_drive_runtime_stats, collect_host_network_stats, collect_iam_stats, collect_ilm_runtime_metric_stats, + collect_internode_network_stats, collect_on_demand_migration_backfill_stats, collect_on_demand_migration_stats, + collect_process_metric_bundle_with, collect_replication_stats, collect_scanner_runtime_metric_stats, + collect_system_cpu_and_memory_stats_with, collect_tier_request_metric_stats, }; +use crate::metrics::storage_snapshot::StorageSnapshotMetrics; use crate::node_identity::{SERVER_LABEL, current_local_node_identity}; use crate::telemetry::retire_metric_series; use futures_util::FutureExt; use rustfs_audit::audit_target_metrics; +use rustfs_config::METER_INTERVAL; +use rustfs_config::observability::ENV_OBS_METER_INTERVAL; use rustfs_io_metrics::ProcessSampler; use rustfs_notify::{notification_metrics_snapshot, notification_target_metrics}; use rustfs_utils::get_env_opt_u64; @@ -171,7 +168,7 @@ use std::time::Duration; use sysinfo::{Networks, System}; use tokio::time::{Instant, Interval, MissedTickBehavior}; use tokio_util::sync::CancellationToken; -use tracing::warn; +use tracing::{error, warn}; const LOG_COMPONENT_OBS: &str = "obs"; const LOG_SUBSYSTEM_METRICS_RUNTIME: &str = "metrics_runtime"; @@ -322,137 +319,10 @@ type AuditLegacyTargetKey = String; type AuditTargetKey = (String, String); // (server, target_id) type NotificationLegacyTargetKey = (String, String); // (target_id, target_type) type NotificationTargetKey = (String, String, String); // (server, target_id, target_type) -type DriveTopologyKey = (String, String, String, String, String); // (server, drive, pool, set, drive_index) -type DriveBasicKey = (String, String); // (server, drive) -type DriveTopologyApiKey = (String, String, String, String, String, String); // (server, drive, pool, set, drive_index, api) -type DriveInfoKey = (String, String, String, String, String, String); // (server, drive, pool, set, drive_index, disk_id) type ScannerCycleBucketDriveResultKey = (String, String, String, String, String); // (server, cycle_scope, bucket, drive, result) type ScannerBucketDriveResultKey = (String, String, String, String); // (server, bucket, drive, result) type ScannerActiveBucketDriveKey = (String, String, String, String); // (server, source, bucket, drive) -fn drive_info_live_keys(stats: &[DriveRuntimeDetailedStats]) -> HashSet { - stats.iter().filter_map(drive_info_key).collect() -} - -fn drive_basic_live_keys(stats: &[DriveRuntimeDetailedStats]) -> HashSet { - stats - .iter() - .map(|stat| (stat.stats.server.clone(), stat.stats.drive.clone())) - .collect() -} - -fn retire_drive_basic_metric_series(key: &DriveBasicKey) -> usize { - let labels = [ - (SERVER_LABEL, Cow::Owned(key.0.clone())), - (DRIVE_LABEL, Cow::Owned(key.1.clone())), - ]; - retire_metric_series(&DRIVE_WRITES_TOTAL_MD.get_full_metric_name(), &labels) - + retire_metric_series(&DRIVE_DELETES_TOTAL_MD.get_full_metric_name(), &labels) -} - -fn drive_topology_live_keys(stats: &[DriveRuntimeDetailedStats]) -> HashSet { - stats.iter().filter_map(drive_topology_key).collect() -} - -fn drive_topology_api_live_keys(stats: &[DriveRuntimeDetailedStats]) -> HashSet { - stats - .iter() - .filter_map(|stat| { - let topology = drive_topology_key(stat)?; - Some( - stat.api_calls - .iter() - .map(|(api, _)| api) - .chain(stat.api_latency_by_api_micros.iter().map(|(api, _)| api)) - .map(move |api| { - ( - topology.0.clone(), - topology.1.clone(), - topology.2.clone(), - topology.3.clone(), - topology.4.clone(), - api.clone(), - ) - }), - ) - }) - .flatten() - .collect() -} - -fn drive_topology_key(stat: &DriveRuntimeDetailedStats) -> Option { - Some(( - stat.stats.server.clone(), - stat.stats.drive.clone(), - stat.pool_index.as_ref()?.clone(), - stat.set_index.as_ref()?.clone(), - stat.drive_index.as_ref()?.clone(), - )) -} - -fn drive_info_key(stat: &DriveRuntimeDetailedStats) -> Option { - let disk_id = stat.disk_id.as_ref().filter(|disk_id| !disk_id.is_empty())?; - Some(( - stat.stats.server.clone(), - stat.stats.drive.clone(), - stat.pool_index.as_ref()?.clone(), - stat.set_index.as_ref()?.clone(), - stat.drive_index.as_ref()?.clone(), - disk_id.clone(), - )) -} - -fn retire_drive_info_metric_series(key: &DriveInfoKey) -> usize { - let labels = [ - (SERVER_LABEL, Cow::Owned(key.0.clone())), - (DRIVE_LABEL, Cow::Owned(key.1.clone())), - (POOL_INDEX_LABEL, Cow::Owned(key.2.clone())), - (SET_INDEX_LABEL, Cow::Owned(key.3.clone())), - (DRIVE_INDEX_LABEL, Cow::Owned(key.4.clone())), - (DISK_ID_LABEL, Cow::Owned(key.5.clone())), - ]; - retire_metric_series(&DRIVE_INFO_MD.get_full_metric_name(), &labels) -} - -fn retire_drive_topology_metric_series(key: &DriveTopologyKey) -> usize { - let labels = [ - (SERVER_LABEL, Cow::Owned(key.0.clone())), - (DRIVE_LABEL, Cow::Owned(key.1.clone())), - (POOL_INDEX_LABEL, Cow::Owned(key.2.clone())), - (SET_INDEX_LABEL, Cow::Owned(key.3.clone())), - (DRIVE_INDEX_LABEL, Cow::Owned(key.4.clone())), - ]; - let mut retired = 0; - for descriptor in [&DRIVE_HEALING_MD, &DRIVE_SCANNING_MD, &DRIVE_OFFLINE_DURATION_SECONDS_MD] { - retired += retire_metric_series(&descriptor.get_full_metric_name(), &labels); - } - for state in ["online", "offline", "returning", "suspect", "unknown"] { - let state_labels = [ - (SERVER_LABEL, Cow::Owned(key.0.clone())), - (DRIVE_LABEL, Cow::Owned(key.1.clone())), - (POOL_INDEX_LABEL, Cow::Owned(key.2.clone())), - (SET_INDEX_LABEL, Cow::Owned(key.3.clone())), - (DRIVE_INDEX_LABEL, Cow::Owned(key.4.clone())), - (DRIVE_STATE_LABEL, Cow::Borrowed(state)), - ]; - retired += retire_metric_series(&DRIVE_RUNTIME_STATE_MD.get_full_metric_name(), &state_labels); - } - retired -} - -fn retire_drive_topology_api_metric_series(key: &DriveTopologyApiKey) -> usize { - let labels = [ - (SERVER_LABEL, Cow::Owned(key.0.clone())), - (DRIVE_LABEL, Cow::Owned(key.1.clone())), - (POOL_INDEX_LABEL, Cow::Owned(key.2.clone())), - (SET_INDEX_LABEL, Cow::Owned(key.3.clone())), - (DRIVE_INDEX_LABEL, Cow::Owned(key.4.clone())), - (DRIVE_API_LABEL, Cow::Owned(key.5.clone())), - ]; - retire_metric_series(&DRIVE_API_CALLS_MD.get_full_metric_name(), &labels) - + retire_metric_series(&DRIVE_API_LATENCY_BY_API_MD.get_full_metric_name(), &labels) -} - fn scanner_cycle_bucket_drive_result_live_keys(stats: &ScannerRuntimeStats) -> HashSet { stats .current_cycle_bucket_drive_results @@ -846,6 +716,21 @@ fn stagger_duration(period: Duration, numerator: u32, denominator: u32) -> Durat Duration::from_nanos(u64::try_from(staggered_nanos).unwrap_or(u64::MAX)) } +fn storage_snapshot_metrics(scope: &'static str, collection_interval: Duration) -> StorageSnapshotMetrics { + let export_interval = Duration::from_secs( + get_env_opt_u64(ENV_OBS_METER_INTERVAL) + .filter(|value| *value > 0) + .unwrap_or(METER_INTERVAL), + ); + let max_age = collection_interval.max(export_interval).saturating_mul(3); + StorageSnapshotMetrics::new( + opentelemetry::global::meter("rustfs.storage"), + scope, + current_local_node_identity(), + max_age, + ) +} + fn metrics_interval(period: Duration, initial_delay: Duration) -> Interval { let mut interval = tokio::time::interval_at(Instant::now() + initial_delay, period); interval.set_missed_tick_behavior(MissedTickBehavior::Delay); @@ -1770,26 +1655,21 @@ pub fn init_metrics_runtime(token: CancellationToken) { let token_clone = token.clone(); tokio::spawn(async move { let mut interval = metrics_interval(cluster_interval, Duration::ZERO); - let mut objects_count_was_authoritative = false; - let mut buckets_count_was_authoritative = false; + let mut snapshot = storage_snapshot_metrics("cluster", cluster_interval); loop { tokio::select! { _ = interval.tick() => { run_metrics_collector_tick(health, MetricsCollectorTaskId::ClusterStats, "cluster_stats", async { - let (stats, cluster_health) = collect_cluster_and_health_stats().await; - if objects_count_was_authoritative && stats.objects_count.is_none() { - let labels: [(&'static str, Cow<'static, str>); 0] = []; - let _ = retire_metric_series(&CLUSTER_OBJECTS_TOTAL_MD.get_full_metric_name(), &labels); + let collection_started = std::time::Instant::now(); + if let Some(stats) = collect_cluster_storage_snapshot().await { + let mut metrics = collect_cluster_metrics(&stats.cluster); + metrics.extend(collect_cluster_health_metrics(&stats.health)); + metrics.extend(collect_cluster_drive_metrics(&stats.drives)); + metrics.extend(collect_erasure_set_metrics(&stats.erasure_sets)); + if let Err(error) = snapshot.replace_collected(metrics, collection_started) { + error!(event = EVENT_METRICS_RUNTIME_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_METRICS_RUNTIME, collector = "cluster_stats", result = "invalid_snapshot", error = %error, "storage metrics snapshot rejected"); + } } - if buckets_count_was_authoritative && stats.buckets_count.is_none() { - let labels: [(&'static str, Cow<'static, str>); 0] = []; - let _ = retire_metric_series(&CLUSTER_BUCKETS_TOTAL_MD.get_full_metric_name(), &labels); - } - objects_count_was_authoritative = stats.objects_count.is_some(); - buckets_count_was_authoritative = stats.buckets_count.is_some(); - let mut metrics = collect_cluster_metrics(&stats); - metrics.extend(collect_cluster_health_metrics(&cluster_health)); - report_metrics(&metrics); }).await; } _ = token_clone.cancelled() => { @@ -1829,11 +1709,6 @@ pub fn init_metrics_runtime(token: CancellationToken) { 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)); } @@ -1994,63 +1869,19 @@ pub fn init_metrics_runtime(token: CancellationToken) { let token_clone = token.clone(); tokio::spawn(async move { let mut interval = metrics_interval(node_interval, Duration::ZERO); - let mut prev_drive_basic_keys: HashSet = HashSet::new(); - let mut prev_drive_info_keys: HashSet = HashSet::new(); - let mut prev_drive_topology_keys: HashSet = HashSet::new(); - let mut prev_drive_topology_api_keys: HashSet = HashSet::new(); - let mut has_seen_drive_info_snapshot = false; + let mut snapshot = storage_snapshot_metrics("local", node_interval); loop { tokio::select! { _ = interval.tick() => { run_metrics_collector_tick(health, MetricsCollectorTaskId::NodeDiskStats, "node_disk_stats", async { - let (disk_stats, drive_stats, drive_counts) = collect_disk_and_system_drive_runtime_stats().await; - let current_drive_info_keys = drive_info_live_keys(&drive_stats); - let current_drive_basic_keys = drive_basic_live_keys(&drive_stats); - let current_drive_topology_keys = drive_topology_live_keys(&drive_stats); - let current_drive_topology_api_keys = drive_topology_api_live_keys(&drive_stats); - let retire_drive_info_keys = if has_seen_drive_info_snapshot { - prev_drive_info_keys.difference(¤t_drive_info_keys).cloned().collect::>() - } else { - Vec::new() - }; - let retire_drive_basic_keys = if has_seen_drive_info_snapshot { - prev_drive_basic_keys.difference(¤t_drive_basic_keys).cloned().collect::>() - } else { - Vec::new() - }; - let retire_drive_topology_keys = if has_seen_drive_info_snapshot { - prev_drive_topology_keys.difference(¤t_drive_topology_keys).cloned().collect::>() - } else { - Vec::new() - }; - let retire_drive_topology_api_keys = if has_seen_drive_info_snapshot { - prev_drive_topology_api_keys - .difference(¤t_drive_topology_api_keys) - .cloned() - .collect::>() - } else { - Vec::new() - }; - prev_drive_info_keys = current_drive_info_keys; - prev_drive_basic_keys = current_drive_basic_keys; - prev_drive_topology_keys = current_drive_topology_keys; - prev_drive_topology_api_keys = current_drive_topology_api_keys; - has_seen_drive_info_snapshot = true; - let mut metrics = collect_node_metrics(&disk_stats); - metrics.extend(collect_drive_runtime_detailed_metrics(&drive_stats)); - metrics.extend(collect_drive_count_metrics(&drive_counts)); - report_metrics(&metrics); - for key in retire_drive_info_keys { - let _ = retire_drive_info_metric_series(&key); - } - for key in retire_drive_basic_keys { - let _ = retire_drive_basic_metric_series(&key); - } - for key in retire_drive_topology_keys { - let _ = retire_drive_topology_metric_series(&key); - } - for key in retire_drive_topology_api_keys { - let _ = retire_drive_topology_api_metric_series(&key); + let collection_started = std::time::Instant::now(); + if let Some((disk_stats, drive_stats, drive_counts)) = collect_disk_and_system_drive_runtime_stats().await { + let mut metrics = collect_node_metrics(&disk_stats); + metrics.extend(collect_drive_runtime_detailed_metrics(&drive_stats)); + metrics.extend(collect_drive_count_metrics(&drive_counts)); + if let Err(error) = snapshot.replace_collected(metrics, collection_started) { + error!(event = EVENT_METRICS_RUNTIME_STATE, component = LOG_COMPONENT_OBS, subsystem = LOG_SUBSYSTEM_METRICS_RUNTIME, collector = "node_disk_stats", result = "invalid_snapshot", error = %error, "storage metrics snapshot rejected"); + } } }).await; } @@ -2777,24 +2608,6 @@ mod tests { (server.to_string(), target_id.to_string(), target_type.to_string()) } - fn drive_info_stat(disk_id: &str) -> DriveRuntimeDetailedStats { - DriveRuntimeDetailedStats { - pool_index: Some("0".to_string()), - set_index: Some("1".to_string()), - drive_index: Some("2".to_string()), - disk_id: Some(disk_id.to_string()), - runtime_state: Some("online".to_string()), - api_calls: vec![("read_all".to_string(), 1)], - api_latency_by_api_micros: vec![("write_all".to_string(), 2)], - stats: crate::metrics::DriveDetailedStats { - server: "server-a".to_string(), - drive: "/data1".to_string(), - ..Default::default() - }, - ..Default::default() - } - } - fn scanner_stats_with_last_result(bucket: &str) -> ScannerRuntimeStats { ScannerRuntimeStats { server: "server-a".to_string(), @@ -2834,69 +2647,6 @@ mod tests { } } - #[test] - fn drive_info_live_keys_detect_disk_identity_replacement() { - let previous = drive_info_live_keys(&[drive_info_stat("disk-old")]); - let current = drive_info_live_keys(&[drive_info_stat("disk-new")]); - let retired = previous.difference(¤t).cloned().collect::>(); - - assert!(current.contains(&( - "server-a".to_string(), - "/data1".to_string(), - "0".to_string(), - "1".to_string(), - "2".to_string(), - "disk-new".to_string(), - ))); - assert!(retired.contains(&( - "server-a".to_string(), - "/data1".to_string(), - "0".to_string(), - "1".to_string(), - "2".to_string(), - "disk-old".to_string(), - ))); - } - - #[test] - fn drive_topology_keys_detect_removed_drives() { - let previous = drive_topology_live_keys(&[drive_info_stat("disk-old")]); - let current = drive_topology_live_keys(&[]); - let retired = previous.difference(¤t).cloned().collect::>(); - - assert!(retired.contains(&( - "server-a".to_string(), - "/data1".to_string(), - "0".to_string(), - "1".to_string(), - "2".to_string(), - ))); - } - - #[test] - fn drive_topology_api_keys_detect_removed_drives() { - let previous = drive_topology_api_live_keys(&[drive_info_stat("disk-old")]); - let current = drive_topology_api_live_keys(&[]); - let retired = previous.difference(¤t).cloned().collect::>(); - - assert!(retired.contains(&( - "server-a".to_string(), - "/data1".to_string(), - "0".to_string(), - "1".to_string(), - "2".to_string(), - "read_all".to_string(), - ))); - assert!(retired.contains(&( - "server-a".to_string(), - "/data1".to_string(), - "0".to_string(), - "1".to_string(), - "2".to_string(), - "write_all".to_string(), - ))); - } - #[test] fn scanner_last_bucket_drive_result_keys_detect_superseded_cycle_results() { let previous = scanner_cycle_bucket_drive_result_live_keys(&scanner_stats_with_last_result("photos")); diff --git a/crates/obs/src/metrics/schema/system_drive.rs b/crates/obs/src/metrics/schema/system_drive.rs index f67449281..f139acd96 100644 --- a/crates/obs/src/metrics/schema/system_drive.rs +++ b/crates/obs/src/metrics/schema/system_drive.rs @@ -34,6 +34,17 @@ pub const STATE_LABEL: &str = "state"; /// All drive-related labels pub const ALL_DRIVE_LABELS: [&str; 2] = [SERVER_LABEL, DRIVE_LABEL]; +/// Physical disk identity separates counter resets caused by disk replacement. +pub const DRIVE_COUNTER_LABELS: [&str; 3] = [SERVER_LABEL, DRIVE_LABEL, DISK_ID_LABEL]; +pub const DRIVE_TOPOLOGY_API_COUNTER_LABELS: [&str; 7] = [ + SERVER_LABEL, + DRIVE_LABEL, + POOL_INDEX_LABEL, + SET_INDEX_LABEL, + DRIVE_INDEX_LABEL, + API_LABEL, + DISK_ID_LABEL, +]; /// Drive labels with erasure-set topology. pub const DRIVE_TOPOLOGY_LABELS: [&str; 5] = [ SERVER_LABEL, @@ -70,6 +81,15 @@ pub const DRIVE_TOPOLOGY_API_LABELS: [&str; 6] = [ API_LABEL, ]; +pub static DRIVE_PRESENT_MD: LazyLock = LazyLock::new(|| { + new_gauge_md( + MetricName::Custom("present".to_string()), + "Configured local drive slot", + &ALL_DRIVE_LABELS, + subsystems::SYSTEM_DRIVE, + ) +}); + pub static DRIVE_INFO_MD: LazyLock = LazyLock::new(|| { new_gauge_md( MetricName::Custom("info".to_string()), @@ -119,7 +139,7 @@ pub static DRIVE_API_CALLS_MD: LazyLock = LazyLock::new(|| { new_counter_md( MetricName::Custom("api_calls_total".to_string()), "Total drive API calls by operation", - &DRIVE_TOPOLOGY_API_LABELS, + &DRIVE_TOPOLOGY_API_COUNTER_LABELS, subsystems::SYSTEM_DRIVE, ) }); @@ -209,7 +229,7 @@ pub static DRIVE_TIMEOUT_ERRORS_MD: LazyLock = LazyLock::new(| new_counter_md( MetricName::DriveTimeoutErrorsTotal, "Total timeout errors on a drive", - &ALL_DRIVE_LABELS[..], + &DRIVE_COUNTER_LABELS[..], subsystems::SYSTEM_DRIVE, ) }); @@ -218,7 +238,7 @@ pub static DRIVE_IO_ERRORS_MD: LazyLock = LazyLock::new(|| { new_counter_md( MetricName::DriveIOErrorsTotal, "Total I/O errors on a drive", - &ALL_DRIVE_LABELS[..], + &DRIVE_COUNTER_LABELS[..], subsystems::SYSTEM_DRIVE, ) }); @@ -227,7 +247,7 @@ pub static DRIVE_AVAILABILITY_ERRORS_MD: LazyLock = LazyLock:: new_counter_md( MetricName::DriveAvailabilityErrorsTotal, "Total availability errors (I/O errors, timeouts) on a drive", - &ALL_DRIVE_LABELS[..], + &DRIVE_COUNTER_LABELS[..], subsystems::SYSTEM_DRIVE, ) }); @@ -263,7 +283,7 @@ pub static DRIVE_WRITES_TOTAL_MD: LazyLock = LazyLock::new(|| new_counter_md( MetricName::DriveWritesTotal, "Total successful write operations on a drive", - &ALL_DRIVE_LABELS[..], + &DRIVE_COUNTER_LABELS[..], subsystems::SYSTEM_DRIVE, ) }); @@ -272,7 +292,7 @@ pub static DRIVE_DELETES_TOTAL_MD: LazyLock = LazyLock::new(|| new_counter_md( MetricName::DriveDeletesTotal, "Total successful delete operations on a drive", - &ALL_DRIVE_LABELS[..], + &DRIVE_COUNTER_LABELS[..], subsystems::SYSTEM_DRIVE, ) }); diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs index 1a2021b26..09eb63aca 100644 --- a/crates/obs/src/metrics/stats_collector.rs +++ b/crates/obs/src/metrics/stats_collector.rs @@ -18,6 +18,7 @@ //! RustFS internal sources (storage layer, bucket monitor, system info) //! and convert them to the Stats structs used by collectors. +use crate::metrics::collectors::cluster_drive::ClusterDriveStats; use crate::metrics::collectors::scanner::{ScannerActiveBucketDriveStats, ScannerBucketDriveResultStats, ScannerSourceWorkStats}; use crate::metrics::collectors::{ ApiRequestMetricSupport, ApiRequestStats, BucketReplicationBacklogStats, BucketReplicationBandwidthStats, @@ -487,13 +488,51 @@ pub struct ProcessMetricBundle { pub disk_write_bytes: u64, } +pub(crate) struct ClusterStorageSnapshot { + pub cluster: ClusterStats, + pub health: ClusterHealthStats, + pub drives: Vec, + pub erasure_sets: Vec, +} + +fn cluster_drive_stats_from_storage(storage: &ObsStorageInfo, local_server: &str) -> Vec { + storage + .disks + .iter() + .map(|disk| { + let (capacity_state, capacity_age_seconds) = disk_capacity_observation_state( + disk.capacity_observation_source.as_deref(), + disk.capacity_observation_age_seconds, + ); + ClusterDriveStats { + server: drive_server_label(&disk.endpoint, local_server), + drive: disk.drive_path.clone(), + pool_index: disk_topology_label(disk.pool_index).unwrap_or_default(), + set_index: disk_topology_label(disk.set_index).unwrap_or_default(), + drive_index: disk_topology_label(disk.disk_index).unwrap_or_default(), + disk_id: non_empty_disk_id(&disk.uuid).unwrap_or_default(), + runtime_state: disk.runtime_state.as_deref().unwrap_or("unknown").to_ascii_lowercase(), + offline_duration_seconds: disk.offline_duration_seconds, + capacity_state, + capacity_age_seconds, + total_bytes: disk.total_space, + used_bytes: disk.used_space, + free_bytes: disk.available_space, + } + }) + .collect() +} + /// Collect cluster and cluster-health statistics from a single storage snapshot. -pub async fn collect_cluster_and_health_stats() -> (ClusterStats, ClusterHealthStats) { - let Some(store) = resolve_obs_object_store_handle() else { - return (ClusterStats::default(), ClusterHealthStats::default()); - }; +pub(crate) async fn collect_cluster_storage_snapshot() -> Option { + let store = resolve_obs_object_store_handle()?; let storage_info = StorageAdminApi::storage_info(store.as_ref()).await; + if storage_info.disks.is_empty() && storage_info.backend.drives_per_set.is_empty() { + return None; + } + let drives = cluster_drive_stats_from_storage(&storage_info, ¤t_local_node_identity()); + let erasure_sets = erasure_set_stats_from_backend(&storage_info, &storage_info.backend); let raw_capacity: u64 = storage_info.disks.iter().map(|d| d.total_space).sum(); let usable_capacity = obs_total_usable_capacity_bytes(&storage_info); let free = obs_total_usable_capacity_free_bytes(&storage_info); @@ -536,8 +575,8 @@ pub async fn collect_cluster_and_health_stats() -> (ClusterStats, ClusterHealthS } } - ( - ClusterStats { + Some(ClusterStorageSnapshot { + cluster: ClusterStats { raw_capacity_bytes: raw_capacity, usable_capacity_bytes: usable_capacity, used_bytes: used, @@ -547,12 +586,22 @@ pub async fn collect_cluster_and_health_stats() -> (ClusterStats, ClusterHealthS objects_count, buckets_count, }, - ClusterHealthStats { + health: ClusterHealthStats { drives_offline_count: offline, drives_online_count: online, drives_count: storage_info.disks.len() as u64, }, - ) + drives, + erasure_sets, + }) +} + +/// Collect cluster statistics using the same observer snapshot as topology and health. +pub async fn collect_cluster_and_health_stats() -> (ClusterStats, ClusterHealthStats) { + collect_cluster_storage_snapshot() + .await + .map(|snapshot| (snapshot.cluster, snapshot.health)) + .unwrap_or_default() } /// Collect cluster statistics from the storage layer. @@ -763,23 +812,31 @@ pub fn collect_system_memory_stats() -> MemoryStats { /// Collect node disk stats and drive stats from a single storage snapshot. pub async fn collect_disk_and_system_drive_stats() -> (Vec, Vec, DriveCountStats) { - let (disk_stats, drive_stats, drive_count_stats) = collect_disk_and_system_drive_runtime_stats().await; + let (disk_stats, drive_stats, drive_count_stats) = collect_disk_and_system_drive_runtime_stats().await.unwrap_or_default(); (disk_stats, drive_stats.into_iter().map(|stat| stat.stats).collect(), drive_count_stats) } pub(crate) async fn collect_disk_and_system_drive_runtime_stats() --> (Vec, Vec, DriveCountStats) { - let Some(store) = resolve_obs_object_store_handle() else { - return (Vec::new(), Vec::new(), DriveCountStats::default()); - }; +-> Option<(Vec, Vec, DriveCountStats)> { + let store = resolve_obs_object_store_handle()?; + let storage_info = StorageAdminApi::local_storage_info(store.as_ref()).await; + Some(local_drive_stats_from_storage(&storage_info, ¤t_local_node_identity())) +} - let storage_info = StorageAdminApi::storage_info(store.as_ref()).await; - let local_server = current_local_node_identity(); +fn local_drive_stats_from_storage( + storage_info: &ObsStorageInfo, + local_server: &str, +) -> (Vec, Vec, DriveCountStats) { let disk_stats = storage_info .disks .iter() + .filter(|disk| disk.local) + .filter(|disk| { + disk_capacity_observation_state(disk.capacity_observation_source.as_deref(), disk.capacity_observation_age_seconds).0 + != CAPACITY_OBSERVATION_MISSING + }) .map(|disk| DiskStats { - server: drive_server_label(&disk.endpoint, &local_server), + server: drive_server_label(&disk.endpoint, local_server), drive: disk.drive_path.clone(), total_bytes: disk.total_space, used_bytes: disk.used_space, @@ -792,6 +849,7 @@ pub(crate) async fn collect_disk_and_system_drive_runtime_stats() let drive_stats = storage_info .disks .iter() + .filter(|disk| disk.local) .map(|disk| { let is_online = disk_is_online_for_metrics(disk.state.as_str(), disk.runtime_state.as_deref()); let (capacity_observation_state, capacity_observation_age_seconds) = disk_capacity_observation_state( @@ -832,7 +890,7 @@ pub(crate) async fn collect_disk_and_system_drive_runtime_stats() }) .unwrap_or_default(), stats: DriveDetailedStats { - server: drive_server_label(&disk.endpoint, &local_server), + server: drive_server_label(&disk.endpoint, local_server), drive: disk.drive_path.clone(), total_bytes: disk.total_space, used_bytes: disk.used_space, @@ -1714,6 +1772,43 @@ mod tests { info } + #[test] + fn local_details_exclude_remote_copies_but_keep_offline_configured_slots() { + let mut info = storage_info_with_one_online_disk(); + info.disks[0].local = true; + info.disks[0].endpoint = "http://owner:9000/data".into(); + info.disks[0].drive_path = "/data".into(); + info.disks[0].uuid = "disk-old".into(); + info.disks[0].capacity_observation_source = Some("live_probe".into()); + let mut remote = info.disks[0].clone(); + remote.local = false; + remote.endpoint = "http://peer:9000/data".into(); + remote.pool_index = 1; + remote.uuid = "peer-disk".into(); + info.disks.push(remote); + let (disks, local, counts) = local_drive_stats_from_storage(&info, "owner:9000"); + assert_eq!(disks.len(), 1); + assert_eq!(local.len(), 1); + assert_eq!(local[0].stats.server, "owner:9000"); + assert_eq!(local[0].disk_id.as_deref(), Some("disk-old")); + assert_eq!(counts.total_count, 1); + let global = cluster_drive_stats_from_storage(&info, "owner:9000"); + assert_eq!(global.len(), 2); + assert_eq!(global[1].server, "peer:9000"); + assert_eq!(global[1].pool_index, "1"); + // A disconnected configured local slot has no disk ID, but is not removed. + info.disks[0].uuid.clear(); + info.disks[0].state = "offline".into(); + info.disks[0].runtime_state = Some("offline".into()); + info.disks[0].capacity_observation_source = None; + let (_, local, counts) = local_drive_stats_from_storage(&info, "owner:9000"); + assert_eq!(local.len(), 1); + assert_eq!(local[0].disk_id, None); + assert_eq!(local[0].stats.capacity_observation_state, "missing"); + assert_eq!(counts.offline_count, 1); + assert_eq!(counts.total_count, 1); + } + #[test] fn bucket_usage_metrics_distinguish_unknown_from_confirmed_zero() { assert_eq!(bucket_usage_metric_values(None, "bucket"), (None, None)); diff --git a/crates/obs/src/metrics/storage_snapshot.rs b/crates/obs/src/metrics/storage_snapshot.rs new file mode 100644 index 000000000..4ba5f612a --- /dev/null +++ b/crates/obs/src/metrics/storage_snapshot.rs @@ -0,0 +1,407 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +//! Current storage snapshots, without cumulative SDK retention of removed series. + +use super::report::{PrometheusMetric, counter_value_from_f64}; +use super::schema::MetricType; +use opentelemetry::{KeyValue, metrics::Meter}; +use std::collections::HashMap; +use std::sync::{Arc, RwLock}; +use std::time::{Duration, Instant, SystemTime, UNIX_EPOCH}; + +pub(crate) const COLLECTION_SCOPE: &str = "collection_scope"; +pub(crate) const OBSERVER: &str = "observer"; +pub(crate) const LAST_SUCCESS: &str = "rustfs_storage_snapshot_last_success_timestamp_seconds"; +pub(crate) const MAX_AGE: &str = "rustfs_storage_snapshot_max_age_seconds"; + +#[derive(Debug, thiserror::Error)] +pub(crate) enum SnapshotError { + #[error("storage snapshots only support finite gauges and nonnegative counters")] + InvalidValue, + #[error("storage snapshot metric changed its instrument type")] + ChangedType, +} + +#[derive(Clone)] +struct Point { + value: f64, + counter: Option, + attributes: Vec, +} + +#[derive(Default)] +struct Snapshot { + points: HashMap>, + collected_at: Option, + timestamp_seconds: f64, + remaining_age: Duration, +} + +/// One collector owns each instance. Export callbacks only read its published memory. +pub(crate) struct StorageSnapshotMetrics { + meter: Meter, + snapshot: Arc>, + instruments: HashMap, + labels: Vec, + max_age: Duration, +} + +impl StorageSnapshotMetrics { + pub(crate) fn new(meter: Meter, scope: &'static str, observer: String, max_age: Duration) -> Self { + let snapshot = Arc::new(RwLock::new(Snapshot::default())); + let labels = vec![KeyValue::new(COLLECTION_SCOPE, scope), KeyValue::new(OBSERVER, observer)]; + let weak = Arc::downgrade(&snapshot); + let timestamp_labels = labels.clone(); + meter + .f64_observable_gauge(LAST_SUCCESS) + .with_description("Unix timestamp of the last successful storage snapshot collection") + .with_callback(move |observer| { + let Some(snapshot) = weak.upgrade() else { return }; + let Ok(snapshot) = snapshot.read() else { return }; + if snapshot.collected_at.is_some() { + observer.observe(snapshot.timestamp_seconds, ×tamp_labels); + } + }) + .build(); + let weak = Arc::downgrade(&snapshot); + let age_labels = labels.clone(); + meter + .f64_observable_gauge(MAX_AGE) + .with_description("Validity budget in seconds remaining after storage snapshot collection completed") + .with_callback(move |observer| { + let Some(snapshot) = weak.upgrade() else { return }; + let Ok(snapshot) = snapshot.read() else { return }; + if snapshot.collected_at.is_some() { + observer.observe(snapshot.remaining_age.as_secs_f64(), &age_labels); + } + }) + .build(); + Self { + meter, + snapshot, + instruments: HashMap::new(), + labels, + max_age, + } + } + + #[cfg(test)] + fn replace(&mut self, metrics: Vec) -> Result<(), SnapshotError> { + self.replace_collected(metrics, Instant::now()) + } + + pub(crate) fn replace_collected( + &mut self, + metrics: Vec, + collection_started: Instant, + ) -> Result<(), SnapshotError> { + // Validate the entire update before changing membership or publishing any values. + let mut types = self.instruments.clone(); + for metric in &metrics { + match metric.metric_type { + MetricType::Gauge if metric.value.is_finite() => (), + MetricType::Counter if counter_value_from_f64(metric.value).is_some() => (), + _ => return Err(SnapshotError::InvalidValue), + } + if types + .insert(metric.name.to_string(), metric.metric_type) + .is_some_and(|kind| kind != metric.metric_type) + { + return Err(SnapshotError::ChangedType); + } + } + + let mut points: HashMap> = HashMap::new(); + for metric in metrics { + let name = metric.name.to_string(); + if !self.instruments.contains_key(&name) { + self.register(&name, metric.metric_type, metric.help.as_ref())?; + self.instruments.insert(name.clone(), metric.metric_type); + } + let mut attributes = self.labels.clone(); + attributes.extend( + metric + .labels + .iter() + .map(|(key, value)| KeyValue::new(*key, value.to_string())), + ); + let point = Point { + value: metric.value, + counter: (metric.metric_type == MetricType::Counter) + .then(|| counter_value_from_f64(metric.value)) + .flatten(), + attributes, + }; + points.entry(name).or_default().push(point); + } + let timestamp_seconds = SystemTime::now().duration_since(UNIX_EPOCH).unwrap_or_default().as_secs_f64(); + let mut snapshot = self.snapshot.write().unwrap_or_else(|poisoned| poisoned.into_inner()); + *snapshot = Snapshot { + points, + collected_at: Some(collection_started), + timestamp_seconds, + // The publication timestamp withdraws cached fields from the previous + // snapshot. Deduct collection time separately so a slow RPC or usage + // read cannot make an old observation fresh merely by completing. + remaining_age: self.max_age.saturating_sub(collection_started.elapsed()), + }; + Ok(()) + } + + fn register(&self, name: &str, kind: MetricType, description: &str) -> Result<(), SnapshotError> { + let weak = Arc::downgrade(&self.snapshot); + let name_owned = name.to_string(); + let max_age = self.max_age; + match kind { + MetricType::Counter => { + self.meter + .u64_observable_counter(name.to_string()) + .with_description(description.to_string()) + .with_callback(move |observer| { + let Some(snapshot) = weak.upgrade() else { return }; + let Ok(snapshot) = snapshot.read() else { return }; + if snapshot.collected_at.is_none_or(|when| when.elapsed() > max_age) { + return; + } + if let Some(points) = snapshot.points.get(&name_owned) { + for point in points { + if let Some(value) = point.counter { + observer.observe(value, &point.attributes); + } + } + } + }) + .build(); + } + MetricType::Gauge => { + self.meter + .f64_observable_gauge(name.to_string()) + .with_description(description.to_string()) + .with_callback(move |observer| { + let Some(snapshot) = weak.upgrade() else { return }; + let Ok(snapshot) = snapshot.read() else { return }; + if snapshot.collected_at.is_none_or(|when| when.elapsed() > max_age) { + return; + } + if let Some(points) = snapshot.points.get(&name_owned) { + for point in points { + observer.observe(point.value, &point.attributes); + } + } + }) + .build(); + } + MetricType::Histogram => return Err(SnapshotError::InvalidValue), + } + Ok(()) + } +} + +#[cfg(test)] +mod tests { + use super::*; + use opentelemetry::metrics::MeterProvider; + use opentelemetry_sdk::error::OTelSdkResult; + use opentelemetry_sdk::metrics::data::{AggregatedMetrics, MetricData, ResourceMetrics}; + use opentelemetry_sdk::metrics::exporter::PushMetricExporter; + use opentelemetry_sdk::metrics::{PeriodicReader, SdkMeterProvider, Temporality}; + use std::sync::Mutex; + + #[derive(Clone, Debug)] + struct ExportedPoint { + name: String, + value: f64, + attributes: Vec, + } + + #[derive(Clone, Debug, Default)] + struct Exporter(Arc>>); + + impl PushMetricExporter for Exporter { + async fn export(&self, metrics: &ResourceMetrics) -> OTelSdkResult { + let mut exported = self.0.lock().unwrap(); + exported.clear(); + for metric in metrics.scope_metrics().flat_map(|scope| scope.metrics()) { + match metric.data() { + AggregatedMetrics::F64(MetricData::Gauge(gauge)) => { + for point in gauge.data_points() { + exported.push(ExportedPoint { + name: metric.name().to_string(), + value: point.value(), + attributes: point.attributes().cloned().collect(), + }); + } + } + AggregatedMetrics::U64(MetricData::Sum(sum)) => { + assert!(sum.is_monotonic()); + for point in sum.data_points() { + exported.push(ExportedPoint { + name: metric.name().to_string(), + value: point.value() as f64, + attributes: point.attributes().cloned().collect(), + }); + } + } + data => panic!("unexpected storage aggregation: {data:?}"), + } + } + Ok(()) + } + + fn force_flush(&self) -> OTelSdkResult { + Ok(()) + } + fn shutdown_with_timeout(&self, _timeout: Duration) -> OTelSdkResult { + Ok(()) + } + fn temporality(&self) -> Temporality { + Temporality::Cumulative + } + } + + fn setup() -> (SdkMeterProvider, Exporter, StorageSnapshotMetrics) { + let exporter = Exporter::default(); + let reader = PeriodicReader::builder(exporter.clone()) + .with_interval(Duration::from_secs(3600)) + .build(); + let provider = SdkMeterProvider::builder().with_reader(reader).build(); + let snapshot = + StorageSnapshotMetrics::new(provider.meter("storage-test"), "local", "node1:9000".into(), Duration::from_secs(30)); + (provider, exporter, snapshot) + } + + fn drive(name: &'static str, kind: MetricType, id: &'static str, value: f64) -> PrometheusMetric { + PrometheusMetric::new(name, kind, "test", value) + .with_label("server", "node1:9000") + .with_label("drive", "/data") + .with_label("disk_id", id) + } + + fn flush(provider: &SdkMeterProvider, exporter: &Exporter) -> Vec { + // The SDK skips export entirely when no callback observed any point. + exporter.0.lock().unwrap().clear(); + provider.force_flush().unwrap(); + exporter.0.lock().unwrap().clone() + } + + #[test] + fn actual_sdk_drops_removed_gauges_counters_and_replaced_disk_identity() { + let (provider, exporter, mut snapshot) = setup(); + snapshot + .replace(vec![ + drive("test_info", MetricType::Gauge, "old", 1.0), + drive("test_calls", MetricType::Counter, "old", 7.0), + ]) + .unwrap(); + let first = flush(&provider, &exporter); + assert!(first.iter().any(|point| point.name == "test_calls" && point.value == 7.0)); + snapshot + .replace(vec![drive("test_info", MetricType::Gauge, "new", 1.0)]) + .unwrap(); + let second = flush(&provider, &exporter); + assert!(!second.iter().any(|point| point.name == "test_calls")); + let info = second.iter().filter(|point| point.name == "test_info").collect::>(); + assert_eq!(info.len(), 1); + assert!(info[0].attributes.contains(&KeyValue::new("disk_id", "new"))); + assert!(info[0].attributes.contains(&KeyValue::new(COLLECTION_SCOPE, "local"))); + assert!(info[0].attributes.contains(&KeyValue::new(OBSERVER, "node1:9000"))); + snapshot.replace(Vec::new()).unwrap(); + assert!( + flush(&provider, &exporter) + .iter() + .all(|point| point.name == LAST_SUCCESS || point.name == MAX_AGE) + ); + provider.shutdown().unwrap(); + } + + #[test] + fn actual_sdk_preserves_counter_resets_instead_of_accumulating_old_absolute_values() { + let (provider, exporter, mut snapshot) = setup(); + for value in [7.0, 2.0, 5.0] { + snapshot + .replace(vec![drive("test_calls", MetricType::Counter, "disk", value)]) + .unwrap(); + assert_eq!( + flush(&provider, &exporter) + .iter() + .find(|point| point.name == "test_calls") + .unwrap() + .value, + value + ); + } + provider.shutdown().unwrap(); + } + + #[test] + fn failed_or_stalled_collection_cannot_refresh_last_success_or_export_old_points() { + let (provider, exporter, mut snapshot) = setup(); + snapshot + .replace(vec![drive("test_health", MetricType::Gauge, "disk", 1.0)]) + .unwrap(); + let first = flush(&provider, &exporter); + let timestamp = first.iter().find(|point| point.name == LAST_SUCCESS).unwrap().value; + assert!( + snapshot + .replace(vec![drive("test_health", MetricType::Gauge, "disk", f64::NAN)]) + .is_err() + ); + assert!( + snapshot + .replace(vec![drive("test_health", MetricType::Counter, "disk", 3.0)]) + .is_err() + ); + assert!( + snapshot + .replace(vec![ + drive("new_name", MetricType::Gauge, "disk", 1.0), + drive("new_name", MetricType::Counter, "disk", 1.0), + ]) + .is_err() + ); + let second = flush(&provider, &exporter); + assert_eq!(second.iter().find(|point| point.name == LAST_SUCCESS).unwrap().value, timestamp); + assert!(second.iter().any(|point| point.name == "test_health" && point.value == 1.0)); + snapshot.snapshot.write().unwrap().collected_at = Some(Instant::now() - Duration::from_secs(31)); + let expired = flush(&provider, &exporter); + assert!(!expired.iter().any(|point| point.name == "test_health")); + assert_eq!(expired.iter().find(|point| point.name == LAST_SUCCESS).unwrap().value, timestamp); + drop(snapshot); + assert!(flush(&provider, &exporter).is_empty()); + provider.shutdown().unwrap(); + } + + #[test] + fn a_slow_successful_collection_cannot_publish_an_already_expired_observation() { + let (provider, exporter, mut snapshot) = setup(); + snapshot + .replace_collected( + vec![drive("test_health", MetricType::Gauge, "disk", 1.0)], + Instant::now() - Duration::from_secs(31), + ) + .expect("publish a completed but slow snapshot"); + let points = flush(&provider, &exporter); + assert!(!points.iter().any(|point| point.name == "test_health")); + assert_eq!( + points + .iter() + .find(|point| point.name == MAX_AGE) + .expect("validity budget") + .value, + 0.0 + ); + provider.shutdown().expect("shutdown test provider"); + } +} diff --git a/docker-compose.yml b/docker-compose.yml index a28946575..bc76cdadf 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -40,6 +40,7 @@ services: - RUSTFS_OBS_LOGGER_LEVEL=info - RUSTFS_TLS_PATH=/opt/tls - RUSTFS_OBS_ENDPOINT=http://otel-collector:4318 + - OTEL_RESOURCE_ATTRIBUTES=${OTEL_RESOURCE_ATTRIBUTES:-rustfs.cluster.id=rustfs-dev} volumes: - ./deploy/data/pro:/data - ./deploy/logs:/app/logs diff --git a/docs/README.md b/docs/README.md index 872b5b9be..35d3324af 100644 --- a/docs/README.md +++ b/docs/README.md @@ -21,3 +21,6 @@ operators should start with: Other runbooks remain grouped by filename in [`operations/`](operations/); architecture pages link to the relevant runbook where a cross-boundary procedure is required. + +For storage dashboards, see [Storage metrics and observer selection](operations/storage-metrics.md): +drive ownership, snapshot freshness, counter queries, and rolling upgrades. diff --git a/docs/operations/storage-metrics.md b/docs/operations/storage-metrics.md new file mode 100644 index 000000000..d3b22bea4 --- /dev/null +++ b/docs/operations/storage-metrics.md @@ -0,0 +1,142 @@ +# Storage metrics and observer selection + +**Use this when:** configuring storage dashboards, migrating from replicated +per-drive metrics, or diagnosing stale and missing disk observations. + +## Ownership and identity + +The storage collectors have two distinct scopes: + +| Scope | Metric families | Meaning | +|---|---|---| +| `collection_scope="local"` | `rustfs_system_drive_*`, `rustfs_node_disk_*` | Detailed metrics for drives owned by the reporting node; includes configured offline slots. Drive counts are local counts. | +| `collection_scope="cluster"` | Cluster capacity, health, objects/buckets, erasure sets, and `rustfs_cluster_drive_*` | One reporting node's observation of the whole cluster. | + +`observer` identifies the reporting node. `server` identifies the drive owner; +`drive` alone is not a unique disk key. Keep `rustfs_cluster_id`, `server`, and +`drive` when grouping drives, and retain pool/set/drive indices for topology. +Drive counters also carry `disk_id`, so a physical replacement starts a separate +series. Unknown IDs are empty; `rustfs_system_drive_info` is emitted only when +the ID and topology are known. `rustfs_system_drive_present` preserves a configured +slot even when its disk is disconnected and its ID is unavailable. + +Global drive metrics contain membership, runtime state, capacity and its source. +They do not duplicate remote API counters, error counters, or detailed I/O metrics. +An unreachable peer remains in the global inventory with the storage layer's +unknown/offline state and stale/missing capacity provenance. Missing capacity is +omitted from per-drive byte metrics. Cluster capacity may still include cached +observations: check `capacity_stale_drives` and `capacity_missing_drives` before +interpreting it. The existing cluster offline count includes drives not currently +observed online; use global per-drive runtime states to distinguish unknown. + +## Configure the pipeline + +Give every node of a deployment the same stable, unique resource attribute: + +```bash +OTEL_RESOURCE_ATTRIBUTES=rustfs.cluster.id=production-a +``` + +Keep existing resource attributes in the comma-separated value. The example +Compose files accept `OTEL_RESOURCE_ATTRIBUTES` and default to the development +cluster ID `rustfs-dev`; set a distinct ID for each deployment. The Collector's +`resource_to_telemetry_conversion.enabled: true` promotes it to +`rustfs_cluster_id` on scraped samples. Prometheus `external_labels` are not a +substitute: they do not add a cluster label to the local time series. + +Use the shipped Collector configuration with `send_timestamps: true` and keep +Prometheus's default `honor_timestamps: true`. Load +`.docker/observability/prometheus-rules/rustfs-storage.yml`. Use one ingestion +route per deployment/node; scraping replicas of the same Collector requires a +separate HA deduplication policy. Keep node and Prometheus clocks synchronized. + +Storage export callbacks read the current snapshot from memory and stop exporting +removed series. They do not probe storage or run peer RPCs. A successful collection +updates `rustfs_storage_snapshot_last_success_timestamp_seconds`; OTLP export +alone does not update this value. The source age limit is three times the larger +of the collection interval and `RUSTFS_OBS_METER_INTERVAL` (with its normal default). +`rustfs_storage_snapshot_max_age_seconds` exports the remaining validity budget +after deducting collection time from that limit. A slow RPC or usage read therefore +cannot make an already expired observation fresh by completing. An unavailable +source leaves the last success unchanged; a stalled collector stops exporting its +old snapshot after the source age limit. + +A Collector may cache a point after RustFS stops exporting it. The recording rules +therefore require both a fresh source and a raw point timestamp at least as new as +that source's last successful collection. This also removes cached optional fields +and old disk IDs when a new snapshot no longer contains them. The timestamp check +must run on the raw selector, before label rewriting or recording the value; +otherwise PromQL can substitute the query evaluation time for the original sample +time. The comparison uses Prometheus millisecond precision so co-published points +are not excluded by submillisecond rounding. Current recording rules preserve the +original name in `source_metric`. +When adding a storage metric, add its matching rule to that file. + +Collection and export are asynchronous, not a transaction across instruments. A +snapshot published during an export can briefly mix adjacent observations or +withhold a value; the next complete export/collection converges. Source failure visibility is bounded +by the published maximum age plus the scrape/rule intervals. Cached capacity has +its own observation age and state; a fresh collection does not make that capacity +live. Bucket/object counts retain the existing scanner update delay. + +## Queries and dashboards + +The bundled dashboard requires the storage recording rules. Select one **Storage +cluster** and one fresh **Cluster observer**. Global panels use that observer's +complete view. If it expires, the panel shows no data; select another observer. +Do not sum replicated global totals or independently take maxima/minima of fields +from different observers: they may describe different moments or partitions. + +For example, select the observed cluster raw capacity: + +```promql +rustfs:storage:current{source_metric="rustfs_cluster_capacity_raw_total_bytes",rustfs_cluster_id="production-a",collection_scope="cluster",observer="node1:9000"} +``` + +For a per-drive API rate, select owner-scoped counters before `rate`, then require +a current observation. Sum the resulting rates only across the desired drives: + +```promql +sum by (rustfs_cluster_id, api) ( + rate(rustfs_system_drive_api_calls_total{rustfs_cluster_id="production-a",collection_scope="local"}[5m]) + and ignoring(source_metric) + rustfs:storage:current{source_metric="rustfs_system_drive_api_calls_total",rustfs_cluster_id="production-a",collection_scope="local"} +) +``` + +A missing series is unknown, not zero. Local totals include only fresh owners and +can decrease when a node becomes unreachable. Use a selected global observer for +configured topology and its explicit unknown/stale indicators. + +## Rolling upgrades and validation + +Install the rules, resource attribute, and dashboard together. During rolling +upgrades, old releases have no `collection_scope`; strict new-scope selectors +exclude their replicated series. Local panels initially contain only upgraded +owners. Global panels become available when an upgraded observer publishes a fresh +snapshot. Historical old-label series remain available for retrospective queries. +Do not combine old and new counter histories into one rate. + +Run the PromQL regression fixtures from `.docker/observability/tests`: + +```bash +promtool test rules storage-rules.test.yml +``` + +The native pipeline test in +`crates/e2e_test/src/storage_metric_ownership_test.rs` requires pinned Collector, +Prometheus, previous-release RustFS, and current RustFS executables. Set +`RUSTFS_OTELCOL_BINARY`, `RUSTFS_PROMETHEUS_BINARY`, +`RUSTFS_METRICS_BASELINE_BINARY`, and `CARGO_BIN_EXE_rustfs` to those files. +Optionally set `RUSTFS_METRICS_E2E_ARTIFACTS` to retain logs and Prometheus data. +Run only this external-tool test: + +```bash +cargo test --locked -p e2e_test storage_metric_ownership_pipeline -- --ignored --nocapture +``` + +The test first reproduces duplicated global details with the previous release, +rolls four nodes forward, verifies owner and disk identity, and checks node loss +and recovery while the Collector remains running. SDK unit tests cover removed +series, replacements, counter resets, invalid updates, and a stalled collection; +PromQL fixtures cover shared paths across clusters/pools and stale cached fields.