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

fix(admin): correct pool lifecycle status mapping (#3823)

Browse Source
This commit is contained in:
cxymds
2026-06-24 17:05:07 +08:00
committed by GitHub
parent eef8f43251
commit 57e12e56d2
4 changed files with 251 additions and 53 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/pools.rs
+72 -9
View File
@@ -277,6 +277,9 @@ fn first_resumable_decommission_queue_indices(meta: &PoolMeta) -> Vec<usize> {
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,
crates/ecstore/src/rebalance/control.rs
+43 -5
View File
@@ -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<S>(
&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<String> {
@@ -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));
}
}
crates/ecstore/src/store/init.rs
+1 -1
View File
@@ -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)
})
}
rustfs/src/app/admin_usecase.rs
+135 -38
View File
@@ -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<AdminDecommissionStatus> {
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");