diff --git a/crates/io-metrics/src/capacity_metrics.rs b/crates/io-metrics/src/capacity_metrics.rs index b7118a57b..543b5d4d5 100644 --- a/crates/io-metrics/src/capacity_metrics.rs +++ b/crates/io-metrics/src/capacity_metrics.rs @@ -55,6 +55,13 @@ pub fn record_capacity_update_completed(source: &'static str, duration: Duration .increment(1); } +/// Record a committed capacity reading that is degraded: the refresh behind it +/// had partial disk failures, so some disks kept last-known values. +#[inline(always)] +pub fn record_capacity_degraded_reading(source: &'static str) { + counter!("rustfs_capacity_degraded_readings_total", "source" => source).increment(1); +} + /// Record failed capacity update. #[inline(always)] pub fn record_capacity_update_failed(source: &'static str) { diff --git a/crates/object-capacity/src/capacity_manager.rs b/crates/object-capacity/src/capacity_manager.rs index 850e7a05d..c4c8c49b5 100644 --- a/crates/object-capacity/src/capacity_manager.rs +++ b/crates/object-capacity/src/capacity_manager.rs @@ -30,9 +30,9 @@ use rustfs_config::{ ENV_CAPACITY_WRITE_TRIGGER_DELAY, }; use rustfs_io_metrics::capacity_metrics::{ - record_capacity_current_bytes, record_capacity_dirty_disk_count, record_capacity_refresh_inflight, - record_capacity_refresh_joiner, record_capacity_refresh_result, record_capacity_update_completed, - record_capacity_update_failed, record_capacity_write_operation, + record_capacity_current_bytes, record_capacity_degraded_reading, record_capacity_dirty_disk_count, + record_capacity_refresh_inflight, record_capacity_refresh_joiner, record_capacity_refresh_result, + record_capacity_update_completed, record_capacity_update_failed, record_capacity_write_operation, }; use rustfs_utils::{get_env_bool, get_env_u64, get_env_usize}; use std::collections::{HashMap, HashSet}; @@ -323,6 +323,9 @@ pub struct CachedCapacity { pub file_count: usize, /// Whether it's an estimated value pub is_estimated: bool, + /// Whether the value comes from a refresh that had partial disk failures + /// (some disks kept their last-known values or were dropped entirely). + pub degraded: bool, /// Data source pub source: DataSource, } @@ -336,6 +339,10 @@ pub struct CapacityUpdate { pub file_count: usize, /// Whether the value is estimated instead of exact. pub is_estimated: bool, + /// Whether the refresh behind this update had partial disk failures. + /// `is_estimated` only reflects sampling; this flag is the only carrier of + /// the "some disks failed to scan" fact (backlog#1014). + pub degraded: bool, /// Per-disk breakdown captured from a successful refresh. pub per_disk: Vec, /// Expected disk count for a complete disk cache. @@ -353,6 +360,7 @@ impl CapacityUpdate { total_used, file_count, is_estimated: false, + degraded: false, per_disk: Vec::new(), expected_disk_count: None, replaces_disk_cache: false, @@ -366,6 +374,7 @@ impl CapacityUpdate { total_used, file_count, is_estimated: true, + degraded: false, per_disk: Vec::new(), expected_disk_count: None, replaces_disk_cache: false, @@ -379,6 +388,7 @@ impl CapacityUpdate { total_used, file_count: 0, is_estimated: true, + degraded: false, per_disk: Vec::new(), expected_disk_count: None, replaces_disk_cache: false, @@ -671,11 +681,14 @@ impl HybridCapacityManager { pub async fn update_capacity(&self, mut update: CapacityUpdate, source: DataSource) -> CapacityUpdate { let start = Instant::now(); - if !update.per_disk.is_empty() { + if !update.per_disk.is_empty() || update.degraded { let mut disk_cache = self.disk_cache.write().await; let mut disk_cache_complete = self.disk_cache_complete.write().await; - let recompute = if update.replaces_disk_cache && update.expected_disk_count == Some(update.per_disk.len()) { + let recompute = if !update.per_disk.is_empty() + && update.replaces_disk_cache + && update.expected_disk_count == Some(update.per_disk.len()) + { disk_cache.clear(); for entry in &update.per_disk { disk_cache.insert( @@ -690,6 +703,11 @@ impl HybridCapacityManager { *disk_cache_complete = true; true } else if *disk_cache_complete { + // Merging over a complete cache also covers a degraded + // (partial-failure) full refresh: successfully scanned disks + // are refreshed while failed disks keep their last-known + // values, so the published total does not dip to the + // surviving-subset sum and bounce back on the next refresh. for entry in &update.per_disk { disk_cache.insert( entry.disk.clone(), @@ -722,6 +740,7 @@ impl HybridCapacityManager { last_update: Instant::now(), file_count: update.file_count, is_estimated: update.is_estimated, + degraded: update.degraded, source, }); @@ -741,12 +760,16 @@ impl HybridCapacityManager { total_used, file_count = update.file_count, estimated = update.is_estimated, + degraded = update.degraded, source = source.as_metric_label(), elapsed_ms = start.elapsed().as_millis() as u64, "capacity refresh cache updated" ); record_capacity_current_bytes(total_used); record_capacity_update_completed(source.as_metric_label(), start.elapsed(), total_used, update.is_estimated); + if update.degraded { + record_capacity_degraded_reading(source.as_metric_label()); + } update } @@ -1675,6 +1698,7 @@ mod tests { total_used: 300, file_count: 3, is_estimated: false, + degraded: false, per_disk: vec![ DiskCapacityUpdate { disk: CapacityScopeDisk { @@ -1709,6 +1733,7 @@ mod tests { total_used: 150, file_count: 1, is_estimated: true, + degraded: false, per_disk: vec![DiskCapacityUpdate { disk: CapacityScopeDisk { endpoint: "node-a".to_string(), @@ -1739,6 +1764,138 @@ mod tests { assert!(cached.is_estimated); } + fn disk_entry(endpoint: &str, drive_path: &str, used_bytes: u64, file_count: usize) -> DiskCapacityUpdate { + DiskCapacityUpdate { + disk: CapacityScopeDisk { + endpoint: endpoint.to_string(), + drive_path: drive_path.to_string(), + }, + used_bytes, + file_count, + is_estimated: false, + } + } + + fn full_two_disk_update() -> CapacityUpdate { + CapacityUpdate { + total_used: 200, + file_count: 2, + is_estimated: false, + degraded: false, + per_disk: vec![ + disk_entry("node-a", "/tmp/disk-a", 100, 1), + disk_entry("node-b", "/tmp/disk-b", 100, 1), + ], + expected_disk_count: Some(2), + replaces_disk_cache: true, + clear_dirty_disks: Vec::new(), + } + } + + #[tokio::test] + #[serial] + async fn test_update_capacity_degraded_full_refresh_merges_cache_and_does_not_oscillate() { + let manager = create_isolated_manager(HybridStrategyConfig::default()); + + // Round 1: healthy full refresh seeds a complete cache (A=100, B=100). + manager.update_capacity(full_two_disk_update(), DataSource::RealTime).await; + + // Round 2: disk B fails mid-scan; the degraded update carries only the + // surviving subset's totals and per-disk entries. + let degraded = manager + .update_capacity( + CapacityUpdate { + total_used: 100, + file_count: 1, + is_estimated: false, + degraded: true, + per_disk: vec![disk_entry("node-a", "/tmp/disk-a", 100, 1)], + expected_disk_count: None, + replaces_disk_cache: false, + clear_dirty_disks: Vec::new(), + }, + DataSource::Scheduled, + ) + .await; + + // Disk B keeps its last-known 100 bytes: the published total must not + // dip to the surviving-subset sum. + assert_eq!(degraded.total_used, 200); + assert!(degraded.degraded); + let cached = manager.get_capacity().await.unwrap(); + assert_eq!(cached.total_used, 200); + assert!(cached.degraded); + + // Round 3: disk B recovers; the healthy full refresh clears the flag + // and the reported total never oscillated (200 → 200 → 200). + let recovered = manager.update_capacity(full_two_disk_update(), DataSource::Scheduled).await; + assert_eq!(recovered.total_used, 200); + assert!(!recovered.degraded); + let cached = manager.get_capacity().await.unwrap(); + assert_eq!(cached.total_used, 200); + assert!(!cached.degraded); + } + + #[tokio::test] + #[serial] + async fn test_update_capacity_degraded_with_empty_per_disk_serves_merged_cache() { + let manager = create_isolated_manager(HybridStrategyConfig::default()); + manager.update_capacity(full_two_disk_update(), DataSource::RealTime).await; + + // Every surviving disk had intra-disk errors, so no per-disk entry is + // trustworthy; the complete cache must still back the published total. + let degraded = manager + .update_capacity( + CapacityUpdate { + total_used: 40, + file_count: 1, + is_estimated: false, + degraded: true, + per_disk: Vec::new(), + expected_disk_count: None, + replaces_disk_cache: false, + clear_dirty_disks: Vec::new(), + }, + DataSource::Scheduled, + ) + .await; + + assert_eq!(degraded.total_used, 200); + assert!(degraded.degraded); + } + + #[tokio::test] + #[serial] + async fn test_update_capacity_degraded_without_complete_cache_keeps_partial_sum() { + let manager = create_isolated_manager(HybridStrategyConfig::default()); + + // No complete cache exists: nothing to merge, the partial sum is the + // best available value, but it must be visibly marked degraded and the + // partial per-disk data must not mark the cache complete (#805). + let degraded = manager + .update_capacity( + CapacityUpdate { + total_used: 100, + file_count: 1, + is_estimated: false, + degraded: true, + per_disk: vec![disk_entry("node-a", "/tmp/disk-a", 100, 1)], + expected_disk_count: None, + replaces_disk_cache: false, + clear_dirty_disks: Vec::new(), + }, + DataSource::Scheduled, + ) + .await; + + assert_eq!(degraded.total_used, 100); + assert!(degraded.degraded); + let cached = manager.get_capacity().await.unwrap(); + assert_eq!(cached.total_used, 100); + assert!(cached.degraded); + assert!(!manager.can_refresh_dirty_subset().await); + } + #[tokio::test] #[serial] async fn test_refresh_or_join_returns_cluster_total_for_dirty_subset() { @@ -1751,6 +1908,7 @@ mod tests { total_used: 300, file_count: 3, is_estimated: false, + degraded: false, per_disk: vec![ DiskCapacityUpdate { disk: CapacityScopeDisk { @@ -1784,6 +1942,7 @@ mod tests { total_used: 150, file_count: 1, is_estimated: true, + degraded: false, per_disk: vec![DiskCapacityUpdate { disk: CapacityScopeDisk { endpoint: "node-a".to_string(), diff --git a/crates/object-capacity/src/scan.rs b/crates/object-capacity/src/scan.rs index 0e4dd7146..75ad85f11 100644 --- a/crates/object-capacity/src/scan.rs +++ b/crates/object-capacity/src/scan.rs @@ -95,6 +95,26 @@ impl CapacityScanReport { update.expected_disk_count = Some(expected_disk_count); update.replaces_disk_cache = replaces_disk_cache; update.clear_dirty_disks = update.per_disk.iter().map(|entry| entry.disk.clone()).collect(); + } else if self.summary.had_partial_errors { + // Partial failure: mark the reading degraded and surface only the + // disks whose own scan fully succeeded, so the manager can merge + // them over a complete disk cache while failed disks keep their + // last-known values. A disk with intra-disk errors under-counts, + // so its stale cache entry is closer to the truth than its scan. + // The cache is never replaced from a degraded refresh. + update.degraded = true; + update.per_disk = self + .per_disk + .into_iter() + .filter(|entry| !entry.scan.had_partial_errors) + .map(|entry| DiskCapacityUpdate { + disk: entry.disk, + used_bytes: entry.scan.used_bytes, + file_count: entry.scan.file_count, + is_estimated: entry.scan.is_estimated, + }) + .collect(); + update.clear_dirty_disks = update.per_disk.iter().map(|entry| entry.disk.clone()).collect(); } update @@ -1122,6 +1142,88 @@ mod tests { assert!(result.had_partial_errors); } + #[tokio::test] + async fn test_partial_scan_report_produces_degraded_update() { + use std::fs::File; + use std::io::Write; + use tempfile::TempDir; + + let temp_dir = TempDir::new().unwrap(); + let mut file = File::create(temp_dir.path().join("test.txt")).unwrap(); + file.write_all(b"Hello, World!").unwrap(); + drop(file); + + let good_disk = CapacityDiskRef { + endpoint: "disk-1".to_string(), + drive_path: temp_dir.path().to_string_lossy().into_owned(), + }; + let disks = vec![ + good_disk.clone(), + CapacityDiskRef { + endpoint: "disk-2".to_string(), + drive_path: "/nonexistent/path".to_string(), + }, + ]; + + let report = calculate_data_dir_used_capacity_report(&disks).await.unwrap(); + let update = report.into_capacity_update(disks.len(), true); + + // A partial full refresh is degraded and must never replace the disk + // cache; only the surviving disk is surfaced for cache merging. + assert!(update.degraded); + assert!(!update.replaces_disk_cache); + assert_eq!(update.expected_disk_count, None); + assert_eq!(update.per_disk.len(), 1); + assert_eq!(update.per_disk[0].disk, disk_scope_key(&good_disk)); + assert_eq!(update.clear_dirty_disks, vec![disk_scope_key(&good_disk)]); + } + + #[test] + fn test_into_capacity_update_partial_failure_filters_intra_disk_errors() { + let clean = DiskCapacityScanResult { + disk: CapacityScopeDisk { + endpoint: "node-a".to_string(), + drive_path: "/tmp/disk-a".to_string(), + }, + scan: CapacityScanResult { + used_bytes: 100, + file_count: 1, + ..Default::default() + }, + }; + let partial = DiskCapacityScanResult { + disk: CapacityScopeDisk { + endpoint: "node-b".to_string(), + drive_path: "/tmp/disk-b".to_string(), + }, + scan: CapacityScanResult { + used_bytes: 40, + file_count: 1, + had_partial_errors: true, + ..Default::default() + }, + }; + let report = CapacityScanReport { + summary: CapacityScanResult { + used_bytes: 140, + file_count: 2, + had_partial_errors: true, + ..Default::default() + }, + per_disk: vec![clean.clone(), partial], + }; + + let update = report.into_capacity_update(3, true); + + // A disk whose own scan had intra-disk errors under-counts, so it must + // not overwrite its last-known cached value. + assert!(update.degraded); + assert_eq!(update.per_disk.len(), 1); + assert_eq!(update.per_disk[0].disk, clean.disk); + assert_eq!(update.per_disk[0].used_bytes, 100); + assert!(!update.replaces_disk_cache); + } + #[tokio::test] async fn test_select_capacity_refresh_disks_returns_full_when_disk_cache_incomplete() { let manager = create_isolated_manager(HybridStrategyConfig::default()); @@ -1159,6 +1261,7 @@ mod tests { total_used: 300, file_count: 3, is_estimated: false, + degraded: false, per_disk: vec![ DiskCapacityUpdate { disk: CapacityScopeDisk { diff --git a/rustfs/src/capacity/service.rs b/rustfs/src/capacity/service.rs index 16eda0cbe..f23c9bb96 100644 --- a/rustfs/src/capacity/service.rs +++ b/rustfs/src/capacity/service.rs @@ -127,8 +127,8 @@ pub async fn resolve_admin_used_capacity(disks: &[rustfs_madmin::Disk], fallback if cache_age < fast_update_threshold { record_capacity_cache_served("fresh"); debug!( - "Using cached capacity: {} bytes (age: {:?}, source: {:?}, files={}, estimated={})", - cached.total_used, cache_age, cached.source, cached.file_count, cached.is_estimated + "Using cached capacity: {} bytes (age: {:?}, source: {:?}, files={}, estimated={}, degraded={})", + cached.total_used, cache_age, cached.source, cached.file_count, cached.is_estimated, cached.degraded ); return cached.total_used; } @@ -157,6 +157,7 @@ pub async fn resolve_admin_used_capacity(disks: &[rustfs_madmin::Disk], fallback duration_ms = elapsed.as_millis() as u64, file_count = update.file_count, is_estimated = update.is_estimated, + degraded = update.degraded, "Capacity refresh completed" ); update.total_used @@ -188,6 +189,7 @@ pub async fn resolve_admin_used_capacity(disks: &[rustfs_madmin::Disk], fallback source = ?cached.source, file_count = cached.file_count, is_estimated = cached.is_estimated, + degraded = cached.degraded, needs_update, blocking = should_block, "Served cached capacity" @@ -233,6 +235,7 @@ pub async fn resolve_admin_used_capacity(disks: &[rustfs_madmin::Disk], fallback duration_ms = elapsed.as_millis() as u64, file_count = update.file_count, is_estimated = update.is_estimated, + degraded = update.degraded, "Capacity refresh completed" ); update.total_used