diff --git a/.docker/observability/grafana/dashboards/rustfs.json b/.docker/observability/grafana/dashboards/rustfs.json index b12460dd7..315de3c69 100644 --- a/.docker/observability/grafana/dashboards/rustfs.json +++ b/.docker/observability/grafana/dashboards/rustfs.json @@ -461,6 +461,90 @@ "title": "Capacity", "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": { "type": "prometheus", @@ -1927,6 +2011,299 @@ "title": "System Drive Usage", "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": { "type": "prometheus", diff --git a/Cargo.lock b/Cargo.lock index 5716e3e33..69d1be16e 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -8535,6 +8535,7 @@ dependencies = [ "rustfs-credentials", "rustfs-crypto", "rustfs-ecstore", + "rustfs-io-metrics", "rustfs-madmin", "rustfs-policy", "rustfs-utils", diff --git a/crates/ecstore/src/cache_value/metacache_set.rs b/crates/ecstore/src/cache_value/metacache_set.rs index e5ef60389..c65a67275 100644 --- a/crates/ecstore/src/cache_value/metacache_set.rs +++ b/crates/ecstore/src/cache_value/metacache_set.rs @@ -85,7 +85,7 @@ impl Clone for ListPathRawOptions { pub async fn list_path_raw(rx: CancellationToken, opts: ListPathRawOptions) -> disk::error::Result<()> { if opts.disks.is_empty() { - return Err(DiskError::other("list_path_raw: 0 drives provided")); + return Err(DiskError::ErasureReadQuorum); } let mut jobs: Vec>> = Vec::new(); @@ -415,6 +415,15 @@ mod tests { use super::*; 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] async fn peek_with_timeout_times_out_on_silent_reader() { let (_writer, reader) = tokio::io::duplex(64); diff --git a/crates/ecstore/src/data_usage.rs b/crates/ecstore/src/data_usage.rs index 652aed489..defb7b20e 100644 --- a/crates/ecstore/src/data_usage.rs +++ b/crates/ecstore/src/data_usage.rs @@ -15,7 +15,11 @@ pub mod local_snapshot; 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, }; pub use local_snapshot::{ @@ -26,6 +30,7 @@ pub use local_snapshot::{ use rustfs_common::data_usage::{ BucketTargetUsageInfo, BucketUsageInfo, DataUsageCache, DataUsageEntry, DataUsageInfo, DiskUsageStatus, SizeSummary, }; +use rustfs_io_metrics::record_system_path_failure; use rustfs_utils::path::SLASH_SEPARATOR; use std::{ collections::{HashMap, HashSet, hash_map::Entry}, @@ -109,7 +114,16 @@ pub async fn load_data_usage_from_backend(store: Arc) -> Result = match read_config(store.clone(), &DATA_USAGE_OBJ_NAME_PATH).await { Ok(data) => data, 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 { Ok(data) => data, @@ -117,7 +131,16 @@ pub async fn load_data_usage_from_backend(store: Arc) -> Result 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 pub fn is_faulty(&self) -> bool { self.status.load(Ordering::Acquire) == DISK_HEALTH_FAULTY @@ -463,6 +497,14 @@ impl LocalDiskWrapper { 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)] pub fn force_runtime_state_for_test(&self, state: RuntimeDriveHealthState) { self.health.force_runtime_state_for_test(state); diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index 6c2e05b5e..9602aca63 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/crates/ecstore/src/disk/mod.rs @@ -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)] pub fn force_runtime_state_for_test(&self, state: RuntimeDriveHealthState) { match self { diff --git a/crates/ecstore/src/error.rs b/crates/ecstore/src/error.rs index cedc28348..504f33570 100644 --- a/crates/ecstore/src/error.rs +++ b/crates/ecstore/src/error.rs @@ -678,6 +678,18 @@ pub fn is_err_read_quorum(err: &Error) -> bool { 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 { matches!(err, &StorageError::InvalidUploadID(_, _, _)) } diff --git a/crates/ecstore/src/rpc/remote_disk.rs b/crates/ecstore/src/rpc/remote_disk.rs index f38b13907..c67a2c392 100644 --- a/crates/ecstore/src/rpc/remote_disk.rs +++ b/crates/ecstore/src/rpc/remote_disk.rs @@ -135,6 +135,14 @@ impl RemoteDisk { 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)] pub fn force_runtime_state_for_test(&self, state: RuntimeDriveHealthState) { self.health.force_runtime_state_for_test(state); diff --git a/crates/ecstore/src/set_disk.rs b/crates/ecstore/src/set_disk.rs index 334b3d88e..7f1a973c0 100644 --- a/crates/ecstore/src/set_disk.rs +++ b/crates/ecstore/src/set_disk.rs @@ -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 std::hash::Hash; use std::mem::{self}; -use std::time::{Instant, SystemTime}; +use std::time::{Instant, SystemTime, UNIX_EPOCH}; use std::{ collections::{HashMap, HashSet}, io::{Cursor, Write}, @@ -4146,56 +4146,78 @@ async fn get_disks_info(disks: &[Option], eps: &[Endpoint]) -> Vec ret.push(rustfs_madmin::Disk { - endpoint: eps[i].to_string(), - local: eps[i].is_local, - pool_index: eps[i].pool_idx, - set_index: eps[i].set_idx, - disk_index: eps[i].disk_idx, - state: "ok".to_owned(), + Ok(res) => { + disk.record_capacity_probe(res.total, res.used, res.free); + ret.push(rustfs_madmin::Disk { + endpoint: eps[i].to_string(), + local: eps[i].is_local, + pool_index: eps[i].pool_idx, + set_index: eps[i].set_idx, + disk_index: eps[i].disk_idx, + state: "ok".to_owned(), - root_disk: res.root_disk, - drive_path: res.mount_path.clone(), - healing: res.healing, - scanning: res.scanning, - runtime_state: Some(runtime_state.as_str().to_string()), - offline_duration_seconds, + root_disk: res.root_disk, + drive_path: res.mount_path.clone(), + healing: res.healing, + scanning: res.scanning, + runtime_state: Some(runtime_state.as_str().to_string()), + 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()), - major: res.major as u32, - minor: res.minor as u32, - model: None, - total_space: res.total, - used_space: res.used, - available_space: res.free, - physical_device_ids: (!res.physical_device_ids.is_empty()).then_some(res.physical_device_ids.clone()), - utilization: { - if res.total > 0 { - res.used as f64 / res.total as f64 * 100_f64 - } else { - 0_f64 - } - }, - used_inodes: res.used_inodes, - free_inodes: res.free_inodes, - ..Default::default() - }), - Err(err) => ret.push(rustfs_madmin::Disk { - state: err.to_string(), - endpoint: eps[i].to_string(), - local: eps[i].is_local, - pool_index: eps[i].pool_idx, - set_index: eps[i].set_idx, - disk_index: eps[i].disk_idx, - runtime_state: Some(runtime_state.as_str().to_string()), - offline_duration_seconds, - ..Default::default() - }), + uuid: res.id.map_or_else(|| "".to_string(), |id| id.to_string()), + major: res.major as u32, + minor: res.minor as u32, + model: None, + total_space: res.total, + used_space: res.used, + available_space: res.free, + physical_device_ids: (!res.physical_device_ids.is_empty()).then_some(res.physical_device_ids.clone()), + utilization: utilization_percent(res.total, res.used), + used_inodes: res.used_inodes, + free_inodes: res.free_inodes, + ..Default::default() + }); + } + Err(err) => { + let mut disk_info = rustfs_madmin::Disk { + state: err.to_string(), + endpoint: eps[i].to_string(), + local: eps[i].is_local, + pool_index: eps[i].pool_idx, + set_index: eps[i].set_idx, + disk_index: eps[i].disk_idx, + runtime_state: Some(runtime_state.as_str().to_string()), + offline_duration_seconds, + ..Default::default() + }; + if let Some((total, used, free, _)) = capacity_snapshot { + disk_info.total_space = total; + 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 { - 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 { ret.push(rustfs_madmin::Disk { @@ -4207,6 +4229,8 @@ async fn get_disks_info(disks: &[Option], eps: &[Endpoint]) -> Vec, + capacity_snapshot: Option<(u64, u64, u64, u64)>, ) -> rustfs_madmin::Disk { - rustfs_madmin::Disk { + let mut disk = rustfs_madmin::Disk { endpoint: endpoint.to_string(), local: endpoint.is_local, pool_index: endpoint.pool_idx, @@ -4230,7 +4255,38 @@ fn build_runtime_snapshot_disk( runtime_state: Some(runtime_state.as_str().to_string()), offline_duration_seconds, ..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], eps: &[Endpoint]) -> rustfs_madmin::StorageInfo { // let mut disks = get_disks_info(disks, eps).await; @@ -5313,7 +5369,7 @@ mod tests { } #[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 mut temp_dirs = Vec::new(); let mut endpoints = Vec::new(); @@ -5342,9 +5398,9 @@ mod tests { 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_eq!(info[1].state, "suspect"); + assert_eq!(info[1].state, "ok"); 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].runtime_state.as_deref(), Some("offline")); diff --git a/crates/ecstore/src/set_disk/lock.rs b/crates/ecstore/src/set_disk/lock.rs index 4c8fad46d..4a6cc7d5d 100644 --- a/crates/ecstore/src/set_disk/lock.rs +++ b/crates/ecstore/src/set_disk/lock.rs @@ -107,101 +107,111 @@ impl SetDisks { 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, Vec, usize) { let snapshot = self.drive_membership_snapshot().await; - let mut disks = snapshot.scanner_heal_candidates().into_iter().map(Some).collect::>(); + let mut membership_candidates = snapshot.scanner_heal_candidates(); + let mut reprobed = false; - let mut infos: Vec> = vec![None; disks.len()]; + loop { + let mut disks = membership_candidates.clone(); + let mut infos: Vec> = vec![None; disks.len()]; - let mut futures = Vec::with_capacity(disks.len()); - { - let mut rng = rand::rng(); - disks.shuffle(&mut rng); - } + let mut futures = Vec::with_capacity(disks.len()); + { + let mut rng = rand::rng(); + disks.shuffle(&mut rng); + } - for (i, disk) in disks.iter().cloned().enumerate() { - futures.push(async move { - let info = if let Some(disk) = disk { - match disk.disk_info(&DiskInfoOptions::default()).await { + for (i, disk) in disks.iter().cloned().enumerate() { + futures.push(async move { + let info = match disk.disk_info(&DiskInfoOptions::default()).await { Ok(info) => info, Err(err) => DiskInfo { error: err.to_string(), ..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 { - DiskInfo { - error: DiskError::DiskNotFound.to_string(), - ..Default::default() + 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; }; - Ok((i, info)) - }); - } - - // 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); + if !info.error.is_empty() { + continue; } - 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.push(disk.unwrap()); - new_infos.push(info); - } else { - scanning_disks.push(disk.unwrap()); - scanning_infos.push(info); + new_disks.extend(scanning_disks); + new_infos.extend(scanning_infos); + new_disks.extend(healing_disks); + new_infos.extend(healing_infos); + + 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> { @@ -527,4 +537,50 @@ mod tests { 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); + } } diff --git a/crates/ecstore/src/store_list_objects.rs b/crates/ecstore/src/store_list_objects.rs index 57a5f5ac4..96a6163d0 100644 --- a/crates/ecstore/src/store_list_objects.rs +++ b/crates/ecstore/src/store_list_objects.rs @@ -40,7 +40,7 @@ use std::sync::Arc; use tokio::sync::broadcast::{self}; use tokio::sync::mpsc::{self, Receiver, Sender}; use tokio_util::sync::CancellationToken; -use tracing::{error, info}; +use tracing::{error, info, warn}; use uuid::Uuid; const MAX_OBJECT_LIST: i32 = 1000; @@ -50,6 +50,19 @@ const MAX_OBJECT_LIST: i32 = 1000; 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 { let mut max_keys = max_keys; if !(0..=MAX_OBJECT_LIST).contains(&max_keys) { @@ -761,6 +774,7 @@ impl ECStore { }; let path = base_dir_from_prefix(prefix); + ensure_non_empty_listing_disks(bucket, &path, &disks)?; let mut filter_prefix = { prefix @@ -1253,6 +1267,7 @@ impl SetDisks { } 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(); diff --git a/crates/iam/Cargo.toml b/crates/iam/Cargo.toml index a13244855..b9c2d7482 100644 --- a/crates/iam/Cargo.toml +++ b/crates/iam/Cargo.toml @@ -48,6 +48,7 @@ jsonwebtoken = { workspace = true } tracing.workspace = true rustfs-madmin.workspace = true rustfs-utils = { workspace = true, features = ["path"] } +rustfs-io-metrics.workspace = true tokio-util.workspace = true pollster.workspace = true reqwest = { workspace = true } diff --git a/crates/iam/src/store/object.rs b/crates/iam/src/store/object.rs index 2cf71203b..b626719a2 100644 --- a/crates/iam/src/store/object.rs +++ b/crates/iam/src/store/object.rs @@ -22,7 +22,7 @@ use crate::{ }; use futures::future::join_all; 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::{ config::{ @@ -32,6 +32,7 @@ use rustfs_ecstore::{ store::ECStore, store_api::{HTTPPreconditions, ObjectInfo, ObjectOptions}, }; +use rustfs_io_metrics::record_system_path_failure; use rustfs_policy::{auth::UserIdentity, policy::PolicyDoc}; use rustfs_utils::path::{SLASH_SEPARATOR, path_join_buf}; use serde::{Serialize, de::DeserializeOwned}; @@ -366,11 +367,16 @@ impl ObjectStore { .walk(ctx.clone(), Self::BUCKET_NAME, &path, tx, WalkOptions::default()) .await { + let reason = classify_system_path_failure_reason(&err); + record_system_path_failure("iam_config", "walk", reason); error!( + path_kind = "iam_config", + operation = "walk", + reason, bucket = Self::BUCKET_NAME, prefix = %path, - error = ?err, - "list_iam_config_items walk task failed" + error = %err, + "system path walk failed" ); } }); @@ -418,7 +424,14 @@ impl ObjectStore { while let Some(v) = rx.recv().await { 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(); return Err(err); diff --git a/crates/io-metrics/src/lib.rs b/crates/io-metrics/src/lib.rs index 20412555a..bf1ac0665 100644 --- a/crates/io-metrics/src/lib.rs +++ b/crates/io-metrics/src/lib.rs @@ -65,6 +65,7 @@ pub mod lock_metrics; pub mod performance; pub mod process_lock_metrics; pub mod sampler; +pub mod system_path_metrics; pub mod timeout_metrics; pub use autotuner::{AutoTuner, TunerConfig, TuningResult}; @@ -121,6 +122,7 @@ pub use sampler::{ ProcessResourceSnapshot, ProcessStatusSnapshot, ProcessSystemSnapshot, snapshot_process_platform, snapshot_process_resource, snapshot_process_resource_and_system, snapshot_process_system, }; +pub use system_path_metrics::record_system_path_failure; // Timeout metrics exports pub use timeout_metrics::{ diff --git a/crates/io-metrics/src/system_path_metrics.rs b/crates/io-metrics/src/system_path_metrics.rs new file mode 100644 index 000000000..06326ca8b --- /dev/null +++ b/crates/io-metrics/src/system_path_metrics.rs @@ -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); +} diff --git a/crates/madmin/src/info_commands.rs b/crates/madmin/src/info_commands.rs index 9bbe5bc06..aad577d6d 100644 --- a/crates/madmin/src/info_commands.rs +++ b/crates/madmin/src/info_commands.rs @@ -97,6 +97,10 @@ pub struct Disk { pub runtime_state: Option, #[serde(rename = "offlineDurationSeconds", default, skip_serializing_if = "Option::is_none")] pub offline_duration_seconds: Option, + #[serde(rename = "capacityObservationSource", default, skip_serializing_if = "Option::is_none")] + pub capacity_observation_source: Option, + #[serde(rename = "capacityObservationAgeSeconds", default, skip_serializing_if = "Option::is_none")] + pub capacity_observation_age_seconds: Option, /// Leaf physical block devices backing this disk path when the platform can resolve them. #[serde(rename = "physicalDeviceIds", default, skip_serializing_if = "Option::is_none")] pub physical_device_ids: Option>, @@ -515,6 +519,8 @@ mod tests { metrics: Some(DiskMetrics::default()), runtime_state: Some("online".to_string()), offline_duration_seconds: Some(0), + capacity_observation_source: None, + capacity_observation_age_seconds: None, heal_info: None, used_inodes: 1000000, free_inodes: 9000000, @@ -616,6 +622,8 @@ mod tests { physical_device_ids: Some(vec!["nvme0n1".to_string(), "nvme1n1".to_string()]), runtime_state: Some("online".to_string()), offline_duration_seconds: Some(0), + capacity_observation_source: None, + capacity_observation_age_seconds: None, }; let mut encoded = Vec::new(); diff --git a/crates/obs/src/metrics/collectors/cluster.rs b/crates/obs/src/metrics/collectors/cluster.rs index 87a961d96..284fdcd68 100644 --- a/crates/obs/src/metrics/collectors/cluster.rs +++ b/crates/obs/src/metrics/collectors/cluster.rs @@ -38,6 +38,10 @@ pub struct ClusterStats { pub used_bytes: u64, /// Available free storage in bytes 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 pub objects_count: u64, /// Total number of buckets in the cluster @@ -54,6 +58,8 @@ pub fn collect_cluster_metrics(stats: &ClusterStats) -> Vec { 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_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_BUCKETS_TOTAL_MD, stats.buckets_count as f64), ] @@ -71,6 +77,8 @@ mod tests { usable_capacity_bytes: 2500, used_bytes: 1200, free_bytes: 1300, + stale_capacity_drives: 1, + missing_capacity_drives: 0, objects_count: 100, buckets_count: 5, }; @@ -78,7 +86,7 @@ mod tests { let metrics = collect_cluster_metrics(&stats); report_metrics(&metrics); - assert_eq!(metrics.len(), 6); + assert_eq!(metrics.len(), 8); // Verify raw capacity 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); report_metrics(&metrics); - assert_eq!(metrics.len(), 6); + assert_eq!(metrics.len(), 8); // All values should be zero for metric in &metrics { @@ -124,6 +132,8 @@ mod tests { assert_eq!(stats.usable_capacity_bytes, 0); assert_eq!(stats.used_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.buckets_count, 0); } diff --git a/crates/obs/src/metrics/collectors/system_drive.rs b/crates/obs/src/metrics/collectors/system_drive.rs index 6d47f4dd1..9829ba7b2 100644 --- a/crates/obs/src/metrics/collectors/system_drive.rs +++ b/crates/obs/src/metrics/collectors/system_drive.rs @@ -40,6 +40,10 @@ pub struct DriveDetailedStats { pub used_bytes: u64, /// Free capacity in bytes 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 pub used_inodes: u64, /// Free inodes @@ -103,7 +107,7 @@ pub fn collect_drive_detailed_metrics(stats: &[DriveDetailedStats]) -> Vec Vec = LazyLock ) }); +/// Number of drives whose capacity is served from a stale snapshot. +pub static CLUSTER_CAPACITY_STALE_DRIVES_MD: LazyLock = 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 = 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 pub static CLUSTER_OBJECTS_TOTAL_MD: LazyLock = LazyLock::new(|| { new_gauge_md( diff --git a/crates/obs/src/metrics/schema/system_drive.rs b/crates/obs/src/metrics/schema/system_drive.rs index ab8e07b2b..2f0ddea33 100644 --- a/crates/obs/src/metrics/schema/system_drive.rs +++ b/crates/obs/src/metrics/schema/system_drive.rs @@ -60,6 +60,24 @@ pub static DRIVE_TOTAL_BYTES_MD: LazyLock = LazyLock::new(|| { ) }); +pub static DRIVE_CAPACITY_OBSERVATION_STATE_MD: LazyLock = 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 = 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 = LazyLock::new(|| { new_gauge_md( MetricName::DriveUsedInodes, diff --git a/crates/obs/src/metrics/stats_collector.rs b/crates/obs/src/metrics/stats_collector.rs index 9dff01d14..7cfa4991f 100644 --- a/crates/obs/src/metrics/stats_collector.rs +++ b/crates/obs/src/metrics/stats_collector.rs @@ -47,6 +47,9 @@ const DRIVE_STATE_OK: &str = "ok"; const DRIVE_STATE_ONLINE: &str = "online"; const DRIVE_STATE_UNFORMATTED: &str = "unformatted"; 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)] struct ErasureSetQuorumShape { @@ -71,6 +74,15 @@ fn disk_is_online_for_metrics(state: &str, runtime_state: Option<&str>) -> bool state_is_acceptable } +fn disk_capacity_observation_state(source: Option<&str>, age_seconds: Option) -> (&'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 { let data_shards = set_drive_count.saturating_sub(parity); 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 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 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. 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, used_bytes: used, free_bytes: free, + stale_capacity_drives, + missing_capacity_drives, objects_count, buckets_count, }, @@ -483,6 +513,10 @@ pub async fn collect_disk_and_system_drive_stats() -> (Vec, Vec (Vec, Vec