gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-09 13:46:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(metrics): prevent duplicate drive reporting across nodes (#7582)

Browse Source 
Co-authored-by: houseme <housemecn@gmail.com>
This commit is contained in:
cxymds
2026-09-09 19:00:20 +08:00
committed by GitHub
parent d9ae1654b6
commit b3b0d892bb
23 changed files with 2188 additions and 389 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.docker/compose/docker-compose.cluster.local-build.profiling-amd64.yml
+4
View File
@@ -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}
.docker/compose/docker-compose.cluster.local-build.yml
+4
View File
@@ -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}
.docker/compose/docker-compose.observability.yaml
+4
View File
@@ -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:
.docker/observability/README.md
+4
View File
@@ -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:
.docker/observability/docker-compose-example-for-rustfs.yml
+1
View File
@@ -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
.docker/observability/grafana/dashboards/rustfs.json
+101 -45
View File
@@ -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,
.docker/observability/prometheus-rules/rustfs-storage.yml
+575
View File
@@ -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))
.docker/observability/tests/storage-rules.test.yml
+189
View File
@@ -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
crates/e2e_test/README.md
+2
View File
@@ -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
crates/e2e_test/src/lib.rs
+3
View File
@@ -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;
crates/e2e_test/src/storage_metric_ownership_test.rs
+335
View File
@@ -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<T = ()> = Result<T, Box<dyn std::error::Error + Send + Sync>>;
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<PathBuf> {
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<u16> {
Ok(TcpListener::bind("127.0.0.1:0")?.local_addr()?.port())
}
fn start_tool(binary: &Path, args: &[String], log: &Path) -> TestResult<ToolProcess> {
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<Value> {
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::<u64>().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::<Vec<_>>();
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, &current).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(&current).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(())
}
crates/obs/src/metrics/collectors/cluster_drive.rs
+150
View File
@@ -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<u64>,
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<PrometheusMetric> {
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"))
);
}
}
crates/obs/src/metrics/collectors/mod.rs
+1
View File
@@ -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;
crates/obs/src/metrics/collectors/system_drive.rs
+77 -25
View File
@@ -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::<Vec<_>>();
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()
crates/obs/src/metrics/mod.rs
+1
View File
@@ -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::*;
crates/obs/src/metrics/report.rs
+1 -1
View File
@@ -69,7 +69,7 @@ fn describe_metric_once(name: &'static str, metric_type: MetricType, help: &Cow<
}
}
fn counter_value_from_f64(value: f64) -> Option<u64> {
pub(crate) fn counter_value_from_f64(value: f64) -> Option<u64> {
if !value.is_finite() || value < 0.0 {
return None;
}
crates/obs/src/metrics/scheduler.rs
+45 -295
View File
@@ -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<DriveInfoKey> {
stats.iter().filter_map(drive_info_key).collect()
}
fn drive_basic_live_keys(stats: &[DriveRuntimeDetailedStats]) -> HashSet<DriveBasicKey> {
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<DriveTopologyKey> {
stats.iter().filter_map(drive_topology_key).collect()
}
fn drive_topology_api_live_keys(stats: &[DriveRuntimeDetailedStats]) -> HashSet<DriveTopologyApiKey> {
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<DriveTopologyKey> {
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<DriveInfoKey> {
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<ScannerCycleBucketDriveResultKey> {
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<DriveBasicKey> = HashSet::new();
let mut prev_drive_info_keys: HashSet<DriveInfoKey> = HashSet::new();
let mut prev_drive_topology_keys: HashSet<DriveTopologyKey> = HashSet::new();
let mut prev_drive_topology_api_keys: HashSet<DriveTopologyApiKey> = 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(&current_drive_info_keys).cloned().collect::<Vec<_>>()
} else {
Vec::new()
};
let retire_drive_basic_keys = if has_seen_drive_info_snapshot {
prev_drive_basic_keys.difference(&current_drive_basic_keys).cloned().collect::<Vec<_>>()
} else {
Vec::new()
};
let retire_drive_topology_keys = if has_seen_drive_info_snapshot {
prev_drive_topology_keys.difference(&current_drive_topology_keys).cloned().collect::<Vec<_>>()
} else {
Vec::new()
};
let retire_drive_topology_api_keys = if has_seen_drive_info_snapshot {
prev_drive_topology_api_keys
.difference(&current_drive_topology_api_keys)
.cloned()
.collect::<Vec<_>>()
} 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(&current).cloned().collect::<HashSet<_>>();
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(&current).cloned().collect::<HashSet<_>>();
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(&current).cloned().collect::<HashSet<_>>();
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"));
crates/obs/src/metrics/schema/system_drive.rs
+26 -6
View File
@@ -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<MetricDescriptor> = 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<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
MetricName::Custom("info".to_string()),
@@ -119,7 +139,7 @@ pub static DRIVE_API_CALLS_MD: LazyLock<MetricDescriptor> = 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<MetricDescriptor> = 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<MetricDescriptor> = 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<MetricDescriptor> = 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<MetricDescriptor> = 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<MetricDescriptor> = LazyLock::new(||
new_counter_md(
MetricName::DriveDeletesTotal,
"Total successful delete operations on a drive",
&ALL_DRIVE_LABELS[..],
&DRIVE_COUNTER_LABELS[..],
subsystems::SYSTEM_DRIVE,
)
});
crates/obs/src/metrics/stats_collector.rs
+112 -17
View File
@@ -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<ClusterDriveStats>,
pub erasure_sets: Vec<ErasureSetStats>,
}
fn cluster_drive_stats_from_storage(storage: &ObsStorageInfo, local_server: &str) -> Vec<ClusterDriveStats> {
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<ClusterStorageSnapshot> {
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, &current_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<DiskStats>, Vec<DriveDetailedStats>, 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<DiskStats>, Vec<DriveRuntimeDetailedStats>, DriveCountStats) {
let Some(store) = resolve_obs_object_store_handle() else {
return (Vec::new(), Vec::new(), DriveCountStats::default());
};
-> Option<(Vec<DiskStats>, Vec<DriveRuntimeDetailedStats>, 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, &current_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<DiskStats>, Vec<DriveRuntimeDetailedStats>, 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));
crates/obs/src/metrics/storage_snapshot.rs
+407
View File
@@ -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<u64>,
attributes: Vec<KeyValue>,
}
#[derive(Default)]
struct Snapshot {
points: HashMap<String, Vec<Point>>,
collected_at: Option<Instant>,
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<RwLock<Snapshot>>,
instruments: HashMap<String, MetricType>,
labels: Vec<KeyValue>,
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, &timestamp_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<PrometheusMetric>) -> Result<(), SnapshotError> {
self.replace_collected(metrics, Instant::now())
}
pub(crate) fn replace_collected(
&mut self,
metrics: Vec<PrometheusMetric>,
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<String, Vec<Point>> = 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<KeyValue>,
}
#[derive(Clone, Debug, Default)]
struct Exporter(Arc<Mutex<Vec<ExportedPoint>>>);
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<ExportedPoint> {
// 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::<Vec<_>>();
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");
}
}
docker-compose.yml
+1
View File
@@ -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
docs/README.md
+3
View File
@@ -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.
docs/operations/storage-metrics.md
+142
View File
@@ -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.