diff --git a/crates/ecstore/src/core/pools_test.rs b/crates/ecstore/src/core/pools_test.rs index 11313568a..dd7fa752d 100644 --- a/crates/ecstore/src/core/pools_test.rs +++ b/crates/ecstore/src/core/pools_test.rs @@ -87,50 +87,57 @@ mod capacity_dedup_tests { } #[test] - fn test_four_disk_erasure_coding() { - // 4-disk erasure coding: 2 data disks + 2 parity disks + fn test_four_node_ec_2_2_reports_stable_usable_capacity() { + const TIB: u64 = 1 << 40; + const DISK_TOTAL: u64 = 2 * TIB; + const DISK_FREE: u64 = TIB / 5; + let disks = vec![ rustfs_madmin::Disk { endpoint: "node1".to_string(), - drive_path: "/mnt/disk1".to_string(), + drive_path: "/media/rustfs-01".to_string(), pool_index: 0, set_index: 0, disk_index: 0, - total_space: 1_000_000_000_000, // 1TB - available_space: 250_000_000_000, + total_space: DISK_TOTAL, + available_space: DISK_FREE, + used_space: DISK_TOTAL - DISK_FREE, state: "ok".to_string(), ..Default::default() }, rustfs_madmin::Disk { - endpoint: "node1".to_string(), - drive_path: "/mnt/disk2".to_string(), + endpoint: "node2".to_string(), + drive_path: "/media/rustfs-01".to_string(), pool_index: 0, set_index: 0, disk_index: 1, - total_space: 1_000_000_000_000, - available_space: 250_000_000_000, + total_space: DISK_TOTAL, + available_space: DISK_FREE, + used_space: DISK_TOTAL - DISK_FREE, state: "ok".to_string(), ..Default::default() }, rustfs_madmin::Disk { - endpoint: "node1".to_string(), - drive_path: "/mnt/disk3".to_string(), + endpoint: "node3".to_string(), + drive_path: "/media/rustfs-01".to_string(), pool_index: 0, set_index: 0, disk_index: 2, - total_space: 1_000_000_000_000, - available_space: 250_000_000_000, + total_space: DISK_TOTAL, + available_space: DISK_FREE, + used_space: DISK_TOTAL - DISK_FREE, state: "ok".to_string(), ..Default::default() }, rustfs_madmin::Disk { - endpoint: "node1".to_string(), - drive_path: "/mnt/disk4".to_string(), + endpoint: "node4".to_string(), + drive_path: "/media/rustfs-01".to_string(), pool_index: 0, set_index: 0, disk_index: 3, - total_space: 1_000_000_000_000, - available_space: 250_000_000_000, + total_space: DISK_TOTAL, + available_space: DISK_FREE, + used_space: DISK_TOTAL - DISK_FREE, state: "ok".to_string(), ..Default::default() }, @@ -146,9 +153,17 @@ mod capacity_dedup_tests { }; let total = get_total_usable_capacity(&disks, &info); + let free = get_total_usable_capacity_free(&disks, &info); + let used = total.saturating_sub(free); + let expected_total = usize::try_from(4 * TIB).expect("4 TiB must fit the supported platform's usize"); + let expected_free = usize::try_from(2 * DISK_FREE).expect("usable free capacity must fit usize"); + let single_node_used = usize::try_from(DISK_TOTAL - DISK_FREE).expect("single-node used capacity must fit usize"); - // Only count data disks (disk_index < 2) - assert_eq!(total, 2_000_000_000_000, "Should count only data disks (2 × 1TB)"); + assert_eq!(total, expected_total, "usable total must count the two data disks"); + assert_eq!(free, expected_free, "usable free must count the two data disks"); + assert_eq!(used, 2 * single_node_used, "usable used must be approximately 3.6 TiB"); + assert_ne!(used, single_node_used, "used must not collapse to one node's approximately 1.8 TiB"); + assert_ne!(used, 4 * single_node_used, "used must not report the approximately 7.2 TiB raw aggregate"); } #[test] diff --git a/crates/ecstore/src/services/notification_sys.rs b/crates/ecstore/src/services/notification_sys.rs index 7560aca9e..70d10fa0a 100644 --- a/crates/ecstore/src/services/notification_sys.rs +++ b/crates/ecstore/src/services/notification_sys.rs @@ -319,8 +319,8 @@ impl NotificationSys { if let Some(client) = client { let host = client.host.to_string(); match timeout(peer_timeout, client.local_storage_info()).await { - Ok(Ok(info)) => { - update_storage_info_cache(cache, &host, &info); + Ok(Ok(mut info)) => { + normalize_and_cache_peer_storage_info(cache, &host, &mut info); Some(info) } Ok(Err(err)) => { @@ -1156,7 +1156,13 @@ fn handle_peer_failure( None } -fn update_storage_info_cache(cache: Option<&Mutex>, host: &str, info: &StorageInfo) { +fn normalize_and_cache_peer_storage_info(cache: Option<&Mutex>, host: &str, info: &mut StorageInfo) { + // `Disk::local` is relative to this aggregator, not to the peer that + // produced the response. + for disk in &mut info.disks { + disk.local = false; + } + let Some(cache) = cache else { return; }; @@ -1760,6 +1766,53 @@ mod tests { assert_eq!(cache.lock().unwrap().storage_failures, 1); } + #[test] + fn normalize_and_cache_peer_storage_info_marks_disks_remote() { + let cache = Mutex::new(PeerAdminCache::new()); + let mut info = StorageInfo { + disks: vec![ + rustfs_madmin::Disk { + endpoint: "http://node2:9000/media/rustfs-01".to_string(), + drive_path: "/media/rustfs-01".to_string(), + local: true, + ..Default::default() + }, + rustfs_madmin::Disk { + endpoint: "http://node3:9000/media/rustfs-01".to_string(), + drive_path: "/media/rustfs-01".to_string(), + local: true, + ..Default::default() + }, + rustfs_madmin::Disk { + endpoint: "http://node4:9000/media/rustfs-01".to_string(), + drive_path: "/media/rustfs-01".to_string(), + local: true, + ..Default::default() + }, + ], + ..Default::default() + }; + + normalize_and_cache_peer_storage_info(Some(&cache), "peer-1", &mut info); + + assert!(info.disks.iter().all(|disk| !disk.local)); + let cached = cache.lock().expect("peer cache must remain available"); + assert!( + cached + .last_storage_info + .as_ref() + .expect("successful peer response must be cached") + .disks + .iter() + .all(|disk| !disk.local) + ); + drop(cached); + + let degraded = handle_peer_failure(Some(&cache), "peer-1", &EndpointServerPools::default()) + .expect("first peer failure must return the cached snapshot"); + assert!(degraded.disks.iter().all(|disk| !disk.local)); + } + #[test] fn handle_peer_failure_returns_offline_after_threshold_exceeded() { let cached_info = StorageInfo { @@ -2059,10 +2112,10 @@ mod tests { panic!("poison server cache mutex"); }); - update_storage_info_cache( + normalize_and_cache_peer_storage_info( Some(&storage_cache), "peer-1", - &StorageInfo { + &mut StorageInfo { disks: vec![rustfs_madmin::Disk { endpoint: "disk-0".to_string(), state: "ok".to_string(), diff --git a/rustfs/src/app/admin_usecase.rs b/rustfs/src/app/admin_usecase.rs index e3901a793..bd32ee91a 100644 --- a/rustfs/src/app/admin_usecase.rs +++ b/rustfs/src/app/admin_usecase.rs @@ -24,7 +24,6 @@ use super::storage_api::admin_usecase::{ECStore, EndpointServerPools}; use crate::app::runtime_sources::{ AppContext, current_app_context, current_endpoints_handle, current_object_store_handle_for_context, }; -use crate::capacity::resolve_admin_used_capacity; use crate::cluster_snapshot::{ ClusterReadOnlySnapshot, ClusterRuntimeStatusSnapshot, cluster_read_only_snapshot_from_endpoint_pools, collect_cluster_read_only_snapshot, @@ -311,8 +310,7 @@ impl DefaultAdminUsecase { info.total_free_capacity = free_u64; } - info.total_used_capacity = - resolve_admin_used_capacity(&storage_info.disks, info.total_capacity.saturating_sub(info.total_free_capacity)).await; + info.total_used_capacity = info.total_capacity.saturating_sub(info.total_free_capacity); debug!( "Capacity statistics: total={:.2} TiB, free={:.2} TiB, used={:.2} TiB", info.total_capacity as f64 / (1024.0_f64.powi(4)), diff --git a/rustfs/src/app/data_usage_snapshot_gating_test.rs b/rustfs/src/app/data_usage_snapshot_gating_test.rs index eb6b27771..6dd1d7de4 100644 --- a/rustfs/src/app/data_usage_snapshot_gating_test.rs +++ b/rustfs/src/app/data_usage_snapshot_gating_test.rs @@ -129,18 +129,45 @@ async fn data_usage_endpoint_serves_snapshot_without_live_listing() { record_bucket_object_write_memory(&overlay_bucket, None, 512).await; let before_endpoint = live_bucket_usage_computations(); - let info = DefaultAdminUsecase::query_data_usage_info_with_store(ecstore.clone()) + let first_info = DefaultAdminUsecase::query_data_usage_info_with_store(ecstore.clone()) .await - .expect("query data usage info"); + .expect("query data usage info first time"); + let second_info = DefaultAdminUsecase::query_data_usage_info_with_store(ecstore.clone()) + .await + .expect("query data usage info second time"); assert_eq!( live_bucket_usage_computations(), before_endpoint, - "revert detector: the data usage endpoint must not run live full-version listings" + "revert detector: repeated data usage requests must not run live full-version listings" + ); + assert_eq!( + first_info.total_used_capacity, + first_info.total_capacity.saturating_sub(first_info.total_free_capacity), + "server used capacity must use the same usable-capacity scope as total and free" + ); + assert!( + first_info.total_capacity > 0 && first_info.total_free_capacity > 0, + "test fixture must expose non-zero capacity: total={}, free={}", + first_info.total_capacity, + first_info.total_free_capacity + ); + assert_eq!( + (first_info.total_capacity, first_info.total_free_capacity, first_info.total_used_capacity,), + ( + second_info.total_capacity, + second_info.total_free_capacity, + second_info.total_used_capacity, + ), + "repeated data usage requests must report stable capacity values" ); // The endpoint must serve the seeded snapshot numbers, not recomputed ones. - assert_eq!(info.last_update, Some(seeded_at), "endpoint must report the snapshot timestamp"); - let seeded_usage = info + assert_eq!(first_info.last_update, Some(seeded_at), "endpoint must report the snapshot timestamp"); + assert_eq!( + second_info.last_update, first_info.last_update, + "repeated requests must use the same snapshot" + ); + let seeded_usage = first_info .buckets_usage .get(&seeded_bucket) .expect("seeded bucket must come from the snapshot"); @@ -148,7 +175,7 @@ async fn data_usage_endpoint_serves_snapshot_without_live_listing() { assert_eq!(seeded_usage.objects_count, SEEDED_BUCKET_OBJECTS); // The in-memory overlay stays applied on top of the snapshot. - let overlay_usage = info + let overlay_usage = first_info .buckets_usage .get(&overlay_bucket) .expect("overlay bucket must come from the memory overlay"); diff --git a/rustfs/src/app/gating_test_env.rs b/rustfs/src/app/gating_test_env.rs index 32587ca09..31390ece5 100644 --- a/rustfs/src/app/gating_test_env.rs +++ b/rustfs/src/app/gating_test_env.rs @@ -78,6 +78,9 @@ pub(crate) async fn shared_gating_ecstore() -> Arc { super::storage_api::test::runtime::init_local_disks(endpoint_pools.clone()) .await .unwrap(); + crate::storage::storage_api::new_global_notification_sys(endpoint_pools.clone()) + .await + .expect("initialize notification system for gating test env"); let server_addr: std::net::SocketAddr = "127.0.0.1:0".parse().unwrap(); let ecstore = ECStore::new(server_addr, endpoint_pools, CancellationToken::new()) diff --git a/rustfs/src/capacity/mod.rs b/rustfs/src/capacity/mod.rs index 71440cb1f..400322f69 100644 --- a/rustfs/src/capacity/mod.rs +++ b/rustfs/src/capacity/mod.rs @@ -54,5 +54,4 @@ pub mod service; pub use service::{ capacity_disk_ref, get_cached_capacity_with_metrics, init_capacity_management_for_local_disks, record_capacity_write, - refresh_or_join_admin_disks, resolve_admin_used_capacity, spawn_refresh_if_needed_admin_disks, }; diff --git a/rustfs/src/capacity/service.rs b/rustfs/src/capacity/service.rs index f23c9bb96..91d21ac7a 100644 --- a/rustfs/src/capacity/service.rs +++ b/rustfs/src/capacity/service.rs @@ -13,14 +13,9 @@ // limitations under the License. use crate::storage_api::capacity::service::{all_local_disk, disk_drive_path, disk_endpoint}; -use rustfs_io_metrics::capacity_metrics::{ - record_capacity_cache_hit, record_capacity_cache_miss, record_capacity_cache_served, record_capacity_refresh_request, - record_capacity_scan_mode, -}; -use rustfs_object_capacity::{CapacityDiskRef, capacity_manager, scan}; -use std::sync::Arc; -use std::time::Instant; -use tracing::{debug, info, warn}; +use rustfs_io_metrics::capacity_metrics::{record_capacity_cache_hit, record_capacity_cache_miss}; +use rustfs_object_capacity::{CapacityDiskRef, capacity_manager}; +use tracing::{info, warn}; const LOG_COMPONENT_CAPACITY: &str = "capacity"; const LOG_SUBSYSTEM_CAPACITY: &str = "capacity"; @@ -32,237 +27,12 @@ pub fn capacity_disk_ref(endpoint: impl Into, drive_path: impl Into Vec { - // Admin callers pass cluster-wide `storage_info` disks. The scan walks - // `drive_path` on the local filesystem, so remote disks must be excluded: - // scanning them either double-counts local bytes (same mount layout on every - // node) or fails with NotFound (per-node layouts), and both poison the - // per-disk cache that the scheduled local-only refresh relies on. - disks - .iter() - .filter(|disk| disk.local) - .map(|disk| capacity_disk_ref(disk.endpoint.clone(), disk.drive_path.clone())) - .collect() -} - -async fn refresh_admin_disks_with_subset_fallback( - capacity_manager: &capacity_manager::HybridCapacityManager, - all_disks: Vec, - allow_dirty_subset: bool, -) -> Result { - let (refresh_disks, dirty_subset) = if allow_dirty_subset { - scan::select_capacity_refresh_disks(capacity_manager, &all_disks).await - } else { - (all_disks.clone(), false) - }; - - match scan::refresh_capacity_with_scope(refresh_disks.clone(), dirty_subset).await { - Ok(update) => Ok(update), - Err(err) if dirty_subset => { - warn!( - component = LOG_COMPONENT_CAPACITY, - subsystem = LOG_SUBSYSTEM_CAPACITY, - event = "capacity_refresh_retry", - scope = "dirty_subset", - fallback_scope = "full_disk", - error = %err, - "Capacity refresh failed and will retry with full-disk scope" - ); - scan::refresh_capacity_with_scope(all_disks, false).await - } - Err(err) => Err(err), - } -} - -pub async fn refresh_or_join_admin_disks( - capacity_manager: Arc, - source: capacity_manager::DataSource, - disks: &[rustfs_madmin::Disk], - allow_dirty_subset: bool, -) -> Result { - let all_disks = capacity_disk_refs(disks); - let refresh_manager = capacity_manager.clone(); - - capacity_manager - .refresh_or_join(source, move || { - let capacity_manager = refresh_manager.clone(); - let all_disks = all_disks.clone(); - async move { - refresh_admin_disks_with_subset_fallback(capacity_manager.as_ref(), all_disks, allow_dirty_subset).await - } - }) - .await -} - -pub async fn spawn_refresh_if_needed_admin_disks( - capacity_manager: Arc, - source: capacity_manager::DataSource, - disks: &[rustfs_madmin::Disk], - allow_dirty_subset: bool, -) -> bool { - let all_disks = capacity_disk_refs(disks); - let refresh_manager = capacity_manager.clone(); - - capacity_manager - .spawn_refresh_if_needed(source, move || async move { - refresh_admin_disks_with_subset_fallback(refresh_manager.as_ref(), all_disks, allow_dirty_subset).await - }) - .await -} - pub async fn record_capacity_write(scope_token: Option) { capacity_manager::get_capacity_manager() .record_write_operation_with_scope_token(scope_token) .await; } -pub async fn resolve_admin_used_capacity(disks: &[rustfs_madmin::Disk], fallback_used_capacity: u64) -> u64 { - let capacity_manager = capacity_manager::get_capacity_manager(); - - if let Some(cached) = capacity_manager.get_capacity().await { - record_capacity_cache_hit(); - let cache_age = cached.last_update.elapsed(); - let fast_update_threshold = capacity_manager.get_config().fast_update_threshold; - - if cache_age < fast_update_threshold { - record_capacity_cache_served("fresh"); - debug!( - "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; - } - - let needs_update = capacity_manager.needs_fast_update().await; - let should_block = capacity_manager.should_block_on_refresh(cache_age); - - if needs_update && should_block { - let start = Instant::now(); - record_capacity_refresh_request("blocking", capacity_manager::DataSource::WriteTriggered.as_metric_label()); - return match refresh_or_join_admin_disks( - capacity_manager.clone(), - capacity_manager::DataSource::WriteTriggered, - disks, - true, - ) - .await - { - Ok(update) => { - let elapsed = start.elapsed(); - debug!( - component = LOG_COMPONENT_CAPACITY, - subsystem = LOG_SUBSYSTEM_CAPACITY, - event = "capacity_refresh_completed", - mode = "foreground", - 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 - } - Err(err) => { - warn!( - component = LOG_COMPONENT_CAPACITY, - subsystem = LOG_SUBSYSTEM_CAPACITY, - event = "capacity_refresh_failed", - mode = "foreground", - fallback = "cached_value", - error = %err, - "Capacity refresh failed" - ); - record_capacity_cache_served("stale"); - cached.total_used - } - }; - } - - record_capacity_cache_served("stale"); - debug!( - component = LOG_COMPONENT_CAPACITY, - subsystem = LOG_SUBSYSTEM_CAPACITY, - event = "capacity_cache_served", - cache_state = "stale", - total_used = cached.total_used, - age_ms = cache_age.as_millis() as u64, - source = ?cached.source, - file_count = cached.file_count, - is_estimated = cached.is_estimated, - degraded = cached.degraded, - needs_update, - blocking = should_block, - "Served cached capacity" - ); - - record_capacity_refresh_request("background", capacity_manager::DataSource::Scheduled.as_metric_label()); - if spawn_refresh_if_needed_admin_disks(capacity_manager.clone(), capacity_manager::DataSource::Scheduled, disks, true) - .await - { - debug!( - component = LOG_COMPONENT_CAPACITY, - subsystem = LOG_SUBSYSTEM_CAPACITY, - event = "capacity_refresh_background", - state = "started", - "Background capacity refresh state changed" - ); - } else { - debug!( - component = LOG_COMPONENT_CAPACITY, - subsystem = LOG_SUBSYSTEM_CAPACITY, - event = "capacity_refresh_background", - state = "skipped", - reason = "already_running", - "Background capacity refresh state changed" - ); - } - - return cached.total_used; - } - - let start = Instant::now(); - record_capacity_cache_miss(); - record_capacity_refresh_request("initial", capacity_manager::DataSource::RealTime.as_metric_label()); - match refresh_or_join_admin_disks(capacity_manager.clone(), capacity_manager::DataSource::RealTime, disks, false).await { - Ok(update) => { - let elapsed = start.elapsed(); - info!( - component = LOG_COMPONENT_CAPACITY, - subsystem = LOG_SUBSYSTEM_CAPACITY, - event = "capacity_refresh_completed", - mode = "initial", - total_used = update.total_used, - 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 - } - Err(err) => { - warn!( - component = LOG_COMPONENT_CAPACITY, - subsystem = LOG_SUBSYSTEM_CAPACITY, - event = "capacity_refresh_failed", - mode = "initial", - fallback = "disk_used_capacity", - error = %err, - "Capacity refresh failed" - ); - record_capacity_cache_served("fallback"); - record_capacity_scan_mode("fallback"); - capacity_manager - .update_capacity( - capacity_manager::CapacityUpdate::fallback(fallback_used_capacity), - capacity_manager::DataSource::Fallback, - ) - .await; - fallback_used_capacity - } - } -} - pub async fn init_capacity_management_for_local_disks() { info!( component = LOG_COMPONENT_CAPACITY, @@ -336,30 +106,3 @@ fn capacity_source_label(source: capacity_manager::DataSource) -> &'static str { capacity_manager::DataSource::Fallback => "fallback", } } - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn capacity_disk_refs_excludes_remote_disks() { - let disks = vec![ - rustfs_madmin::Disk { - endpoint: "http://node1:9000/data/rustfs0".to_string(), - drive_path: "/data/rustfs0".to_string(), - local: true, - ..Default::default() - }, - rustfs_madmin::Disk { - endpoint: "http://node2:9000/data/rustfs0".to_string(), - drive_path: "/data/rustfs0".to_string(), - local: false, - ..Default::default() - }, - ]; - - let refs = capacity_disk_refs(&disks); - assert_eq!(refs.len(), 1); - assert_eq!(refs[0].endpoint, "http://node1:9000/data/rustfs0"); - } -}