diff --git a/crates/ecstore/src/pools.rs b/crates/ecstore/src/pools.rs index 679be7d99..779f47397 100644 --- a/crates/ecstore/src/pools.rs +++ b/crates/ecstore/src/pools.rs @@ -277,6 +277,9 @@ fn first_resumable_decommission_queue_indices(meta: &PoolMeta) -> Vec { let mut indices = Vec::new(); for (idx, pool) in meta.pools.iter().enumerate() { if let Some(decommission) = &pool.decommission { + if !decommission.has_decommission_state() { + continue; + } if decommission.complete { continue; } @@ -398,14 +401,14 @@ fn is_decommission_active(complete: bool, failed: bool, canceled: bool) -> bool pub(crate) fn pool_meta_has_active_decommission(meta: &PoolMeta) -> bool { meta.pools.iter().any(|pool| { - pool.decommission - .as_ref() - .is_some_and(|info| is_decommission_active(info.complete, info.failed, info.canceled)) + pool.decommission.as_ref().is_some_and(|info| { + info.has_decommission_state() && is_decommission_active(info.complete, info.failed, info.canceled) + }) }) } fn is_decommission_suspended(info: &PoolDecommissionInfo) -> bool { - !info.queued + info.has_decommission_state() && !info.queued } fn validate_decommission_terminal_state(complete: bool, failed: bool, canceled: bool) -> Result<()> { @@ -438,6 +441,9 @@ fn decommission_start_pool_state(pool: Option<&PoolStatus>) -> DecommissionStart let Some(info) = pool.decommission.as_ref() else { return DecommissionStartPoolState::Active; }; + if !info.has_decommission_state() { + return DecommissionStartPoolState::Active; + } if info.complete { DecommissionStartPoolState::Decommissioned @@ -1540,7 +1546,7 @@ impl PoolMeta { let (decommission_present, complete, failed, canceled) = pool .decommission .as_ref() - .map(|info| (true, info.complete, info.failed, info.canceled)) + .map(|info| (info.has_decommission_state(), info.complete, info.failed, info.canceled)) .unwrap_or((false, false, false, false)); ensure_decommission_clear_allowed(true, decommission_present, complete, failed, canceled)?; @@ -1806,6 +1812,9 @@ impl PoolMeta { let mut new_pools = Vec::new(); for pool in &self.pools { if let Some(decommission) = &pool.decommission { + if !decommission.has_decommission_state() { + continue; + } if decommission.complete || decommission.failed || decommission.canceled { // Recovery is not required when: // - Decommissioning completed @@ -1878,6 +1887,27 @@ pub struct PoolDecommissionInfo { } impl PoolDecommissionInfo { + pub fn has_decommission_state(&self) -> bool { + self.complete + || self.failed + || self.canceled + || self.queued + || self.start_time.is_some() + || self.start_size > 0 + || !self.queued_buckets.is_empty() + || !self.decommissioned_buckets.is_empty() + || !self.bucket.is_empty() + || !self.prefix.is_empty() + || !self.object.is_empty() + || !self.stage.is_empty() + || self.items_decommissioned > 0 + || self.items_decommission_failed > 0 + || self.bytes_done > 0 + || self.bytes_failed > 0 + || self.terminal_reload_attempt_at.is_some() + || !self.terminal_reload_failures.is_empty() + } + fn counted_items(&self) -> usize { self.items_decommissioned.saturating_add(self.items_decommission_failed) } @@ -2215,7 +2245,11 @@ impl ECStore { let (pool_present, decommission_present, terminal) = if let Some(pool) = lock.pools.get(idx) { if let Some(info) = pool.decommission.as_ref() { already_canceled = info.canceled; - (true, true, should_reject_decommission_cancel_as_terminal(info.complete, info.failed)) + ( + true, + info.has_decommission_state(), + should_reject_decommission_cancel_as_terminal(info.complete, info.failed), + ) } else { (true, false, false) } @@ -2339,6 +2373,7 @@ impl ECStore { let pool_meta = self.pool_meta.read().await; for pool in pool_meta.pools.iter() { if let Some(ref info) = pool.decommission + && info.has_decommission_state() && !info.complete && !info.failed && !info.canceled @@ -5101,7 +5136,10 @@ mod pools_tests { ..Default::default() }; let mut active = previous.clone(); - active.pools[0].decommission = Some(PoolDecommissionInfo::default()); + active.pools[0].decommission = Some(PoolDecommissionInfo { + start_time: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }); assert!(active.is_suspended(0)); assert_eq!( @@ -6033,13 +6071,16 @@ mod pools_tests { } #[test] - fn test_pool_meta_has_active_decommission_counts_active_and_queued_states() { + fn test_pool_meta_has_active_decommission_counts_running_and_queued_states() { let active_meta = PoolMeta { pools: vec![PoolStatus { id: 0, cmd_line: "pool-0".to_string(), last_update: OffsetDateTime::UNIX_EPOCH, - decommission: Some(PoolDecommissionInfo::default()), + decommission: Some(PoolDecommissionInfo { + start_time: Some(OffsetDateTime::UNIX_EPOCH), + ..Default::default() + }), }], ..Default::default() }; @@ -6060,6 +6101,27 @@ mod pools_tests { assert!(pool_meta_has_active_decommission(&queued_meta)); } + #[test] + fn test_pool_meta_has_active_decommission_ignores_capacity_placeholder() { + let meta = PoolMeta { + pools: vec![PoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update: OffsetDateTime::UNIX_EPOCH, + decommission: Some(PoolDecommissionInfo { + total_size: 100, + current_size: 75, + ..Default::default() + }), + }], + ..Default::default() + }; + + assert!(!pool_meta_has_active_decommission(&meta)); + assert!(!meta.is_suspended(0)); + assert_eq!(decommission_start_pool_state(meta.pools.first()), DecommissionStartPoolState::Active); + } + #[test] fn test_pool_meta_has_active_decommission_ignores_terminal_states() { let terminal_meta = PoolMeta { @@ -6158,6 +6220,7 @@ mod pools_tests { cmd_line: "pool-0".to_string(), last_update: OffsetDateTime::UNIX_EPOCH, decommission: Some(PoolDecommissionInfo { + start_time: Some(OffsetDateTime::UNIX_EPOCH), complete: false, failed: false, canceled: false, diff --git a/crates/ecstore/src/rebalance/control.rs b/crates/ecstore/src/rebalance/control.rs index 625e0e5bc..58b9b62bf 100644 --- a/crates/ecstore/src/rebalance/control.rs +++ b/crates/ecstore/src/rebalance/control.rs @@ -39,6 +39,13 @@ pub(super) fn validate_rebalance_disk_stats_coverage(disk_stats: &[DiskStat]) -> Ok(()) } +fn pool_rebalance_status_from_meta(meta: Option<&RebalanceMeta>, pool_index: usize) -> (RebalStatus, bool) { + meta.and_then(|meta| meta.pool_stats.get(pool_index)) + .filter(|pool_stat| pool_stat.participating) + .map(|pool_stat| (pool_stat.info.status, pool_stat.info.stopping)) + .unwrap_or_default() +} + impl ECStore { pub(super) async fn save_rebalance_meta_with_merge( &self, @@ -442,11 +449,7 @@ impl ECStore { pub async fn pool_rebalance_status(&self, pool_index: usize) -> (RebalStatus, bool) { let rebalance_meta = self.rebalance_meta.read().await; - rebalance_meta - .as_ref() - .and_then(|meta| meta.pool_stats.get(pool_index)) - .map(|pool_stat| (pool_stat.info.status, pool_stat.info.stopping)) - .unwrap_or_default() + pool_rebalance_status_from_meta(rebalance_meta.as_ref(), pool_index) } pub async fn current_rebalance_id(&self) -> Option { @@ -530,3 +533,38 @@ impl ECStore { Ok(()) } } + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn pool_rebalance_status_ignores_non_participating_pool_state() { + let meta = RebalanceMeta { + pool_stats: vec![ + RebalanceStats { + participating: false, + info: RebalanceInfo { + status: RebalStatus::Started, + stopping: true, + ..Default::default() + }, + ..Default::default() + }, + RebalanceStats { + participating: true, + info: RebalanceInfo { + status: RebalStatus::Started, + ..Default::default() + }, + ..Default::default() + }, + ], + ..Default::default() + }; + + assert_eq!(pool_rebalance_status_from_meta(Some(&meta), 0), (RebalStatus::None, false)); + assert_eq!(pool_rebalance_status_from_meta(Some(&meta), 1), (RebalStatus::Started, false)); + assert_eq!(pool_rebalance_status_from_meta(Some(&meta), 2), (RebalStatus::None, false)); + } +} diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs index 8867d979a..c12dffd75 100644 --- a/crates/ecstore/src/store/init.rs +++ b/crates/ecstore/src/store/init.rs @@ -60,7 +60,7 @@ fn pool_meta_has_active_decommission(meta: &PoolMeta) -> bool { meta.pools.iter().any(|pool| { pool.decommission .as_ref() - .is_some_and(|info| !info.complete && !info.failed && !info.canceled) + .is_some_and(|info| info.has_decommission_state() && !info.complete && !info.failed && !info.canceled) }) } diff --git a/rustfs/src/app/admin_usecase.rs b/rustfs/src/app/admin_usecase.rs index 0524b9280..4ac352b00 100644 --- a/rustfs/src/app/admin_usecase.rs +++ b/rustfs/src/app/admin_usecase.rs @@ -165,6 +165,7 @@ impl DefaultAdminUsecase { const POOL_STATE_BLOCKED: &'static str = "blocked"; const POOL_STATE_DECOMMISSIONED: &'static str = "decommissioned"; const POOL_STATE_DECOMMISSIONING: &'static str = "decommissioning"; + const POOL_STATE_REBALANCING: &'static str = "rebalancing"; const REBALANCE_STATUS_COMPLETED: &'static str = "completed"; const REBALANCE_STATUS_FAILED: &'static str = "failed"; const REBALANCE_STATUS_NONE: &'static str = "none"; @@ -367,13 +368,16 @@ impl DefaultAdminUsecase { } pub async fn execute_list_decommission_status(&self) -> AdminUsecaseResult { + let Some(store) = self.object_store() else { + return Err(Self::app_error(S3ErrorCode::InternalError, "Not init")); + }; let pool_statuses = self.execute_list_pool_statuses().await?; - Ok(AdminDecommissionStatus { - pools: pool_statuses - .into_iter() - .map(Self::decommission_pool_status_from_status) - .collect(), - }) + let mut pools = Vec::with_capacity(pool_statuses.len()); + for status in pool_statuses { + let rebalance_status = store.pool_rebalance_status(status.id).await; + pools.push(Self::decommission_pool_status_from_status(status, rebalance_status)); + } + Ok(AdminDecommissionStatus { pools }) } pub async fn execute_query_decommission_status( @@ -395,7 +399,8 @@ impl DefaultAdminUsecase { }; let status = store.status(idx).await.map_err(ApiError::from)?; - Ok(Self::decommission_pool_status_from_status(status)) + let rebalance_status = store.pool_rebalance_status(idx).await; + Ok(Self::decommission_pool_status_from_status(status, rebalance_status)) } fn pool_list_item_from_status(status: PoolStatus, rebalance_status: (RebalStatus, bool)) -> AdminPoolListItem { @@ -408,7 +413,10 @@ impl DefaultAdminUsecase { let total_size = decommission.as_ref().map(|info| info.total_size).unwrap_or_default(); let current_size = decommission.as_ref().map(|info| info.current_size).unwrap_or_default(); let used_size = total_size.saturating_sub(current_size); - let pool_state = Self::pool_lifecycle_state(decommission.as_ref()); + let decommission = decommission.filter(PoolDecommissionInfo::has_decommission_state); + let decommission_status = Self::pool_decommission_status(decommission.as_ref()); + let rebalance_status = Self::pool_rebalance_status(rebalance_status); + let pool_state = Self::pool_lifecycle_state(decommission.as_ref(), rebalance_status); AdminPoolStatus { id, @@ -419,17 +427,22 @@ impl DefaultAdminUsecase { used_size, used: Self::used_ratio(total_size, used_size), status: pool_state.to_string(), - decommission_status: Self::pool_decommission_status(decommission.as_ref()).to_string(), - rebalance_status: Self::pool_rebalance_status(rebalance_status).to_string(), + decommission_status: decommission_status.to_string(), + rebalance_status: rebalance_status.to_string(), decommission: decommission.map(Self::admin_decommission_info_from_pool), } } - fn pool_lifecycle_state(decommission: Option<&PoolDecommissionInfo>) -> &'static str { + fn pool_lifecycle_state(decommission: Option<&PoolDecommissionInfo>, rebalance_status: &'static str) -> &'static str { match decommission { Some(info) if info.complete => Self::POOL_STATE_DECOMMISSIONED, Some(info) if info.failed || info.canceled => Self::POOL_STATE_BLOCKED, + Some(info) if !info.has_decommission_state() && Self::rebalance_status_is_active(rebalance_status) => { + Self::POOL_STATE_REBALANCING + } + Some(info) if !info.has_decommission_state() => Self::POOL_STATE_ACTIVE, Some(_) => Self::POOL_STATE_DECOMMISSIONING, + None if Self::rebalance_status_is_active(rebalance_status) => Self::POOL_STATE_REBALANCING, None => Self::POOL_STATE_ACTIVE, } } @@ -441,11 +454,16 @@ impl DefaultAdminUsecase { Some(info) if info.canceled => Self::POOL_STATUS_CANCELED, Some(info) if info.queued => Self::POOL_STATUS_QUEUED, Some(info) if info.start_time.is_some() => Self::POOL_STATUS_RUNNING, + Some(info) if !info.has_decommission_state() => Self::REBALANCE_STATUS_NONE, Some(_) => Self::POOL_STATUS_UNKNOWN, None => Self::REBALANCE_STATUS_NONE, } } + fn rebalance_status_is_active(status: &'static str) -> bool { + matches!(status, Self::REBALANCE_STATUS_STARTED | Self::REBALANCE_STATUS_STOPPING) + } + fn pool_rebalance_status((status, stopping): (RebalStatus, bool)) -> &'static str { if stopping { return Self::REBALANCE_STATUS_STOPPING; @@ -460,15 +478,20 @@ impl DefaultAdminUsecase { } } - fn decommission_pool_status_from_status(status: PoolStatus) -> AdminDecommissionPoolStatus { + fn decommission_pool_status_from_status( + status: PoolStatus, + rebalance_status: (RebalStatus, bool), + ) -> AdminDecommissionPoolStatus { let PoolStatus { id, cmd_line, decommission, .. } = status; - let pool_status = Self::pool_lifecycle_state(decommission.as_ref()).to_string(); + let decommission = decommission.filter(PoolDecommissionInfo::has_decommission_state); let status = Self::pool_decommission_status(decommission.as_ref()).to_string(); + let rebalance_status = Self::pool_rebalance_status(rebalance_status); + let pool_status = Self::pool_lifecycle_state(decommission.as_ref(), rebalance_status).to_string(); AdminDecommissionPoolStatus { id, @@ -505,7 +528,7 @@ impl DefaultAdminUsecase { } fn decommission_waiting_reason(info: &PoolDecommissionInfo) -> Option<&'static str> { - if info.complete || info.failed || info.canceled || info.start_time.is_some() { + if !info.has_decommission_state() || info.complete || info.failed || info.canceled || info.start_time.is_some() { return None; } if info.queued { @@ -614,7 +637,7 @@ mod tests { } #[test] - fn admin_pool_list_item_maps_capacity_and_unknown_decommission_status() { + fn admin_pool_list_item_maps_capacity_without_decommission_state_as_active() { let now = OffsetDateTime::UNIX_EPOCH; let pool = PoolStatus { id: 2, @@ -634,9 +657,32 @@ mod tests { assert_eq!(item.current_size, 250); assert_eq!(item.used_size, 750); assert!((item.used - 0.75).abs() < f64::EPSILON); + assert_eq!(item.status, "active"); + assert_eq!(item.decommission_status, "none"); + assert_eq!(item.rebalance_status, "none"); + assert!(item.decommission.is_none()); + } + + #[test] + fn admin_pool_list_item_maps_inconsistent_decommission_progress_as_unknown() { + let item = DefaultAdminUsecase::pool_list_item_from_status( + PoolStatus { + id: 2, + cmd_line: "http://node{1...4}/disk{1...4}".to_string(), + last_update: OffsetDateTime::UNIX_EPOCH, + decommission: Some(PoolDecommissionInfo { + total_size: 1_000, + current_size: 250, + items_decommissioned: 1, + ..Default::default() + }), + }, + (RebalStatus::None, false), + ); + assert_eq!(item.status, "decommissioning"); assert_eq!(item.decommission_status, "unknown"); - assert_eq!(item.rebalance_status, "none"); + assert!(item.decommission.is_some()); } #[test] @@ -713,6 +759,26 @@ mod tests { assert_eq!(item.rebalance_status, "started"); } + #[test] + fn admin_pool_list_item_keeps_decommission_state_ahead_of_rebalance_state() { + let item = DefaultAdminUsecase::pool_list_item_from_status( + PoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update: OffsetDateTime::UNIX_EPOCH, + decommission: Some(PoolDecommissionInfo { + failed: true, + ..Default::default() + }), + }, + (RebalStatus::Started, false), + ); + + assert_eq!(item.status, "blocked"); + assert_eq!(item.decommission_status, "failed"); + assert_eq!(item.rebalance_status, "started"); + } + #[test] fn admin_pool_list_item_exposes_queued_decommission_state() { let item = DefaultAdminUsecase::pool_list_item_from_status( @@ -802,6 +868,23 @@ mod tests { assert_eq!(item.rebalance_status, "failed"); } + #[test] + fn admin_pool_list_item_maps_started_rebalance_to_pool_rebalancing() { + let item = DefaultAdminUsecase::pool_list_item_from_status( + PoolStatus { + id: 1, + cmd_line: "pool-1".to_string(), + last_update: OffsetDateTime::UNIX_EPOCH, + decommission: None, + }, + (RebalStatus::Started, false), + ); + + assert_eq!(item.status, "rebalancing"); + assert_eq!(item.decommission_status, "none"); + assert_eq!(item.rebalance_status, "started"); + } + #[test] fn admin_pool_list_item_maps_stopping_rebalance_status() { let item = DefaultAdminUsecase::pool_list_item_from_status( @@ -814,42 +897,56 @@ mod tests { (RebalStatus::Started, true), ); - assert_eq!(item.status, "active"); + assert_eq!(item.status, "rebalancing"); assert_eq!(item.decommission_status, "none"); assert_eq!(item.rebalance_status, "stopping"); } #[test] fn admin_pool_lifecycle_state_distinguishes_decommission_terminal_states() { - let complete = DefaultAdminUsecase::pool_lifecycle_state(Some(&PoolDecommissionInfo { - complete: true, - ..Default::default() - })); - let failed = DefaultAdminUsecase::pool_lifecycle_state(Some(&PoolDecommissionInfo { - failed: true, - ..Default::default() - })); - let canceled = DefaultAdminUsecase::pool_lifecycle_state(Some(&PoolDecommissionInfo { - canceled: true, - ..Default::default() - })); + let complete = DefaultAdminUsecase::pool_lifecycle_state( + Some(&PoolDecommissionInfo { + complete: true, + ..Default::default() + }), + DefaultAdminUsecase::REBALANCE_STATUS_NONE, + ); + let failed = DefaultAdminUsecase::pool_lifecycle_state( + Some(&PoolDecommissionInfo { + failed: true, + ..Default::default() + }), + DefaultAdminUsecase::REBALANCE_STATUS_NONE, + ); + let canceled = DefaultAdminUsecase::pool_lifecycle_state( + Some(&PoolDecommissionInfo { + canceled: true, + ..Default::default() + }), + DefaultAdminUsecase::REBALANCE_STATUS_NONE, + ); + let rebalancing = DefaultAdminUsecase::pool_lifecycle_state(None, DefaultAdminUsecase::REBALANCE_STATUS_STARTED); assert_eq!(complete, "decommissioned"); assert_eq!(failed, "blocked"); assert_eq!(canceled, "blocked"); + assert_eq!(rebalancing, "rebalancing"); } #[test] fn admin_decommission_status_serializes_task_status_and_pool_status() { - let item = DefaultAdminUsecase::decommission_pool_status_from_status(PoolStatus { - id: 3, - cmd_line: "pool-3".to_string(), - last_update: OffsetDateTime::UNIX_EPOCH, - decommission: Some(PoolDecommissionInfo { - failed: true, - ..Default::default() - }), - }); + let item = DefaultAdminUsecase::decommission_pool_status_from_status( + PoolStatus { + id: 3, + cmd_line: "pool-3".to_string(), + last_update: OffsetDateTime::UNIX_EPOCH, + decommission: Some(PoolDecommissionInfo { + failed: true, + ..Default::default() + }), + }, + (RebalStatus::Started, false), + ); let value = serde_json::to_value(item).expect("decommission status should serialize");