gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-28 16:07:05 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(ecstore): harden runtime read-path quorum handling (#2872)

Browse Source
This commit is contained in:
houseme
2026-05-08 17:56:39 +08:00
committed by GitHub
parent 03045ff2e6
commit c90bfe2b23
21 changed files with 907 additions and 139 deletions
Download Patch File Download Diff File Expand all files Collapse all files
.docker/observability/grafana/dashboards/rustfs.json
+377
View File
@@ -461,6 +461,90 @@
"title": "Capacity", "title": "Capacity",
"type": "stat" "type": "stat"
}, },
{
"datasource": {
"type": "prometheus",
"uid": "${datasource}"
},
"fieldConfig": {
"defaults": {
"color": {
"mode": "thresholds"
},
"mappings": [],
"thresholds": {
"mode": "absolute",
"steps": [
{
"color": "green",
"value": 0
},
{
"color": "yellow",
"value": 1
},
{
"color": "red",
"value": 2
}
]
},
"unit": "none"
},
"overrides": []
},
"gridPos": {
"h": 5,
"w": 6,
"x": 6,
"y": 1
},
"id": 201,
"options": {
"colorMode": "value",
"graphMode": "none",
"justifyMode": "auto",
"orientation": "auto",
"percentChangeColorMode": "standard",
"reduceOptions": {
"calcs": [
"lastNotNull"
],
"fields": "",
"values": false
},
"showPercentChange": false,
"textMode": "value_and_name",
"wideLayout": true
},
"pluginVersion": "12.3.2",
"targets": [
{
"datasource": {
"type": "prometheus",
"uid": "${datasource}"
},
"editorMode": "code",
"expr": "sum(rustfs_cluster_capacity_stale_drives{job=~\"$job\"})",
"legendFormat": "Stale Drives",
"range": true,
"refId": "A"
},
{
"datasource": {
"type": "prometheus",
"uid": "${datasource}"
},
"editorMode": "code",
"expr": "sum(rustfs_cluster_capacity_missing_drives{job=~\"$job\"})",
"legendFormat": "Missing Drives",
"range": true,
"refId": "B"
}
],
"title": "Capacity Observation",
"type": "stat"
},
{ {
"datasource": { "datasource": {
"type": "prometheus", "type": "prometheus",
@@ -1927,6 +2011,299 @@
"title": "System Drive Usage", "title": "System Drive Usage",
"type": "timeseries" "type": "timeseries"
}, },
{
"datasource": {
"type": "prometheus",
"uid": "${datasource}"
},
"fieldConfig": {
"defaults": {
"color": {
"mode": "palette-classic"
},
"custom": {
"axisBorderShow": false,
"axisCenteredZero": false,
"axisColorMode": "text",
"axisLabel": "",
"axisPlacement": "auto",
"barAlignment": 0,
"drawStyle": "line",
"fillOpacity": 10,
"gradientMode": "none",
"hideFrom": {
"legend": false,
"tooltip": false,
"viz": false
},
"lineInterpolation": "smooth",
"lineWidth": 1,
"pointSize": 5,
"scaleDistribution": {
"type": "linear"
},
"showPoints": "never",
"spanNulls": false,
"stacking": {
"group": "A",
"mode": "none"
},
"thresholdsStyle": {
"mode": "off"
}
},
"mappings": [],
"thresholds": {
"mode": "absolute",
"steps": [
{
"color": "green",
"value": 0
}
]
},
"unit": "s"
},
"overrides": []
},
"gridPos": {
"h": 6,
"w": 12,
"x": 12,
"y": 42
},
"id": 202,
"options": {
"legend": {
"calcs": [
"lastNotNull",
"max"
],
"displayMode": "table",
"placement": "right",
"showLegend": true
},
"tooltip": {
"hideZeros": false,
"mode": "multi",
"sort": "desc"
}
},
"pluginVersion": "12.3.2",
"targets": [
{
"datasource": {
"type": "prometheus",
"uid": "${datasource}"
},
"editorMode": "code",
"expr": "max by (drive) (rustfs_system_drive_capacity_observation_age_seconds{job=~\"$job\", drive=~\"$drive\"})",
"legendFormat": "{{drive}}",
"range": true,
"refId": "A"
}
],
"title": "Drive Capacity Observation Age",
"type": "timeseries"
},
{
"datasource": {
"type": "prometheus",
"uid": "${datasource}"
},
"fieldConfig": {
"defaults": {
"color": {
"mode": "palette-classic"
},
"custom": {
"axisBorderShow": false,
"axisCenteredZero": true,
"axisColorMode": "text",
"axisPlacement": "auto",
"barAlignment": 0,
"drawStyle": "line",
"fillOpacity": 10,
"gradientMode": "none",
"hideFrom": {
"legend": false,
"tooltip": false,
"viz": false
},
"lineInterpolation": "stepAfter",
"lineWidth": 2,
"pointSize": 5,
"scaleDistribution": {
"type": "linear"
},
"showPoints": "never",
"spanNulls": false,
"stacking": {
"group": "A",
"mode": "none"
},
"thresholdsStyle": {
"mode": "off"
}
},
"mappings": [],
"thresholds": {
"mode": "absolute",
"steps": [
{
"color": "green",
"value": 0
}
]
},
"unit": "none"
},
"overrides": []
},
"gridPos": {
"h": 6,
"w": 12,
"x": 12,
"y": 48
},
"id": 203,
"options": {
"legend": {
"calcs": [
"lastNotNull"
],
"displayMode": "table",
"placement": "right",
"showLegend": true
},
"tooltip": {
"hideZeros": false,
"mode": "multi",
"sort": "desc"
}
},
"pluginVersion": "12.3.2",
"targets": [
{
"datasource": {
"type": "prometheus",
"uid": "${datasource}"
},
"editorMode": "code",
"expr": "max by (drive) (rustfs_system_drive_capacity_observation_state{job=~\"$job\", drive=~\"$drive\", state=\"stale\"})",
"legendFormat": "{{drive}} stale",
"range": true,
"refId": "A"
},
{
"datasource": {
"type": "prometheus",
"uid": "${datasource}"
},
"editorMode": "code",
"expr": "max by (drive) (rustfs_system_drive_capacity_observation_state{job=~\"$job\", drive=~\"$drive\", state=\"missing\"})",
"legendFormat": "{{drive}} missing",
"range": true,
"refId": "B"
}
],
"title": "Drive Capacity Observation State",
"type": "timeseries"
},
{
"datasource": {
"type": "prometheus",
"uid": "${datasource}"
},
"fieldConfig": {
"defaults": {
"color": {
"mode": "palette-classic"
},
"custom": {
"axisBorderShow": false,
"axisCenteredZero": true,
"axisColorMode": "text",
"axisPlacement": "auto",
"barAlignment": 0,
"drawStyle": "bars",
"fillOpacity": 80,
"gradientMode": "none",
"hideFrom": {
"legend": false,
"tooltip": false,
"viz": false
},
"lineInterpolation": "linear",
"lineWidth": 1,
"pointSize": 5,
"scaleDistribution": {
"type": "linear"
},
"showPoints": "never",
"spanNulls": false,
"stacking": {
"group": "A",
"mode": "normal"
},
"thresholdsStyle": {
"mode": "off"
}
},
"mappings": [],
"thresholds": {
"mode": "absolute",
"steps": [
{
"color": "green",
"value": 0
}
]
},
"unit": "short"
},
"overrides": []
},
"gridPos": {
"h": 7,
"w": 24,
"x": 0,
"y": 54
},
"id": 204,
"options": {
"legend": {
"calcs": [
"lastNotNull",
"max"
],
"displayMode": "table",
"placement": "right",
"showLegend": true
},
"tooltip": {
"hideZeros": false,
"mode": "multi",
"sort": "desc"
}
},
"pluginVersion": "12.3.2",
"targets": [
{
"datasource": {
"type": "prometheus",
"uid": "${datasource}"
},
"editorMode": "code",
"expr": "sum by (path_kind, operation, reason) (rate(rustfs_system_path_failures_total{job=~\"$job\"}[5m]))",
"legendFormat": "{{path_kind}} / {{operation}} / {{reason}}",
"range": true,
"refId": "A"
}
],
"title": "System Path Failures",
"type": "timeseries"
},
{ {
"datasource": { "datasource": {
"type": "prometheus", "type": "prometheus",
Cargo.lock
Generated
+1
View File
@@ -8535,6 +8535,7 @@ dependencies = [
"rustfs-credentials", "rustfs-credentials",
"rustfs-crypto", "rustfs-crypto",
"rustfs-ecstore", "rustfs-ecstore",
"rustfs-io-metrics",
"rustfs-madmin", "rustfs-madmin",
"rustfs-policy", "rustfs-policy",
"rustfs-utils", "rustfs-utils",
crates/ecstore/src/cache_value/metacache_set.rs
+10 -1
View File
@@ -85,7 +85,7 @@ impl Clone for ListPathRawOptions {
pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> disk::error::Result<()> { pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> disk::error::Result<()> {
if opts.disks.is_empty() { if opts.disks.is_empty() {
return Err(DiskError::other("list_path_raw: 0 drives provided")); return Err(DiskError::ErasureReadQuorum);
} }
let mut jobs: Vec<tokio::task::JoinHandle<std::result::Result<(), DiskError>>> = Vec::new(); let mut jobs: Vec<tokio::task::JoinHandle<std::result::Result<(), DiskError>>> = Vec::new();
@@ -415,6 +415,15 @@ mod tests {
use super::*; use super::*;
use rustfs_filemeta::MetacacheWriter; use rustfs_filemeta::MetacacheWriter;
#[tokio::test]
async fn list_path_raw_empty_disks_returns_read_quorum() {
let err = list_path_raw(CancellationToken::new(), ListPathRawOptions::default())
.await
.expect_err("empty drive list should fail");
assert_eq!(err, DiskError::ErasureReadQuorum);
}
#[tokio::test] #[tokio::test]
async fn peek_with_timeout_times_out_on_silent_reader() { async fn peek_with_timeout_times_out_on_silent_reader() {
let (_writer, reader) = tokio::io::duplex(64); let (_writer, reader) = tokio::io::duplex(64);
crates/ecstore/src/data_usage.rs
+26 -3
View File
@@ -15,7 +15,11 @@
pub mod local_snapshot; pub mod local_snapshot;
use crate::{ use crate::{
bucket::metadata_sys::get_replication_config, config::com::read_config, disk::DiskAPI, error::Error, store::ECStore, bucket::metadata_sys::get_replication_config,
config::com::read_config,
disk::DiskAPI,
error::{Error, classify_system_path_failure_reason},
store::ECStore,
store_api::ListOperations, store_api::ListOperations,
}; };
pub use local_snapshot::{ pub use local_snapshot::{
@@ -26,6 +30,7 @@ pub use local_snapshot::{
use rustfs_common::data_usage::{ use rustfs_common::data_usage::{
BucketTargetUsageInfo, BucketUsageInfo, DataUsageCache, DataUsageEntry, DataUsageInfo, DiskUsageStatus, SizeSummary, BucketTargetUsageInfo, BucketUsageInfo, DataUsageCache, DataUsageEntry, DataUsageInfo, DiskUsageStatus, SizeSummary,
}; };
use rustfs_io_metrics::record_system_path_failure;
use rustfs_utils::path::SLASH_SEPARATOR; use rustfs_utils::path::SLASH_SEPARATOR;
use std::{ use std::{
collections::{HashMap, HashSet, hash_map::Entry}, collections::{HashMap, HashSet, hash_map::Entry},
@@ -109,7 +114,16 @@ pub async fn load_data_usage_from_backend(store: Arc<ECStore>) -> Result<DataUsa
let buf: Vec<u8> = match read_config(store.clone(), &DATA_USAGE_OBJ_NAME_PATH).await { let buf: Vec<u8> = match read_config(store.clone(), &DATA_USAGE_OBJ_NAME_PATH).await {
Ok(data) => data, Ok(data) => data,
Err(e) => { Err(e) => {
error!("Failed to read data usage info from backend: {}", e); let reason = classify_system_path_failure_reason(&e);
record_system_path_failure("data_usage", "read_primary", reason);
error!(
path_kind = "data_usage",
operation = "read_primary",
reason,
object = %DATA_USAGE_OBJ_NAME_PATH.as_str(),
error = %e,
"system path read failed"
);
match read_config(store.clone(), format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()).as_str()).await { match read_config(store.clone(), format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()).as_str()).await {
Ok(data) => data, Ok(data) => data,
@@ -117,7 +131,16 @@ pub async fn load_data_usage_from_backend(store: Arc<ECStore>) -> Result<DataUsa
if e == Error::ConfigNotFound { if e == Error::ConfigNotFound {
return Ok(DataUsageInfo::default()); return Ok(DataUsageInfo::default());
} }
error!("Failed to read data usage info from backend: {}", e); let reason = classify_system_path_failure_reason(&e);
record_system_path_failure("data_usage", "read_backup", reason);
error!(
path_kind = "data_usage",
operation = "read_backup",
reason,
object = %format!("{}.bkp", DATA_USAGE_OBJ_NAME_PATH.as_str()),
error = %e,
"system path read failed"
);
return Err(Error::other(e)); return Err(Error::other(e));
} }
} }
crates/ecstore/src/disk/disk_store.rs
+43 -1
View File
@@ -31,7 +31,7 @@ use std::{
path::PathBuf, path::PathBuf,
sync::{ sync::{
Arc, Arc,
atomic::{AtomicI64, AtomicU32, Ordering}, atomic::{AtomicI64, AtomicU32, AtomicU64, Ordering},
}, },
time::Duration, time::Duration,
}; };
@@ -138,6 +138,14 @@ pub struct DiskHealthTracker {
pub offline_since_unix_secs: AtomicI64, pub offline_since_unix_secs: AtomicI64,
/// Last runtime state transition timestamp /// Last runtime state transition timestamp
pub last_transition_unix_secs: AtomicI64, pub last_transition_unix_secs: AtomicI64,
/// Last successfully probed total space in bytes
pub last_capacity_total: AtomicU64,
/// Last successfully probed used space in bytes
pub last_capacity_used: AtomicU64,
/// Last successfully probed free space in bytes
pub last_capacity_free: AtomicU64,
/// Last successful capacity probe timestamp
pub last_capacity_probe_unix_secs: AtomicI64,
} }
impl DiskHealthTracker { impl DiskHealthTracker {
@@ -158,6 +166,10 @@ impl DiskHealthTracker {
consecutive_successes: AtomicU32::new(0), consecutive_successes: AtomicU32::new(0),
offline_since_unix_secs: AtomicI64::new(0), offline_since_unix_secs: AtomicI64::new(0),
last_transition_unix_secs: AtomicI64::new(now / 1_000_000_000), last_transition_unix_secs: AtomicI64::new(now / 1_000_000_000),
last_capacity_total: AtomicU64::new(0),
last_capacity_used: AtomicU64::new(0),
last_capacity_free: AtomicU64::new(0),
last_capacity_probe_unix_secs: AtomicI64::new(0),
} }
} }
@@ -170,6 +182,28 @@ impl DiskHealthTracker {
self.last_success.store(now, Ordering::Relaxed); self.last_success.store(now, Ordering::Relaxed);
} }
pub fn record_capacity_probe(&self, total: u64, used: u64, free: u64) {
self.last_capacity_total.store(total, Ordering::Release);
self.last_capacity_used.store(used, Ordering::Release);
self.last_capacity_free.store(free, Ordering::Release);
self.last_capacity_probe_unix_secs
.store(current_unix_secs() as i64, Ordering::Release);
}
pub fn last_capacity_snapshot(&self) -> Option<(u64, u64, u64, u64)> {
let ts = self.last_capacity_probe_unix_secs.load(Ordering::Acquire);
if ts <= 0 {
return None;
}
Some((
self.last_capacity_total.load(Ordering::Acquire),
self.last_capacity_used.load(Ordering::Acquire),
self.last_capacity_free.load(Ordering::Acquire),
ts as u64,
))
}
/// Check if disk is faulty /// Check if disk is faulty
pub fn is_faulty(&self) -> bool { pub fn is_faulty(&self) -> bool {
self.status.load(Ordering::Acquire) == DISK_HEALTH_FAULTY self.status.load(Ordering::Acquire) == DISK_HEALTH_FAULTY
@@ -463,6 +497,14 @@ impl LocalDiskWrapper {
self.health.offline_duration().map(|duration| duration.as_secs()) self.health.offline_duration().map(|duration| duration.as_secs())
} }
pub fn last_capacity_snapshot(&self) -> Option<(u64, u64, u64, u64)> {
self.health.last_capacity_snapshot()
}
pub fn record_capacity_probe(&self, total: u64, used: u64, free: u64) {
self.health.record_capacity_probe(total, used, free);
}
#[cfg(test)] #[cfg(test)]
pub fn force_runtime_state_for_test(&self, state: RuntimeDriveHealthState) { pub fn force_runtime_state_for_test(&self, state: RuntimeDriveHealthState) {
self.health.force_runtime_state_for_test(state); self.health.force_runtime_state_for_test(state);
crates/ecstore/src/disk/mod.rs
+14
View File
@@ -427,6 +427,20 @@ impl Disk {
} }
} }
pub fn last_capacity_snapshot(&self) -> Option<(u64, u64, u64, u64)> {
match self {
Disk::Local(local_disk) => local_disk.last_capacity_snapshot(),
Disk::Remote(remote_disk) => remote_disk.last_capacity_snapshot(),
}
}
pub fn record_capacity_probe(&self, total: u64, used: u64, free: u64) {
match self {
Disk::Local(local_disk) => local_disk.record_capacity_probe(total, used, free),
Disk::Remote(remote_disk) => remote_disk.record_capacity_probe(total, used, free),
}
}
#[cfg(test)] #[cfg(test)]
pub fn force_runtime_state_for_test(&self, state: RuntimeDriveHealthState) { pub fn force_runtime_state_for_test(&self, state: RuntimeDriveHealthState) {
match self { match self {
crates/ecstore/src/error.rs
+12
View File
@@ -678,6 +678,18 @@ pub fn is_err_read_quorum(err: &Error) -> bool {
matches!(err, &StorageError::ErasureReadQuorum) matches!(err, &StorageError::ErasureReadQuorum)
} }
pub fn classify_system_path_failure_reason(err: &Error) -> &'static str {
match err {
StorageError::ConfigNotFound => "config_not_found",
StorageError::ErasureReadQuorum | StorageError::InsufficientReadQuorum(_, _) => "read_quorum",
StorageError::Io(io_err) => match io_err.kind() {
std::io::ErrorKind::TimedOut => "timeout",
_ => "io",
},
_ => "other",
}
}
pub fn is_err_invalid_upload_id(err: &Error) -> bool { pub fn is_err_invalid_upload_id(err: &Error) -> bool {
matches!(err, &StorageError::InvalidUploadID(_, _, _)) matches!(err, &StorageError::InvalidUploadID(_, _, _))
} }
crates/ecstore/src/rpc/remote_disk.rs
+8
View File
@@ -135,6 +135,14 @@ impl RemoteDisk {
self.health.offline_duration().map(|duration| duration.as_secs()) self.health.offline_duration().map(|duration| duration.as_secs())
} }
pub fn last_capacity_snapshot(&self) -> Option<(u64, u64, u64, u64)> {
self.health.last_capacity_snapshot()
}
pub fn record_capacity_probe(&self, total: u64, used: u64, free: u64) {
self.health.record_capacity_probe(total, used, free);
}
#[cfg(test)] #[cfg(test)]
pub fn force_runtime_state_for_test(&self, state: RuntimeDriveHealthState) { pub fn force_runtime_state_for_test(&self, state: RuntimeDriveHealthState) {
self.health.force_runtime_state_for_test(state); self.health.force_runtime_state_for_test(state);
crates/ecstore/src/set_disk.rs
+106 -50
View File
@@ -101,7 +101,7 @@ use s3s::header::{X_AMZ_OBJECT_LOCK_LEGAL_HOLD, X_AMZ_OBJECT_LOCK_MODE, X_AMZ_OB
use sha2::{Digest, Sha256}; use sha2::{Digest, Sha256};
use std::hash::Hash; use std::hash::Hash;
use std::mem::{self}; use std::mem::{self};
use std::time::{Instant, SystemTime}; use std::time::{Instant, SystemTime, UNIX_EPOCH};
use std::{ use std::{
collections::{HashMap, HashSet}, collections::{HashMap, HashSet},
io::{Cursor, Write}, io::{Cursor, Write},
@@ -4146,56 +4146,78 @@ async fn get_disks_info(disks: &[Option<DiskStore>], eps: &[Endpoint]) -> Vec<ru
if let Some(disk) = pool { if let Some(disk) = pool {
let runtime_state = disk.runtime_state(); let runtime_state = disk.runtime_state();
let offline_duration_seconds = disk.offline_duration_secs(); let offline_duration_seconds = disk.offline_duration_secs();
if runtime_state.should_probe_for_admin() { let capacity_snapshot = disk.last_capacity_snapshot();
if runtime_state.should_probe_for_admin()
|| runtime_state == crate::disk::health_state::RuntimeDriveHealthState::Suspect
{
match disk.disk_info(&DiskInfoOptions::default()).await { match disk.disk_info(&DiskInfoOptions::default()).await {
Ok(res) => ret.push(rustfs_madmin::Disk { Ok(res) => {
endpoint: eps[i].to_string(), disk.record_capacity_probe(res.total, res.used, res.free);
local: eps[i].is_local, ret.push(rustfs_madmin::Disk {
pool_index: eps[i].pool_idx, endpoint: eps[i].to_string(),
set_index: eps[i].set_idx, local: eps[i].is_local,
disk_index: eps[i].disk_idx, pool_index: eps[i].pool_idx,
state: "ok".to_owned(), set_index: eps[i].set_idx,
disk_index: eps[i].disk_idx,
state: "ok".to_owned(),
root_disk: res.root_disk, root_disk: res.root_disk,
drive_path: res.mount_path.clone(), drive_path: res.mount_path.clone(),
healing: res.healing, healing: res.healing,
scanning: res.scanning, scanning: res.scanning,
runtime_state: Some(runtime_state.as_str().to_string()), runtime_state: Some(runtime_state.as_str().to_string()),
offline_duration_seconds, offline_duration_seconds,
capacity_observation_source: Some("live_probe".to_owned()),
capacity_observation_age_seconds: Some(0),
uuid: res.id.map_or_else(|| "".to_string(), |id| id.to_string()), uuid: res.id.map_or_else(|| "".to_string(), |id| id.to_string()),
major: res.major as u32, major: res.major as u32,
minor: res.minor as u32, minor: res.minor as u32,
model: None, model: None,
total_space: res.total, total_space: res.total,
used_space: res.used, used_space: res.used,
available_space: res.free, available_space: res.free,
physical_device_ids: (!res.physical_device_ids.is_empty()).then_some(res.physical_device_ids.clone()), physical_device_ids: (!res.physical_device_ids.is_empty()).then_some(res.physical_device_ids.clone()),
utilization: { utilization: utilization_percent(res.total, res.used),
if res.total > 0 { used_inodes: res.used_inodes,
res.used as f64 / res.total as f64 * 100_f64 free_inodes: res.free_inodes,
} else { ..Default::default()
0_f64 });
} }
}, Err(err) => {
used_inodes: res.used_inodes, let mut disk_info = rustfs_madmin::Disk {
free_inodes: res.free_inodes, state: err.to_string(),
..Default::default() endpoint: eps[i].to_string(),
}), local: eps[i].is_local,
Err(err) => ret.push(rustfs_madmin::Disk { pool_index: eps[i].pool_idx,
state: err.to_string(), set_index: eps[i].set_idx,
endpoint: eps[i].to_string(), disk_index: eps[i].disk_idx,
local: eps[i].is_local, runtime_state: Some(runtime_state.as_str().to_string()),
pool_index: eps[i].pool_idx, offline_duration_seconds,
set_index: eps[i].set_idx, ..Default::default()
disk_index: eps[i].disk_idx, };
runtime_state: Some(runtime_state.as_str().to_string()), if let Some((total, used, free, _)) = capacity_snapshot {
offline_duration_seconds, disk_info.total_space = total;
..Default::default() disk_info.used_space = used;
}), disk_info.available_space = free;
disk_info.utilization = utilization_percent(total, used);
disk_info.capacity_observation_source = Some("snapshot".to_owned());
disk_info.capacity_observation_age_seconds = capacity_snapshot
.map(|(_, _, _, probe_unix_secs)| capacity_snapshot_age_seconds(probe_unix_secs));
} else {
disk_info.capacity_observation_source = Some("missing".to_owned());
disk_info.capacity_observation_age_seconds = Some(0);
}
ret.push(disk_info);
}
} }
} else { } else {
ret.push(build_runtime_snapshot_disk(&eps[i], runtime_state, offline_duration_seconds)); ret.push(build_runtime_snapshot_disk(
&eps[i],
runtime_state,
offline_duration_seconds,
capacity_snapshot,
));
} }
} else { } else {
ret.push(rustfs_madmin::Disk { ret.push(rustfs_madmin::Disk {
@@ -4207,6 +4229,8 @@ async fn get_disks_info(disks: &[Option<DiskStore>], eps: &[Endpoint]) -> Vec<ru
runtime_state: None, runtime_state: None,
offline_duration_seconds: None, offline_duration_seconds: None,
state: DiskError::DiskNotFound.to_string(), state: DiskError::DiskNotFound.to_string(),
capacity_observation_source: Some("missing".to_owned()),
capacity_observation_age_seconds: Some(0),
..Default::default() ..Default::default()
}) })
} }
@@ -4219,8 +4243,9 @@ fn build_runtime_snapshot_disk(
endpoint: &Endpoint, endpoint: &Endpoint,
runtime_state: crate::disk::health_state::RuntimeDriveHealthState, runtime_state: crate::disk::health_state::RuntimeDriveHealthState,
offline_duration_seconds: Option<u64>, offline_duration_seconds: Option<u64>,
capacity_snapshot: Option<(u64, u64, u64, u64)>,
) -> rustfs_madmin::Disk { ) -> rustfs_madmin::Disk {
rustfs_madmin::Disk { let mut disk = rustfs_madmin::Disk {
endpoint: endpoint.to_string(), endpoint: endpoint.to_string(),
local: endpoint.is_local, local: endpoint.is_local,
pool_index: endpoint.pool_idx, pool_index: endpoint.pool_idx,
@@ -4230,7 +4255,38 @@ fn build_runtime_snapshot_disk(
runtime_state: Some(runtime_state.as_str().to_string()), runtime_state: Some(runtime_state.as_str().to_string()),
offline_duration_seconds, offline_duration_seconds,
..Default::default() ..Default::default()
};
if let Some((total, used, free, _)) = capacity_snapshot {
disk.total_space = total;
disk.used_space = used;
disk.available_space = free;
disk.utilization = utilization_percent(total, used);
disk.capacity_observation_source = Some("snapshot".to_owned());
disk.capacity_observation_age_seconds =
capacity_snapshot.map(|(_, _, _, probe_unix_secs)| capacity_snapshot_age_seconds(probe_unix_secs));
} else {
disk.capacity_observation_source = Some("missing".to_owned());
disk.capacity_observation_age_seconds = Some(0);
} }
disk
}
fn utilization_percent(total: u64, used: u64) -> f64 {
if total > 0 {
used as f64 / total as f64 * 100_f64
} else {
0_f64
}
}
fn capacity_snapshot_age_seconds(probe_unix_secs: u64) -> u64 {
let now_unix_secs = SystemTime::now()
.duration_since(UNIX_EPOCH)
.map(|dur| dur.as_secs())
.unwrap_or(probe_unix_secs);
now_unix_secs.saturating_sub(probe_unix_secs)
} }
async fn get_storage_info(disks: &[Option<DiskStore>], eps: &[Endpoint]) -> rustfs_madmin::StorageInfo { async fn get_storage_info(disks: &[Option<DiskStore>], eps: &[Endpoint]) -> rustfs_madmin::StorageInfo {
// let mut disks = get_disks_info(disks, eps).await; // let mut disks = get_disks_info(disks, eps).await;
@@ -5313,7 +5369,7 @@ mod tests {
} }
#[tokio::test] #[tokio::test]
async fn test_get_disks_info_uses_runtime_snapshot_for_suspect_and_offline_disks() { async fn test_get_disks_info_preserves_runtime_state_for_suspect_and_offline_disks() {
let format = FormatV3::new(1, 3); let format = FormatV3::new(1, 3);
let mut temp_dirs = Vec::new(); let mut temp_dirs = Vec::new();
let mut endpoints = Vec::new(); let mut endpoints = Vec::new();
@@ -5342,9 +5398,9 @@ mod tests {
assert_eq!(info[0].runtime_state.as_deref(), Some("online")); assert_eq!(info[0].runtime_state.as_deref(), Some("online"));
assert!(!info[0].drive_path.is_empty(), "online disk should keep immediate disk_info probe"); assert!(!info[0].drive_path.is_empty(), "online disk should keep immediate disk_info probe");
assert_eq!(info[1].state, "suspect"); assert_eq!(info[1].state, "ok");
assert_eq!(info[1].runtime_state.as_deref(), Some("suspect")); assert_eq!(info[1].runtime_state.as_deref(), Some("suspect"));
assert!(info[1].drive_path.is_empty(), "suspect disk should use runtime snapshot fallback"); assert!(!info[1].drive_path.is_empty(), "suspect disk should still probe for fresher disk info");
assert_eq!(info[2].state, "offline"); assert_eq!(info[2].state, "offline");
assert_eq!(info[2].runtime_state.as_deref(), Some("offline")); assert_eq!(info[2].runtime_state.as_deref(), Some("offline"));
crates/ecstore/src/set_disk/lock.rs
+131 -75
View File
@@ -107,101 +107,111 @@ impl SetDisks {
DriveMembershipSnapshot::from_optional_disks(&disks) DriveMembershipSnapshot::from_optional_disks(&disks)
} }
fn reprobe_runtime_candidates_once(&self, disks: &[DiskStore]) {
for disk in disks {
if disk.runtime_state() != crate::disk::health_state::RuntimeDriveHealthState::Online {
disk.reset_health_for_store_init_retry();
}
}
}
pub async fn get_online_disks_with_healing_and_info(&self, incl_healing: bool) -> (Vec<DiskStore>, Vec<DiskInfo>, usize) { pub async fn get_online_disks_with_healing_and_info(&self, incl_healing: bool) -> (Vec<DiskStore>, Vec<DiskInfo>, usize) {
let snapshot = self.drive_membership_snapshot().await; let snapshot = self.drive_membership_snapshot().await;
let mut disks = snapshot.scanner_heal_candidates().into_iter().map(Some).collect::<Vec<_>>(); let mut membership_candidates = snapshot.scanner_heal_candidates();
let mut reprobed = false;
let mut infos: Vec<Option<DiskInfo>> = vec![None; disks.len()]; loop {
let mut disks = membership_candidates.clone();
let mut infos: Vec<Option<DiskInfo>> = vec![None; disks.len()];
let mut futures = Vec::with_capacity(disks.len()); let mut futures = Vec::with_capacity(disks.len());
{ {
let mut rng = rand::rng(); let mut rng = rand::rng();
disks.shuffle(&mut rng); disks.shuffle(&mut rng);
} }
for (i, disk) in disks.iter().cloned().enumerate() { for (i, disk) in disks.iter().cloned().enumerate() {
futures.push(async move { futures.push(async move {
let info = if let Some(disk) = disk { let info = match disk.disk_info(&DiskInfoOptions::default()).await {
match disk.disk_info(&DiskInfoOptions::default()).await {
Ok(info) => info, Ok(info) => info,
Err(err) => DiskInfo { Err(err) => DiskInfo {
error: err.to_string(), error: err.to_string(),
..Default::default() ..Default::default()
}, },
};
Ok((i, info))
});
}
let processor = get_global_processors().metadata_processor();
let results = processor.execute_batch(futures).await;
for (submitted_idx, result) in results.into_iter().enumerate() {
match result {
Ok((disk_idx, info)) => {
infos[disk_idx] = Some(info);
} }
} else { Err(err) => {
DiskInfo { infos[submitted_idx] = Some(DiskInfo {
error: DiskError::DiskNotFound.to_string(), error: err.to_string(),
..Default::default() ..Default::default()
});
} }
}
}
let mut healing: usize = 0;
let mut scanning_disks = Vec::new();
let mut healing_disks = Vec::new();
let mut scanning_infos = Vec::new();
let mut healing_infos = Vec::new();
let mut new_disks = Vec::new();
let mut new_infos = Vec::new();
for (disk, info) in disks.into_iter().zip(infos) {
let Some(info) = info else {
continue;
}; };
Ok((i, info)) if !info.error.is_empty() {
}); continue;
}
// Use optimized batch processor for disk info retrieval
let processor = get_global_processors().metadata_processor();
let results = processor.execute_batch(futures).await;
for (submitted_idx, result) in results.into_iter().enumerate() {
match result {
Ok((disk_idx, info)) => {
infos[disk_idx] = Some(info);
}
Err(err) => {
infos[submitted_idx] = Some(DiskInfo {
error: err.to_string(),
..Default::default()
});
}
}
}
let mut healing: usize = 0;
let mut scanning_disks = Vec::new();
let mut healing_disks = Vec::new();
let mut scanning_infos = Vec::new();
let mut healing_infos = Vec::new();
let mut new_disks = Vec::new();
let mut new_infos = Vec::new();
for (disk, info) in disks.into_iter().zip(infos) {
let Some(info) = info else {
continue;
};
if !info.error.is_empty() || disk.is_none() {
continue;
}
if info.healing {
healing += 1;
if incl_healing {
healing_disks.push(disk.unwrap());
healing_infos.push(info);
} }
continue; if info.healing {
healing += 1;
if incl_healing {
healing_disks.push(disk);
healing_infos.push(info);
}
continue;
}
if !info.scanning {
new_disks.push(disk);
new_infos.push(info);
} else {
scanning_disks.push(disk);
scanning_infos.push(info);
}
} }
if !info.healing { new_disks.extend(scanning_disks);
new_disks.push(disk.unwrap()); new_infos.extend(scanning_infos);
new_infos.push(info); new_disks.extend(healing_disks);
} else { new_infos.extend(healing_infos);
scanning_disks.push(disk.unwrap());
scanning_infos.push(info); if !new_disks.is_empty() || membership_candidates.is_empty() || reprobed {
return (new_disks, new_infos, healing);
} }
reprobed = true;
self.reprobe_runtime_candidates_once(&membership_candidates);
membership_candidates = self.drive_membership_snapshot().await.scanner_heal_candidates();
} }
new_disks.extend(scanning_disks);
new_infos.extend(scanning_infos);
new_disks.extend(healing_disks);
new_infos.extend(healing_infos);
(new_disks, new_infos, healing)
} }
pub(super) async fn _get_local_disks(&self) -> Vec<Option<DiskStore>> { pub(super) async fn _get_local_disks(&self) -> Vec<Option<DiskStore>> {
@@ -527,4 +537,50 @@ mod tests {
drop(temp_dirs); drop(temp_dirs);
} }
#[tokio::test]
async fn get_online_disks_with_healing_and_info_reprobes_runtime_candidates_once() {
let disk_count = 4;
let format = FormatV3::new(1, disk_count);
let mut temp_dirs = Vec::with_capacity(disk_count);
let mut endpoints = Vec::with_capacity(disk_count);
let mut disks = Vec::with_capacity(disk_count);
for disk_idx in 0..disk_count {
let (temp_dir, endpoint, disk) = make_formatted_local_disk(disk_idx, &format).await;
temp_dirs.push(temp_dir);
endpoints.push(endpoint);
disks.push(Some(disk));
}
let set_disks = SetDisks::new(
"test-owner".to_string(),
Arc::new(RwLock::new(disks)),
disk_count,
disk_count / 2,
0,
0,
endpoints,
format,
Vec::new(),
)
.await;
let all_disks = set_disks.get_disks_internal().await;
for disk in all_disks.iter().flatten() {
disk.force_runtime_state_for_test(crate::disk::health_state::RuntimeDriveHealthState::Returning);
}
let (online_disks, infos, healing) = set_disks.get_online_disks_with_healing_and_info(false).await;
assert_eq!(healing, 0);
assert_eq!(online_disks.len(), disk_count);
assert_eq!(infos.len(), disk_count);
assert!(
infos.iter().all(|info| info.error.is_empty()),
"runtime reprobe should recover a usable candidate set without probe errors"
);
drop(temp_dirs);
}
} }
crates/ecstore/src/store_list_objects.rs
+16 -1
View File
@@ -40,7 +40,7 @@ use std::sync::Arc;
use tokio::sync::broadcast::{self}; use tokio::sync::broadcast::{self};
use tokio::sync::mpsc::{self, Receiver, Sender}; use tokio::sync::mpsc::{self, Receiver, Sender};
use tokio_util::sync::CancellationToken; use tokio_util::sync::CancellationToken;
use tracing::{error, info}; use tracing::{error, info, warn};
use uuid::Uuid; use uuid::Uuid;
const MAX_OBJECT_LIST: i32 = 1000; const MAX_OBJECT_LIST: i32 = 1000;
@@ -50,6 +50,19 @@ const MAX_OBJECT_LIST: i32 = 1000;
const METACACHE_SHARE_PREFIX: bool = false; const METACACHE_SHARE_PREFIX: bool = false;
fn ensure_non_empty_listing_disks(bucket: &str, path: &str, disks: &[DiskStore]) -> Result<()> {
if disks.is_empty() {
warn!(
bucket = %bucket,
path = %path,
"listing candidate disks collapsed to empty set"
);
return Err(StorageError::ErasureReadQuorum);
}
Ok(())
}
pub fn max_keys_plus_one(max_keys: i32, add_one: bool) -> i32 { pub fn max_keys_plus_one(max_keys: i32, add_one: bool) -> i32 {
let mut max_keys = max_keys; let mut max_keys = max_keys;
if !(0..=MAX_OBJECT_LIST).contains(&max_keys) { if !(0..=MAX_OBJECT_LIST).contains(&max_keys) {
@@ -761,6 +774,7 @@ impl ECStore {
}; };
let path = base_dir_from_prefix(prefix); let path = base_dir_from_prefix(prefix);
ensure_non_empty_listing_disks(bucket, &path, &disks)?;
let mut filter_prefix = { let mut filter_prefix = {
prefix prefix
@@ -1253,6 +1267,7 @@ impl SetDisks {
} }
let listing_quorum = ((ask_disks + 1) / 2) as usize; let listing_quorum = ((ask_disks + 1) / 2) as usize;
ensure_non_empty_listing_disks(&opts.bucket, &opts.base_dir, &disks)?;
let mut fallback_disks = Vec::new(); let mut fallback_disks = Vec::new();
crates/iam/Cargo.toml
+1
View File
@@ -48,6 +48,7 @@ jsonwebtoken = { workspace = true }
tracing.workspace = true tracing.workspace = true
rustfs-madmin.workspace = true rustfs-madmin.workspace = true
rustfs-utils = { workspace = true, features = ["path"] } rustfs-utils = { workspace = true, features = ["path"] }
rustfs-io-metrics.workspace = true
tokio-util.workspace = true tokio-util.workspace = true
pollster.workspace = true pollster.workspace = true
reqwest = { workspace = true } reqwest = { workspace = true }
crates/iam/src/store/object.rs
+17 -4
View File
@@ -22,7 +22,7 @@ use crate::{
}; };
use futures::future::join_all; use futures::future::join_all;
use rustfs_credentials::get_global_action_cred; use rustfs_credentials::get_global_action_cred;
use rustfs_ecstore::error::StorageError; use rustfs_ecstore::error::{StorageError, classify_system_path_failure_reason};
use rustfs_ecstore::store_api::{ListOperations as _, ObjectInfoOrErr, WalkOptions}; use rustfs_ecstore::store_api::{ListOperations as _, ObjectInfoOrErr, WalkOptions};
use rustfs_ecstore::{ use rustfs_ecstore::{
config::{ config::{
@@ -32,6 +32,7 @@ use rustfs_ecstore::{
store::ECStore, store::ECStore,
store_api::{HTTPPreconditions, ObjectInfo, ObjectOptions}, store_api::{HTTPPreconditions, ObjectInfo, ObjectOptions},
}; };
use rustfs_io_metrics::record_system_path_failure;
use rustfs_policy::{auth::UserIdentity, policy::PolicyDoc}; use rustfs_policy::{auth::UserIdentity, policy::PolicyDoc};
use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf}; use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf};
use serde::{Serialize, de::DeserializeOwned}; use serde::{Serialize, de::DeserializeOwned};
@@ -366,11 +367,16 @@ impl ObjectStore {
.walk(ctx.clone(), Self::BUCKET_NAME, &path, tx, WalkOptions::default()) .walk(ctx.clone(), Self::BUCKET_NAME, &path, tx, WalkOptions::default())
.await .await
{ {
let reason = classify_system_path_failure_reason(&err);
record_system_path_failure("iam_config", "walk", reason);
error!( error!(
path_kind = "iam_config",
operation = "walk",
reason,
bucket = Self::BUCKET_NAME, bucket = Self::BUCKET_NAME,
prefix = %path, prefix = %path,
error = ?err, error = %err,
"list_iam_config_items walk task failed" "system path walk failed"
); );
} }
}); });
@@ -418,7 +424,14 @@ impl ObjectStore {
while let Some(v) = rx.recv().await { while let Some(v) = rx.recv().await {
if let Some(err) = v.err { if let Some(err) = v.err {
warn!("list_iam_config_items {:?}", err); warn!(
path_kind = "iam_config",
operation = "list_items",
reason = "walk_result",
error = %err,
"system path list failed"
);
record_system_path_failure("iam_config", "list_items", "walk_result");
ctx.cancel(); ctx.cancel();
return Err(err); return Err(err);
crates/io-metrics/src/lib.rs
+2
View File
@@ -65,6 +65,7 @@ pub mod lock_metrics;
pub mod performance; pub mod performance;
pub mod process_lock_metrics; pub mod process_lock_metrics;
pub mod sampler; pub mod sampler;
pub mod system_path_metrics;
pub mod timeout_metrics; pub mod timeout_metrics;
pub use autotuner::{AutoTuner, TunerConfig, TuningResult}; pub use autotuner::{AutoTuner, TunerConfig, TuningResult};
@@ -121,6 +122,7 @@ pub use sampler::{
ProcessResourceSnapshot, ProcessStatusSnapshot, ProcessSystemSnapshot, snapshot_process_platform, snapshot_process_resource, ProcessResourceSnapshot, ProcessStatusSnapshot, ProcessSystemSnapshot, snapshot_process_platform, snapshot_process_resource,
snapshot_process_resource_and_system, snapshot_process_system, snapshot_process_resource_and_system, snapshot_process_system,
}; };
pub use system_path_metrics::record_system_path_failure;
// Timeout metrics exports // Timeout metrics exports
pub use timeout_metrics::{ pub use timeout_metrics::{
crates/io-metrics/src/system_path_metrics.rs
+23
View File
@@ -0,0 +1,23 @@
// 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.
pub fn record_system_path_failure(path_kind: &'static str, operation: &'static str, reason: &'static str) {
counter!(
"rustfs_system_path_failures_total",
"path_kind" => path_kind,
"operation" => operation,
"reason" => reason
)
.increment(1);
}
crates/madmin/src/info_commands.rs
+8
View File
@@ -97,6 +97,10 @@ pub struct Disk {
pub runtime_state: Option<String>, pub runtime_state: Option<String>,
#[serde(rename = "offlineDurationSeconds", default, skip_serializing_if = "Option::is_none")] #[serde(rename = "offlineDurationSeconds", default, skip_serializing_if = "Option::is_none")]
pub offline_duration_seconds: Option<u64>, pub offline_duration_seconds: Option<u64>,
#[serde(rename = "capacityObservationSource", default, skip_serializing_if = "Option::is_none")]
pub capacity_observation_source: Option<String>,
#[serde(rename = "capacityObservationAgeSeconds", default, skip_serializing_if = "Option::is_none")]
pub capacity_observation_age_seconds: Option<u64>,
/// Leaf physical block devices backing this disk path when the platform can resolve them. /// Leaf physical block devices backing this disk path when the platform can resolve them.
#[serde(rename = "physicalDeviceIds", default, skip_serializing_if = "Option::is_none")] #[serde(rename = "physicalDeviceIds", default, skip_serializing_if = "Option::is_none")]
pub physical_device_ids: Option<Vec<String>>, pub physical_device_ids: Option<Vec<String>>,
@@ -515,6 +519,8 @@ mod tests {
metrics: Some(DiskMetrics::default()), metrics: Some(DiskMetrics::default()),
runtime_state: Some("online".to_string()), runtime_state: Some("online".to_string()),
offline_duration_seconds: Some(0), offline_duration_seconds: Some(0),
capacity_observation_source: None,
capacity_observation_age_seconds: None,
heal_info: None, heal_info: None,
used_inodes: 1000000, used_inodes: 1000000,
free_inodes: 9000000, free_inodes: 9000000,
@@ -616,6 +622,8 @@ mod tests {
physical_device_ids: Some(vec!["nvme0n1".to_string(), "nvme1n1".to_string()]), physical_device_ids: Some(vec!["nvme0n1".to_string(), "nvme1n1".to_string()]),
runtime_state: Some("online".to_string()), runtime_state: Some("online".to_string()),
offline_duration_seconds: Some(0), offline_duration_seconds: Some(0),
capacity_observation_source: None,
capacity_observation_age_seconds: None,
}; };
let mut encoded = Vec::new(); let mut encoded = Vec::new();
crates/obs/src/metrics/collectors/cluster.rs
+12 -2
View File
@@ -38,6 +38,10 @@ pub struct ClusterStats {
pub used_bytes: u64, pub used_bytes: u64,
/// Available free storage in bytes /// Available free storage in bytes
pub free_bytes: u64, pub free_bytes: u64,
/// Number of drives backed by stale capacity snapshots
pub stale_capacity_drives: u64,
/// Number of drives with no capacity observation
pub missing_capacity_drives: u64,
/// Total number of objects in the cluster /// Total number of objects in the cluster
pub objects_count: u64, pub objects_count: u64,
/// Total number of buckets in the cluster /// Total number of buckets in the cluster
@@ -54,6 +58,8 @@ pub fn collect_cluster_metrics(stats: &ClusterStats) -> Vec<PrometheusMetric> {
PrometheusMetric::from_descriptor(&CLUSTER_CAPACITY_USABLE_TOTAL_BYTES_MD, stats.usable_capacity_bytes as f64), PrometheusMetric::from_descriptor(&CLUSTER_CAPACITY_USABLE_TOTAL_BYTES_MD, stats.usable_capacity_bytes as f64),
PrometheusMetric::from_descriptor(&CLUSTER_CAPACITY_USED_BYTES_MD, stats.used_bytes as f64), PrometheusMetric::from_descriptor(&CLUSTER_CAPACITY_USED_BYTES_MD, stats.used_bytes as f64),
PrometheusMetric::from_descriptor(&CLUSTER_CAPACITY_FREE_BYTES_MD, stats.free_bytes as f64), PrometheusMetric::from_descriptor(&CLUSTER_CAPACITY_FREE_BYTES_MD, stats.free_bytes as f64),
PrometheusMetric::from_descriptor(&CLUSTER_CAPACITY_STALE_DRIVES_MD, stats.stale_capacity_drives as f64),
PrometheusMetric::from_descriptor(&CLUSTER_CAPACITY_MISSING_DRIVES_MD, stats.missing_capacity_drives as f64),
PrometheusMetric::from_descriptor(&CLUSTER_OBJECTS_TOTAL_MD, stats.objects_count as f64), PrometheusMetric::from_descriptor(&CLUSTER_OBJECTS_TOTAL_MD, stats.objects_count as f64),
PrometheusMetric::from_descriptor(&CLUSTER_BUCKETS_TOTAL_MD, stats.buckets_count as f64), PrometheusMetric::from_descriptor(&CLUSTER_BUCKETS_TOTAL_MD, stats.buckets_count as f64),
] ]
@@ -71,6 +77,8 @@ mod tests {
usable_capacity_bytes: 2500, usable_capacity_bytes: 2500,
used_bytes: 1200, used_bytes: 1200,
free_bytes: 1300, free_bytes: 1300,
stale_capacity_drives: 1,
missing_capacity_drives: 0,
objects_count: 100, objects_count: 100,
buckets_count: 5, buckets_count: 5,
}; };
@@ -78,7 +86,7 @@ mod tests {
let metrics = collect_cluster_metrics(&stats); let metrics = collect_cluster_metrics(&stats);
report_metrics(&metrics); report_metrics(&metrics);
assert_eq!(metrics.len(), 6); assert_eq!(metrics.len(), 8);
// Verify raw capacity // Verify raw capacity
let raw_capacity_name = CLUSTER_CAPACITY_RAW_TOTAL_BYTES_MD.get_full_metric_name(); let raw_capacity_name = CLUSTER_CAPACITY_RAW_TOTAL_BYTES_MD.get_full_metric_name();
@@ -108,7 +116,7 @@ mod tests {
let metrics = collect_cluster_metrics(&stats); let metrics = collect_cluster_metrics(&stats);
report_metrics(&metrics); report_metrics(&metrics);
assert_eq!(metrics.len(), 6); assert_eq!(metrics.len(), 8);
// All values should be zero // All values should be zero
for metric in &metrics { for metric in &metrics {
@@ -124,6 +132,8 @@ mod tests {
assert_eq!(stats.usable_capacity_bytes, 0); assert_eq!(stats.usable_capacity_bytes, 0);
assert_eq!(stats.used_bytes, 0); assert_eq!(stats.used_bytes, 0);
assert_eq!(stats.free_bytes, 0); assert_eq!(stats.free_bytes, 0);
assert_eq!(stats.stale_capacity_drives, 0);
assert_eq!(stats.missing_capacity_drives, 0);
assert_eq!(stats.objects_count, 0); assert_eq!(stats.objects_count, 0);
assert_eq!(stats.buckets_count, 0); assert_eq!(stats.buckets_count, 0);
} }
crates/obs/src/metrics/collectors/system_drive.rs
+26 -2
View File
@@ -40,6 +40,10 @@ pub struct DriveDetailedStats {
pub used_bytes: u64, pub used_bytes: u64,
/// Free capacity in bytes /// Free capacity in bytes
pub free_bytes: u64, pub free_bytes: u64,
/// Capacity observation state: live, stale, or missing
pub capacity_observation_state: &'static str,
/// Age in seconds of the current capacity observation
pub capacity_observation_age_seconds: u64,
/// Used inodes /// Used inodes
pub used_inodes: u64, pub used_inodes: u64,
/// Free inodes /// Free inodes
@@ -103,7 +107,7 @@ pub fn collect_drive_detailed_metrics(stats: &[DriveDetailedStats]) -> Vec<Prome
); );
} }
let mut metrics = Vec::with_capacity(stats.len() * 19); let mut metrics = Vec::with_capacity(stats.len() * 23);
for stat in stats { for stat in stats {
let server_label = stat.server.as_str(); let server_label = stat.server.as_str();
@@ -112,6 +116,24 @@ pub fn collect_drive_detailed_metrics(stats: &[DriveDetailedStats]) -> Vec<Prome
push_drive_metric(&mut metrics, &DRIVE_TOTAL_BYTES_MD, stat.total_bytes as f64, server_label, drive_label); push_drive_metric(&mut metrics, &DRIVE_TOTAL_BYTES_MD, stat.total_bytes as f64, server_label, drive_label);
push_drive_metric(&mut metrics, &DRIVE_USED_BYTES_MD, stat.used_bytes as f64, server_label, drive_label); push_drive_metric(&mut metrics, &DRIVE_USED_BYTES_MD, stat.used_bytes as f64, server_label, drive_label);
push_drive_metric(&mut metrics, &DRIVE_FREE_BYTES_MD, stat.free_bytes as f64, server_label, drive_label); push_drive_metric(&mut metrics, &DRIVE_FREE_BYTES_MD, stat.free_bytes as f64, server_label, drive_label);
push_drive_metric(
&mut metrics,
&DRIVE_CAPACITY_OBSERVATION_AGE_SECONDS_MD,
stat.capacity_observation_age_seconds as f64,
server_label,
drive_label,
);
for state in ["live", "stale", "missing"] {
metrics.push(
PrometheusMetric::from_descriptor(
&DRIVE_CAPACITY_OBSERVATION_STATE_MD,
if state == stat.capacity_observation_state { 1.0 } else { 0.0 },
)
.with_label_owned(DRIVE_LABEL, drive_label.to_string())
.with_label_owned(SERVER_LABEL, server_label.to_string())
.with_label_owned("state", state.to_string()),
);
}
push_drive_metric(&mut metrics, &DRIVE_USED_INODES_MD, stat.used_inodes as f64, server_label, drive_label); push_drive_metric(&mut metrics, &DRIVE_USED_INODES_MD, stat.used_inodes as f64, server_label, drive_label);
push_drive_metric(&mut metrics, &DRIVE_FREE_INODES_MD, stat.free_inodes as f64, server_label, drive_label); push_drive_metric(&mut metrics, &DRIVE_FREE_INODES_MD, stat.free_inodes as f64, server_label, drive_label);
push_drive_metric(&mut metrics, &DRIVE_TOTAL_INODES_MD, stat.total_inodes as f64, server_label, drive_label); push_drive_metric(&mut metrics, &DRIVE_TOTAL_INODES_MD, stat.total_inodes as f64, server_label, drive_label);
@@ -219,6 +241,8 @@ mod tests {
total_bytes: 1024 * 1024 * 1024 * 100, // 100 GB total_bytes: 1024 * 1024 * 1024 * 100, // 100 GB
used_bytes: 1024 * 1024 * 1024 * 50, // 50 GB used_bytes: 1024 * 1024 * 1024 * 50, // 50 GB
free_bytes: 1024 * 1024 * 1024 * 50, // 50 GB free_bytes: 1024 * 1024 * 1024 * 50, // 50 GB
capacity_observation_state: "live",
capacity_observation_age_seconds: 0,
used_inodes: 100000, used_inodes: 100000,
free_inodes: 900000, free_inodes: 900000,
total_inodes: 1000000, total_inodes: 1000000,
@@ -240,7 +264,7 @@ mod tests {
let metrics = collect_drive_detailed_metrics(&stats); let metrics = collect_drive_detailed_metrics(&stats);
report_metrics(&metrics); report_metrics(&metrics);
assert_eq!(metrics.len(), 19); assert_eq!(metrics.len(), 23);
// Verify total bytes metric // Verify total bytes metric
let total_bytes_name = DRIVE_TOTAL_BYTES_MD.get_full_metric_name(); let total_bytes_name = DRIVE_TOTAL_BYTES_MD.get_full_metric_name();
crates/obs/src/metrics/schema/cluster.rs
+20
View File
@@ -57,6 +57,26 @@ pub static CLUSTER_CAPACITY_FREE_BYTES_MD: LazyLock<MetricDescriptor> = LazyLock
) )
}); });
/// Number of drives whose capacity is served from a stale snapshot.
pub static CLUSTER_CAPACITY_STALE_DRIVES_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
MetricName::Custom("capacity_stale_drives".to_string()),
"Count of drives whose capacity metrics are served from stale snapshots",
&[],
subsystems::CLUSTER_BASE_PATH,
)
});
/// Number of drives with no capacity observation available.
pub static CLUSTER_CAPACITY_MISSING_DRIVES_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
MetricName::Custom("capacity_missing_drives".to_string()),
"Count of drives with missing capacity observations",
&[],
subsystems::CLUSTER_BASE_PATH,
)
});
/// Total number of objects in the cluster /// Total number of objects in the cluster
pub static CLUSTER_OBJECTS_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| { pub static CLUSTER_OBJECTS_TOTAL_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md( new_gauge_md(
crates/obs/src/metrics/schema/system_drive.rs
+18
View File
@@ -60,6 +60,24 @@ pub static DRIVE_TOTAL_BYTES_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
) )
}); });
pub static DRIVE_CAPACITY_OBSERVATION_STATE_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
MetricName::Custom("capacity_observation_state".to_string()),
"Drive capacity observation state (1 for the active state label, 0 otherwise). States: live, stale, missing",
&[&ALL_DRIVE_LABELS[..], &["state"]].concat(),
subsystems::SYSTEM_DRIVE,
)
});
pub static DRIVE_CAPACITY_OBSERVATION_AGE_SECONDS_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md(
MetricName::Custom("capacity_observation_age_seconds".to_string()),
"Age in seconds of the drive capacity observation currently exported",
&ALL_DRIVE_LABELS[..],
subsystems::SYSTEM_DRIVE,
)
});
pub static DRIVE_USED_INODES_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| { pub static DRIVE_USED_INODES_MD: LazyLock<MetricDescriptor> = LazyLock::new(|| {
new_gauge_md( new_gauge_md(
MetricName::DriveUsedInodes, MetricName::DriveUsedInodes,
crates/obs/src/metrics/stats_collector.rs
+36
View File
@@ -47,6 +47,9 @@ const DRIVE_STATE_OK: &str = "ok";
const DRIVE_STATE_ONLINE: &str = "online"; const DRIVE_STATE_ONLINE: &str = "online";
const DRIVE_STATE_UNFORMATTED: &str = "unformatted"; const DRIVE_STATE_UNFORMATTED: &str = "unformatted";
const DRIVE_RUNTIME_STATE_RETURNING: &str = "returning"; const DRIVE_RUNTIME_STATE_RETURNING: &str = "returning";
const CAPACITY_OBSERVATION_LIVE: &str = "live";
const CAPACITY_OBSERVATION_STALE: &str = "stale";
const CAPACITY_OBSERVATION_MISSING: &str = "missing";
#[derive(Debug, Clone, Copy, PartialEq, Eq)] #[derive(Debug, Clone, Copy, PartialEq, Eq)]
struct ErasureSetQuorumShape { struct ErasureSetQuorumShape {
@@ -71,6 +74,15 @@ fn disk_is_online_for_metrics(state: &str, runtime_state: Option<&str>) -> bool
state_is_acceptable state_is_acceptable
} }
fn disk_capacity_observation_state(source: Option<&str>, age_seconds: Option<u64>) -> (&'static str, u64) {
let age_seconds = age_seconds.unwrap_or(0);
match source {
Some("live_probe") => (CAPACITY_OBSERVATION_LIVE, age_seconds),
Some("snapshot") => (CAPACITY_OBSERVATION_STALE, age_seconds),
_ => (CAPACITY_OBSERVATION_MISSING, age_seconds),
}
}
fn derive_erasure_set_quorum_shape(set_drive_count: usize, parity: usize) -> ErasureSetQuorumShape { fn derive_erasure_set_quorum_shape(set_drive_count: usize, parity: usize) -> ErasureSetQuorumShape {
let data_shards = set_drive_count.saturating_sub(parity); let data_shards = set_drive_count.saturating_sub(parity);
let read_quorum = data_shards.max(1); let read_quorum = data_shards.max(1);
@@ -114,6 +126,22 @@ pub async fn collect_cluster_and_health_stats() -> (ClusterStats, ClusterHealthS
let used: u64 = storage_info.disks.iter().map(|d| d.used_space).sum(); let used: u64 = storage_info.disks.iter().map(|d| d.used_space).sum();
let usable_capacity = get_total_usable_capacity(&storage_info.disks, &storage_info) as u64; let usable_capacity = get_total_usable_capacity(&storage_info.disks, &storage_info) as u64;
let free = get_total_usable_capacity_free(&storage_info.disks, &storage_info) as u64; let free = get_total_usable_capacity_free(&storage_info.disks, &storage_info) as u64;
let stale_capacity_drives = storage_info
.disks
.iter()
.filter(|disk| {
disk_capacity_observation_state(disk.capacity_observation_source.as_deref(), disk.capacity_observation_age_seconds).0
== CAPACITY_OBSERVATION_STALE
})
.count() as u64;
let missing_capacity_drives = storage_info
.disks
.iter()
.filter(|disk| {
disk_capacity_observation_state(disk.capacity_observation_source.as_deref(), disk.capacity_observation_age_seconds).0
== CAPACITY_OBSERVATION_MISSING
})
.count() as u64;
// Get bucket and object counts from data usage info. // Get bucket and object counts from data usage info.
let (buckets_count, objects_count) = match load_data_usage_from_backend(store.clone()).await { let (buckets_count, objects_count) = match load_data_usage_from_backend(store.clone()).await {
@@ -151,6 +179,8 @@ pub async fn collect_cluster_and_health_stats() -> (ClusterStats, ClusterHealthS
usable_capacity_bytes: usable_capacity, usable_capacity_bytes: usable_capacity,
used_bytes: used, used_bytes: used,
free_bytes: free, free_bytes: free,
stale_capacity_drives,
missing_capacity_drives,
objects_count, objects_count,
buckets_count, buckets_count,
}, },
@@ -483,6 +513,10 @@ pub async fn collect_disk_and_system_drive_stats() -> (Vec<DiskStats>, Vec<Drive
.iter() .iter()
.map(|disk| { .map(|disk| {
let is_online = disk_is_online_for_metrics(disk.state.as_str(), disk.runtime_state.as_deref()); 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(
disk.capacity_observation_source.as_deref(),
disk.capacity_observation_age_seconds,
);
if is_online { if is_online {
online_count += 1; online_count += 1;
} else { } else {
@@ -495,6 +529,8 @@ pub async fn collect_disk_and_system_drive_stats() -> (Vec<DiskStats>, Vec<Drive
total_bytes: disk.total_space, total_bytes: disk.total_space,
used_bytes: disk.used_space, used_bytes: disk.used_space,
free_bytes: disk.available_space, free_bytes: disk.available_space,
capacity_observation_state,
capacity_observation_age_seconds,
used_inodes: 0, used_inodes: 0,
free_inodes: 0, free_inodes: 0,
total_inodes: 0, total_inodes: 0,