// 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. use crate::bucket::versioning_sys::BucketVersioningSys; use crate::bucket::{ lifecycle::{ bucket_lifecycle_audit::LcEventSrc, bucket_lifecycle_ops::{ LifecycleOps, apply_expiry_on_transitioned_object, apply_expiry_rule, eval_action_from_lifecycle, }, lifecycle::IlmAction, }, metadata_sys, object_lock::objectlock_sys::BucketObjectLockSys, }; use crate::cache_value::metacache_set::{ListPathRawOptions, list_path_raw}; use crate::config::com::{CONFIG_PREFIX, read_config, save_config}; use crate::data_usage::DATA_USAGE_CACHE_NAME; use crate::disk::error::DiskError; use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET}; use crate::error::{Error, Result}; use crate::error::{ StorageError, is_err_bucket_exists, is_err_bucket_not_found, is_err_data_movement_overwrite, is_err_object_not_found, is_err_version_not_found, }; use crate::new_object_layer_fn; use crate::notification_sys::get_global_notification_sys; use crate::set_disk::SetDisks; use crate::store_api::{ BucketOperations, BucketOptions, CompletePart, GetObjectReader, HealOperations, MakeBucketOptions, MultipartOperations, ObjectIO, ObjectOperations, ObjectOptions, PutObjReader, StorageAPI, }; use crate::{global::GLOBAL_LifecycleSys, sets::Sets, store::ECStore}; use byteorder::{ByteOrder, LittleEndian, WriteBytesExt}; use bytes::Bytes; use futures::future::BoxFuture; use http::HeaderMap; #[cfg(test)] use rmp_serde::Deserializer; use rmp_serde::Serializer; use rustfs_common::defer; use rustfs_common::heal_channel::HealOpts; use rustfs_filemeta::{FileInfoVersions, MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams}; use rustfs_rio::{EtagResolvable, HashReader, HashReaderDetector, Index, Reader, TryGetIndex, WarpReader}; use rustfs_utils::path::{SLASH_SEPARATOR, encode_dir_object, path_join}; use rustfs_workers::workers::Workers; use s3s::dto::{BucketLifecycleConfiguration, DefaultRetention, ReplicationConfiguration}; use serde::{Deserialize, Serialize}; use std::collections::{HashMap, HashSet}; use std::fmt::Display; use std::io::{Cursor, Write}; use std::path::PathBuf; use std::pin::Pin; use std::sync::{ Arc, atomic::{AtomicUsize, Ordering}, }; use std::task::{Context, Poll}; use time::{Duration, OffsetDateTime}; use tokio::io::{AsyncRead, AsyncReadExt, BufReader, ReadBuf}; use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, warn}; pub const POOL_META_NAME: &str = "pool.bin"; pub const POOL_META_FORMAT: u16 = 1; pub const POOL_META_VERSION: u16 = 1; #[derive(Debug, Clone, Serialize, Deserialize)] pub struct PoolStatus { #[serde(rename = "id")] pub id: usize, #[serde(rename = "cmdline")] pub cmd_line: String, #[serde(rename = "lastUpdate", with = "time::serde::rfc3339")] pub last_update: OffsetDateTime, #[serde(rename = "decommissionInfo")] pub decommission: Option, } #[derive(Debug, Clone, Default, Serialize, Deserialize)] pub struct PoolMeta { pub version: u16, pub pools: Vec, pub dont_save: bool, } #[derive(Debug, Clone, Serialize, Deserialize)] struct PersistedPoolMeta { pub version: u16, pub pools: Vec, } #[derive(Debug, Clone, Serialize, Deserialize)] struct PersistedPoolStatus { #[serde(rename = "id")] pub id: usize, #[serde(rename = "cmdline")] pub cmd_line: String, #[serde(rename = "lastUpdate", with = "time::serde::rfc3339")] pub last_update: OffsetDateTime, #[serde(rename = "decommissionInfo")] pub decommission: Option, } #[derive(Debug, Clone, Default, Serialize, Deserialize)] struct PersistedPoolDecommissionInfo { #[serde(rename = "startTime", with = "time::serde::rfc3339::option")] pub start_time: Option, #[serde(rename = "startSize")] pub start_size: usize, #[serde(rename = "totalSize")] pub total_size: usize, #[serde(rename = "currentSize")] pub current_size: usize, #[serde(rename = "complete")] pub complete: bool, #[serde(rename = "failed")] pub failed: bool, #[serde(rename = "canceled")] pub canceled: bool, #[serde(rename = "queuedBuckets", default)] pub queued_buckets: Vec, #[serde(rename = "decommissionedBuckets", default)] pub decommissioned_buckets: Vec, #[serde(rename = "bucket", default)] pub bucket: String, #[serde(rename = "prefix", default)] pub prefix: String, #[serde(rename = "object", default)] pub object: String, #[serde(rename = "objectsDecommissioned")] pub items_decommissioned: usize, #[serde(rename = "objectsDecommissionedFailed")] pub items_decommission_failed: usize, #[serde(rename = "bytesDecommissioned")] pub bytes_done: usize, #[serde(rename = "bytesDecommissionedFailed")] pub bytes_failed: usize, } impl From for PoolMeta { fn from(value: PersistedPoolMeta) -> Self { Self { version: value.version, pools: value.pools.into_iter().map(Into::into).collect(), dont_save: false, } } } impl From for PoolStatus { fn from(value: PersistedPoolStatus) -> Self { Self { id: value.id, cmd_line: value.cmd_line, last_update: value.last_update, decommission: value.decommission.map(Into::into), } } } impl From for PoolDecommissionInfo { fn from(value: PersistedPoolDecommissionInfo) -> Self { Self { start_time: value.start_time, start_size: value.start_size, total_size: value.total_size, current_size: value.current_size, complete: value.complete, failed: value.failed, canceled: value.canceled, queued_buckets: value.queued_buckets, decommissioned_buckets: value.decommissioned_buckets, bucket: value.bucket, prefix: value.prefix, object: value.object, items_decommissioned: value.items_decommissioned, items_decommission_failed: value.items_decommission_failed, bytes_done: value.bytes_done, bytes_failed: value.bytes_failed, } } } impl From<&PoolMeta> for PersistedPoolMeta { fn from(value: &PoolMeta) -> Self { Self { version: value.version, pools: value.pools.iter().map(Into::into).collect(), } } } impl From<&PoolStatus> for PersistedPoolStatus { fn from(value: &PoolStatus) -> Self { Self { id: value.id, cmd_line: value.cmd_line.clone(), last_update: value.last_update, decommission: value.decommission.as_ref().map(Into::into), } } } impl From<&PoolDecommissionInfo> for PersistedPoolDecommissionInfo { fn from(value: &PoolDecommissionInfo) -> Self { Self { start_time: value.start_time, start_size: value.start_size, total_size: value.total_size, current_size: value.current_size, complete: value.complete, failed: value.failed, canceled: value.canceled, queued_buckets: value.queued_buckets.clone(), decommissioned_buckets: value.decommissioned_buckets.clone(), bucket: value.bucket.clone(), prefix: value.prefix.clone(), object: value.object.clone(), items_decommissioned: value.items_decommissioned, items_decommission_failed: value.items_decommission_failed, bytes_done: value.bytes_done, bytes_failed: value.bytes_failed, } } } impl PoolMeta { fn decode_pool_meta_payload(payload: &[u8]) -> Result { match rmp_serde::from_slice::(payload) { Ok(meta) => Ok(meta.into()), Err(persisted_err) => { let mut legacy: PoolMeta = rmp_serde::from_slice(payload).map_err(|legacy_err| { Error::other(format!( "PoolMeta decode failed for both persisted and legacy formats: persisted={persisted_err}; legacy={legacy_err}" )) })?; // Runtime-only flag must not be restored from on-disk payload. legacy.dont_save = false; Ok(legacy) } } } pub fn new(pools: &[Arc], prev_meta: &PoolMeta) -> Self { let mut new_meta = Self { version: POOL_META_VERSION, pools: Vec::new(), ..Default::default() }; for (idx, pool) in pools.iter().enumerate() { let mut skip = false; for current_pool in prev_meta.pools.iter() { if current_pool.cmd_line == pool.endpoints.cmd_line { new_meta.pools.push(current_pool.clone()); skip = true; break; } } if skip { continue; } new_meta.pools.push(PoolStatus { cmd_line: pool.endpoints.cmd_line.clone(), id: idx, last_update: OffsetDateTime::now_utc(), decommission: None, }); } new_meta } pub fn is_suspended(&self, idx: usize) -> bool { if idx >= self.pools.len() { return false; } self.pools[idx].decommission.is_some() } pub async fn load(&mut self, pool: Arc, _pools: Vec>) -> Result<()> { let data = match read_config(pool, POOL_META_NAME).await { Ok(data) => { if data.is_empty() { return Ok(()); } else if data.len() <= 4 { return Err(Error::other("poolMeta: no data")); } data } Err(err) => { if err == Error::ConfigNotFound { return Ok(()); } return Err(err); } }; let format = LittleEndian::read_u16(&data[0..2]); if format != POOL_META_FORMAT { return Err(Error::other(format!("PoolMeta: unknown format: {format}"))); } let version = LittleEndian::read_u16(&data[2..4]); if version != POOL_META_VERSION { return Err(Error::other(format!("PoolMeta: unknown version: {version}"))); } *self = Self::decode_pool_meta_payload(&data[4..])?; if self.version != POOL_META_VERSION { return Err(Error::other(format!("unexpected PoolMeta version: {}", self.version))); } Ok(()) } pub async fn save(&self, pools: Vec>) -> Result<()> { if self.dont_save { return Ok(()); } let mut data = Vec::new(); data.write_u16::(POOL_META_FORMAT)?; data.write_u16::(POOL_META_VERSION)?; let mut buf = Vec::new(); PersistedPoolMeta::from(self).serialize(&mut Serializer::new(&mut buf))?; data.write_all(&buf)?; for pool in pools { save_config(pool, POOL_META_NAME, data.clone()).await?; } Ok(()) } pub fn decommission_cancel(&mut self, idx: usize) -> bool { if let Some(stats) = self.pools.get_mut(idx) { if let Some(d) = &stats.decommission { if !d.canceled { stats.last_update = OffsetDateTime::now_utc(); let mut pd = d.clone(); pd.canceled = true; pd.failed = false; pd.complete = false; stats.decommission = Some(pd); true } else { false } } else { false } } else { false } } pub fn decommission_failed(&mut self, idx: usize) -> bool { if let Some(stats) = self.pools.get_mut(idx) { if let Some(d) = &stats.decommission { if !d.failed { stats.last_update = OffsetDateTime::now_utc(); let mut pd = d.clone(); pd.canceled = false; pd.failed = true; pd.complete = false; stats.decommission = Some(pd); true } else { false } } else { false } } else { false } } pub fn decommission_complete(&mut self, idx: usize) -> bool { if let Some(stats) = self.pools.get_mut(idx) { if let Some(d) = &stats.decommission { if !d.complete { stats.last_update = OffsetDateTime::now_utc(); let mut pd = d.clone(); pd.canceled = false; pd.failed = false; pd.complete = true; stats.decommission = Some(pd); true } else { false } } else { false } } else { false } } pub fn decommission(&mut self, idx: usize, pi: PoolSpaceInfo) -> Result<()> { if let Some(pool) = self.pools.get_mut(idx) { if let Some(ref info) = pool.decommission && !info.complete && !info.failed && !info.canceled { return Err(StorageError::DecommissionAlreadyRunning); } let now = OffsetDateTime::now_utc(); pool.last_update = now; pool.decommission = Some(PoolDecommissionInfo { start_time: Some(now), start_size: pi.free, total_size: pi.total, current_size: pi.free, ..Default::default() }); } Ok(()) } pub fn queue_buckets(&mut self, idx: usize, bks: Vec) { for bk in bks.iter() { if let Some(dec) = self.pools[idx].decommission.as_mut() { dec.bucket_push(bk); } } } pub fn pending_buckets(&self, idx: usize) -> Vec { let mut list = Vec::new(); if let Some(pool) = self.pools.get(idx) && let Some(ref info) = pool.decommission { for bk in info.queued_buckets.iter() { let (name, prefix) = path2_bucket_object(bk); list.push(DecomBucketInfo { name, prefix }); } } list } pub fn is_bucket_decommissioned(&self, idx: usize, bucket: String) -> bool { if let Some(ref info) = self.pools[idx].decommission { info.is_bucket_decommissioned(&bucket) } else { false } } pub fn bucket_done(&mut self, idx: usize, bucket: String) -> bool { if let Some(pool) = self.pools.get_mut(idx) { if let Some(info) = pool.decommission.as_mut() { info.bucket_pop(&bucket) } else { false } } else { false } } pub fn count_item(&mut self, idx: usize, size: usize, failed: bool) { if let Some(pool) = self.pools.get_mut(idx) && let Some(info) = pool.decommission.as_mut() { if failed { info.items_decommission_failed += 1; info.bytes_failed += size; } else { info.items_decommissioned += 1; info.bytes_done += size; } } } pub fn track_current_bucket_object(&mut self, idx: usize, bucket: String, object: String) { if self.pools.get(idx).is_none_or(|v| v.decommission.is_none()) { return; } if let Some(pool) = self.pools.get_mut(idx) && let Some(info) = pool.decommission.as_mut() { info.object = object; info.bucket = bucket; } } pub async fn update_after(&mut self, idx: usize, pools: Vec>, duration: Duration) -> Result { if self.pools.get(idx).is_none_or(|v| v.decommission.is_none()) { return Err(Error::other("InvalidArgument")); } let now = OffsetDateTime::now_utc(); if now.unix_timestamp() - self.pools[idx].last_update.unix_timestamp() > duration.whole_seconds() { self.pools[idx].last_update = now; self.save(pools).await?; return Ok(true); } Ok(false) } #[allow(dead_code)] pub fn validate(&self, pools: Vec>) -> Result { struct PoolInfo { position: usize, completed: bool, decom_started: bool, } let mut remembered_pools = HashMap::new(); for (idx, pool) in self.pools.iter().enumerate() { let mut complete = false; let mut decom_started = false; if let Some(decommission) = &pool.decommission { if decommission.complete { complete = true; } decom_started = true; } remembered_pools.insert( pool.cmd_line.clone(), PoolInfo { position: idx, completed: complete, decom_started, }, ); } let mut specified_pools = HashMap::new(); for (idx, pool) in pools.iter().enumerate() { specified_pools.insert(pool.endpoints.cmd_line.clone(), idx); } let mut update = false; // Determine whether the selected pool should be removed from the retired list. for k in specified_pools.keys() { if let Some(pi) = remembered_pools.get(k) { if pi.completed { error!( "pool({}) = {} is decommissioned, please remove from server command line", pi.position + 1, k ); // return Err(Error::other(format!( // "pool({}) = {} is decommissioned, please remove from server command line", // pi.position + 1, // k // ))); } } else { // If the previous pool no longer exists, allow updates because a new pool may have been added. update = true; } } if specified_pools.len() == remembered_pools.len() { for (k, pi) in remembered_pools.iter() { if let Some(pos) = specified_pools.get(k) && *pos != pi.position { update = true; // Pool order changed, allow the update. } } } if !update { update = specified_pools.len() != remembered_pools.len(); } Ok(update) } pub fn return_resumable_pools(&self) -> Vec { let mut new_pools = Vec::new(); for pool in &self.pools { if let Some(decommission) = &pool.decommission { if decommission.complete || decommission.canceled { // Recovery is not required when: // - Decommissioning completed // - Decommissioning was cancelled continue; } // All other scenarios require recovery new_pools.push(pool.clone()); } } new_pools } } pub fn path2_bucket_object(name: &str) -> (String, String) { path2_bucket_object_with_base_path("", name) } pub fn path2_bucket_object_with_base_path(base_path: &str, path: &str) -> (String, String) { // Trim the base path and leading slash let trimmed_path = path .strip_prefix(base_path) .unwrap_or(path) .strip_prefix(SLASH_SEPARATOR) .unwrap_or(path); // Find the position of the first '/' let Some(pos) = trimmed_path.find(SLASH_SEPARATOR) else { return (trimmed_path.to_string(), "".to_string()); }; // Split into bucket and prefix let bucket = &trimmed_path[0..pos]; let prefix = &trimmed_path[pos + 1..]; // +1 to skip the '/' character if it exists (bucket.to_string(), prefix.to_string()) } #[derive(Debug, Clone, Serialize, Deserialize, Default)] pub struct PoolDecommissionInfo { #[serde(rename = "startTime", with = "time::serde::rfc3339::option")] pub start_time: Option, #[serde(rename = "startSize")] pub start_size: usize, #[serde(rename = "totalSize")] pub total_size: usize, #[serde(rename = "currentSize")] pub current_size: usize, #[serde(rename = "complete")] pub complete: bool, #[serde(rename = "failed")] pub failed: bool, #[serde(rename = "canceled")] pub canceled: bool, #[serde(skip)] pub queued_buckets: Vec, #[serde(skip)] pub decommissioned_buckets: Vec, #[serde(skip)] pub bucket: String, #[serde(skip)] pub prefix: String, #[serde(skip)] pub object: String, #[serde(rename = "objectsDecommissioned")] pub items_decommissioned: usize, #[serde(rename = "objectsDecommissionedFailed")] pub items_decommission_failed: usize, #[serde(rename = "bytesDecommissioned")] pub bytes_done: usize, #[serde(rename = "bytesDecommissionedFailed")] pub bytes_failed: usize, } impl PoolDecommissionInfo { pub fn bucket_push(&mut self, bucket: &DecomBucketInfo) { for b in self.queued_buckets.iter() { if self.is_bucket_decommissioned(b) { return; } if b == &bucket.to_string() { return; } } self.queued_buckets.push(bucket.to_string()); self.bucket = bucket.name.clone(); self.prefix = bucket.prefix.clone(); } pub fn is_bucket_decommissioned(&self, bucket: &String) -> bool { for b in self.decommissioned_buckets.iter() { if b == bucket { return true; } } false } pub fn bucket_pop(&mut self, bucket: &String) -> bool { self.decommissioned_buckets.push(bucket.clone()); let mut found = None; for (i, b) in self.queued_buckets.iter().enumerate() { if b == bucket { found = Some(i); break; } } if let Some(i) = found { self.queued_buckets.remove(i); if &self.bucket == bucket { self.bucket = "".to_owned(); self.prefix = "".to_owned(); self.object = "".to_owned(); } return true; } false } } #[derive(Debug)] pub struct PoolSpaceInfo { pub free: usize, pub total: usize, pub used: usize, } #[derive(Debug, Default, Clone)] pub struct DecomBucketInfo { pub name: String, pub prefix: String, } impl Display for DecomBucketInfo { fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { write!( f, "{}", path_join(&[PathBuf::from(self.name.clone()), PathBuf::from(self.prefix.clone())]).to_string_lossy() ) } } #[derive(Debug, Clone, Copy, PartialEq, Eq)] enum DecommissionFinalState { Complete, Failed, } fn determine_decommission_final_state(items_failed: usize, was_cancelled: bool) -> DecommissionFinalState { if items_failed > 0 || was_cancelled { DecommissionFinalState::Failed } else { DecommissionFinalState::Complete } } fn remaining_versions_after_decommission(fivs: &FileInfoVersions) -> usize { fivs.versions.iter().filter(|version| !version.deleted).count() } fn decommission_delete_marker_opts( version: &rustfs_filemeta::FileInfo, version_id: Option, src_pool_idx: usize, ) -> ObjectOptions { ObjectOptions { versioned: true, version_id, mod_time: version.mod_time, src_pool_idx, data_movement: true, delete_marker: true, skip_decommissioned: true, delete_replication: version.replication_state_internal.clone(), ..Default::default() } } async fn should_skip_lifecycle_for_decommission( store: Arc, bucket: &str, version: &rustfs_filemeta::FileInfo, lifecycle_config: Option<&BucketLifecycleConfiguration>, lock_retention: Option, replication_config: Option<(ReplicationConfiguration, OffsetDateTime)>, apply_actions: bool, ) -> bool { let Some(lifecycle_config) = lifecycle_config else { return false; }; let versioned = BucketVersioningSys::prefix_enabled(bucket, &version.name).await; let object_info = crate::store_api::ObjectInfo::from_file_info(version, bucket, &version.name, versioned); let event = eval_action_from_lifecycle(lifecycle_config, lock_retention, replication_config, &object_info).await; match event.action { IlmAction::DeleteRestoredAction | IlmAction::DeleteRestoredVersionAction => { if apply_actions && object_info.is_remote() { let _ = apply_expiry_on_transitioned_object(store, &object_info, &event, &LcEventSrc::Decom).await; } false } IlmAction::DeleteAction | IlmAction::DeleteVersionAction | IlmAction::DeleteAllVersionsAction | IlmAction::DelMarkerDeleteAllVersionsAction => { if apply_actions { let _ = apply_expiry_rule(&event, &LcEventSrc::Decom, &object_info).await; } true } _ => false, } } struct IndexedDecommissionReader { inner: R, index: Option, } impl IndexedDecommissionReader { fn new(inner: R, index: Option) -> Self { Self { inner, index } } } impl AsyncRead for IndexedDecommissionReader { fn poll_read(mut self: Pin<&mut Self>, cx: &mut Context<'_>, buf: &mut ReadBuf<'_>) -> Poll> { Pin::new(&mut self.inner).poll_read(cx, buf) } } impl EtagResolvable for IndexedDecommissionReader {} impl HashReaderDetector for IndexedDecommissionReader {} impl TryGetIndex for IndexedDecommissionReader { fn try_get_index(&self) -> Option<&Index> { self.index.as_ref() } } impl Reader for IndexedDecommissionReader {} fn decode_part_index(index: Option<&Bytes>) -> Option { let bytes = index?; let mut decoded = Index::new(); if decoded.load(bytes.as_ref()).is_ok() { Some(decoded) } else { None } } fn put_obj_reader_from_chunk(chunk: Vec, size: i64, actual_size: i64, index: Option) -> Result { use sha2::{Digest, Sha256}; let sha256hex = if !chunk.is_empty() { Some(hex_simd::encode_to_string(Sha256::digest(&chunk), hex_simd::AsciiCase::Lower)) } else { None }; let reader = IndexedDecommissionReader::new(WarpReader::new(Cursor::new(chunk)), index); let hash_reader = HashReader::new(Box::new(reader), size, actual_size, None, sha256hex, false)?; Ok(PutObjReader::new(hash_reader)) } impl ECStore { pub async fn status(&self, idx: usize) -> Result { let space_info = self.get_decommission_pool_space_info(idx).await?; let pool_meta = self.pool_meta.read().await; let mut pool_info = pool_meta.pools[idx].clone(); if let Some(d) = pool_info.decommission.as_mut() { d.total_size = space_info.total; d.current_size = space_info.free; } else { pool_info.decommission = Some(PoolDecommissionInfo { total_size: space_info.total, current_size: space_info.free, ..Default::default() }); } Ok(pool_info) } async fn get_decommission_pool_space_info(&self, idx: usize) -> Result { if let Some(sets) = self.pools.get(idx) { let mut info = sets.storage_info().await; info.backend = self.backend_info().await; let total = get_total_usable_capacity(&info.disks, &info); let free = get_total_usable_capacity_free(&info.disks, &info); Ok(PoolSpaceInfo { free, total, used: total - free, }) } else { Err(Error::other("InvalidArgument")) } } #[tracing::instrument(skip(self))] pub async fn decommission_cancel(&self, idx: usize) -> Result<()> { if self.single_pool() { return Err(Error::other("InvalidArgument")); } let canceler = { let mut cancelers = self.decommission_cancelers.write().await; let Some(slot) = cancelers.get_mut(idx) else { return Err(Error::other("InvalidArgument")); }; let Some(canceler) = slot.take() else { return Err(StorageError::DecommissionNotStarted); }; canceler }; let mut lock = self.pool_meta.write().await; if lock.decommission_cancel(idx) { lock.save(self.pools.clone()).await?; drop(lock); if let Some(notification_sys) = get_global_notification_sys() { notification_sys.reload_pool_meta().await; } } canceler.cancel(); Ok(()) } pub async fn is_decommission_running(&self) -> bool { let pool_meta = self.pool_meta.read().await; for pool in pool_meta.pools.iter() { if let Some(ref info) = pool.decommission && !info.complete && !info.failed && !info.canceled { return true; } } false } #[tracing::instrument(skip(self, rx))] pub async fn decommission(&self, rx: CancellationToken, indices: Vec) -> Result<()> { warn!("decommission: {:?}", indices); if indices.is_empty() { return Err(Error::other("InvalidArgument")); } if self.single_pool() { return Err(Error::other("InvalidArgument")); } self.start_decommission(indices.clone()).await?; let rx_clone = rx.clone(); tokio::spawn(async move { let Some(store) = new_object_layer_fn() else { error!("store not init"); return; }; for idx in indices.iter() { store.do_decommission_in_routine(rx_clone.clone(), *idx).await; } }); Ok(()) } #[allow(unused_assignments, clippy::too_many_arguments)] #[tracing::instrument(skip(self, set, wk, lifecycle_config, lock_retention, replication_config))] async fn decommission_entry( self: &Arc, idx: usize, entry: MetaCacheEntry, bucket: String, set: Arc, wk: Arc, lifecycle_config: Option, lock_retention: Option, replication_config: Option<(ReplicationConfiguration, OffsetDateTime)>, ) { warn!("decommission_entry: {} {}", &bucket, &entry.name); wk.give().await; if entry.is_dir() { warn!("decommission_entry: skip dir {}", &entry.name); return; } let mut fivs = match entry.file_info_versions(&bucket) { Ok(f) => f, Err(err) => { error!("decommission_pool: file_info_versions err {:?}", &err); return; } }; fivs.versions.sort_by(|a, b| b.mod_time.cmp(&a.mod_time)); let mut decommissioned: usize = 0; let mut expired: usize = 0; for version in fivs.versions.iter() { if should_skip_lifecycle_for_decommission( self.clone(), &bucket, version, lifecycle_config.as_ref(), lock_retention.clone(), replication_config.clone(), true, ) .await { expired += 1; decommissioned += 1; continue; } let remaining_versions = fivs.versions.len() - expired; if version.deleted && remaining_versions == 1 && replication_config.is_none() { // decommissioned += 1; info!("decommission_pool: DELETE marked object with no other non-current versions will be skipped"); continue; } let version_id = version.version_id.map(|v| v.to_string()); let mut ignore = false; let mut failure = false; let mut error = None; if version.deleted { if let Err(err) = self .delete_object( bucket.as_str(), &version.name, decommission_delete_marker_opts(version, version_id.clone(), idx), ) .await { if is_err_object_not_found(&err) || is_err_version_not_found(&err) || is_err_data_movement_overwrite(&err) { warn!( "decommission_pool: ignore delete-marker copy for {}/{} version {:?}: {:?}", &bucket, &version.name, &version_id, &err ); ignore = true; continue; } failure = true; error = Some(err) } { self.pool_meta.write().await.count_item(idx, 0, failure); } if !failure { decommissioned += 1; } info!( "decommission_pool: DecomCopyDeleteMarker {} {} {:?} {:?}", &bucket, &version.name, &version_id, error ); continue; } for _i in 0..3 { if version.is_remote() { if let Err(err) = self .decommission_tiered_object( bucket.as_str(), &version.name, version, &ObjectOptions { version_id: version_id.clone(), mod_time: version.mod_time, user_defined: version.metadata.clone(), src_pool_idx: idx, data_movement: true, ..Default::default() }, ) .await { if is_err_object_not_found(&err) || is_err_version_not_found(&err) || is_err_data_movement_overwrite(&err) { ignore = true; break; } failure = true; error!("decommission_pool: decommission_tiered_object err {:?}", &err); error = Some(err); } break; } let bucket = bucket.clone(); let rd = match set .get_object_reader( bucket.as_str(), &encode_dir_object(&version.name), None, HeaderMap::new(), &ObjectOptions { version_id: version_id.clone(), no_lock: true, ..Default::default() }, ) .await { Ok(rd) => rd, Err(err) => { if is_err_object_not_found(&err) || is_err_version_not_found(&err) { ignore = true; break; } if !ignore { // if bucket == RUSTFS_META_BUCKET && version.name.contains(DATA_USAGE_CACHE_NAME) { ignore = true; error!("decommission_pool: ignore data usage cache {}", &version.name); break; } } failure = true; error!("decommission_pool: get_object_reader err {:?}", &err); continue; } }; let bucket_name = bucket.clone(); let object_name = rd.object_info.name.clone(); if let Err(err) = self.clone().decommission_object(idx, bucket, rd).await { if is_err_object_not_found(&err) || is_err_version_not_found(&err) || is_err_data_movement_overwrite(&err) { ignore = true; break; } failure = true; error!("decommission_pool: decommission_object err {:?}", &err); continue; } warn!( "decommission_pool: decommission_object done {}/{} {}", &bucket_name, &object_name, &version.name ); failure = false; break; } if ignore { info!("decommission_pool: ignore {}", &version.name); continue; } { let size = usize::try_from(version.size).unwrap_or_default(); self.pool_meta.write().await.count_item(idx, size, failure); } if failure { break; } decommissioned += 1; } if decommissioned == fivs.versions.len() && let Err(err) = set .delete_object( bucket.as_str(), &encode_dir_object(&entry.name), ObjectOptions { delete_prefix: true, delete_prefix_object: true, ..Default::default() }, ) .await { error!("decommission_pool: delete_object err {:?}", &err); } else if decommissioned != fivs.versions.len() { warn!( "decommission_pool: source object retained for {}/{} because only {}/{} versions were decommissioned", &bucket, &entry.name, decommissioned, fivs.versions.len() ); } { let mut pool_meta = self.pool_meta.write().await; pool_meta.track_current_bucket_object(idx, bucket.clone(), entry.name.clone()); let ok = pool_meta .update_after(idx, self.pools.clone(), Duration::seconds(30)) .await .unwrap_or_default(); drop(pool_meta); if ok && let Some(notification_sys) = get_global_notification_sys() { notification_sys.reload_pool_meta().await; } } warn!("decommission_pool: decommission_entry done {} {}", &bucket, &entry.name); } #[tracing::instrument(skip(self, rx))] async fn decommission_pool( self: &Arc, rx: CancellationToken, idx: usize, pool: Arc, bi: DecomBucketInfo, ) -> Result<()> { let wk = Workers::new(pool.disk_set.len() * 2).map_err(Error::other)?; let mut lifecycle_config = None; let mut lock_retention = None; let mut replication_config = None; if bi.name != RUSTFS_META_BUCKET { let _ = BucketVersioningSys::get(&bi.name).await?; lifecycle_config = GLOBAL_LifecycleSys.get(&bi.name).await; lock_retention = BucketObjectLockSys::get(&bi.name).await; replication_config = match metadata_sys::get_replication_config(&bi.name).await { Ok(config) => Some(config), Err(Error::ConfigNotFound) => None, Err(err) => return Err(err), }; } for (set_idx, set) in pool.disk_set.iter().enumerate() { wk.clone().take().await; warn!("decommission_pool: decommission_pool {} {}", set_idx, &bi.name); let decommission_entry: ListCallback = Arc::new({ let this = Arc::clone(self); let bucket = bi.name.clone(); let wk = wk.clone(); let set = set.clone(); let lifecycle_config = lifecycle_config.clone(); let lock_retention = lock_retention.clone(); let replication_config = replication_config.clone(); move |entry: MetaCacheEntry| { let this = this.clone(); let bucket = bucket.clone(); let wk = wk.clone(); let set = set.clone(); let lifecycle_config = lifecycle_config.clone(); let lock_retention = lock_retention.clone(); let replication_config = replication_config.clone(); Box::pin(async move { wk.take().await; this.decommission_entry(idx, entry, bucket, set, wk, lifecycle_config, lock_retention, replication_config) .await }) } }); let set = set.clone(); let rx_clone = rx.clone(); let bi = bi.clone(); let set_id = set_idx; let wk_clone = wk.clone(); tokio::spawn(async move { loop { if rx_clone.is_cancelled() { warn!("decommission_pool: cancel {}", set_id); break; } warn!("decommission_pool: list_objects_to_decommission {} {}", set_id, &bi.name); match set .list_objects_to_decommission(rx_clone.clone(), bi.clone(), decommission_entry.clone()) .await { Ok(_) => { warn!("decommission_pool: list_objects_to_decommission {} done", set_id); break; } Err(err) => { error!("decommission_pool: list_objects_to_decommission {} err {:?}", set_id, &err); if is_err_bucket_not_found(&err) { warn!("decommission_pool: list_objects_to_decommission {} volume not found", set_id); break; } tokio::time::sleep(tokio::time::Duration::from_secs(5)).await; } } } wk_clone.give().await; }); } warn!("decommission_pool: decommission_pool wait {} {}", idx, &bi.name); wk.wait().await; warn!("decommission_pool: decommission_pool done {} {}", idx, &bi.name); Ok(()) } #[tracing::instrument(skip(self, rx))] pub async fn do_decommission_in_routine(self: &Arc, rx: CancellationToken, idx: usize) { let decommission_token = rx.child_token(); { let mut cancelers = self.decommission_cancelers.write().await; if let Some(slot) = cancelers.get_mut(idx) { *slot = Some(decommission_token.clone()); } } if let Err(err) = self.decommission_in_background(decommission_token.clone(), idx).await { error!("decom err {:?}", &err); if let Err(er) = self.decommission_failed(idx).await { error!("decom failed err {:?}", &er); } else { warn!("decommission: decommission_failed {}", idx); } return; } warn!("decommission: decommission_in_background complete {}", idx); let (final_state, cmd_line) = { let pool_meta = self.pool_meta.read().await; let final_state = { if let Some(info) = &pool_meta.pools[idx].decommission { determine_decommission_final_state(info.items_decommission_failed, info.canceled) } else { DecommissionFinalState::Failed } }; let cmd_line = pool_meta.pools[idx].cmd_line.clone(); (final_state, cmd_line) }; let mut completed_successfully = false; if final_state == DecommissionFinalState::Complete { warn!("Decommissioning complete for pool {}, verifying for any pending objects", cmd_line); if let Err(err) = self.check_after_decommission(idx).await { error!("decom post-check err {:?}", &err); if let Err(er) = self.decommission_failed(idx).await { error!("decom failed err {:?}", &er); } } else if let Err(er) = self.complete_decommission(idx).await { error!("decom complete err {:?}", &er); } else { completed_successfully = true; } } else if let Err(er) = self.decommission_failed(idx).await { error!("decom failed err {:?}", &er); } { let mut cancelers = self.decommission_cancelers.write().await; if let Some(slot) = cancelers.get_mut(idx) { *slot = None; } } if completed_successfully { warn!("Decommissioning complete for pool {}", cmd_line); } else { warn!("Decommissioning finished in failed state for pool {}", cmd_line); } } #[tracing::instrument(skip(self))] pub async fn decommission_failed(&self, idx: usize) -> Result<()> { if self.single_pool() { return Err(Error::other("errInvalidArgument")); } let mut pool_meta = self.pool_meta.write().await; if pool_meta.decommission_failed(idx) { pool_meta.save(self.pools.clone()).await?; drop(pool_meta); if let Some(notification_sys) = get_global_notification_sys() { notification_sys.reload_pool_meta().await; } } let canceler = { let mut cancelers = self.decommission_cancelers.write().await; cancelers.get_mut(idx).and_then(Option::take) }; if let Some(canceler) = canceler { canceler.cancel(); } Ok(()) } #[tracing::instrument(skip(self))] pub async fn complete_decommission(&self, idx: usize) -> Result<()> { if self.single_pool() { return Err(Error::other("errInvalidArgument")); } let mut pool_meta = self.pool_meta.write().await; if pool_meta.decommission_complete(idx) { pool_meta.save(self.pools.clone()).await?; drop(pool_meta); if let Some(notification_sys) = get_global_notification_sys() { notification_sys.reload_pool_meta().await; } } let canceler = { let mut cancelers = self.decommission_cancelers.write().await; cancelers.get_mut(idx).and_then(Option::take) }; if let Some(canceler) = canceler { canceler.cancel(); } Ok(()) } #[tracing::instrument(skip(self, rx))] async fn decommission_in_background(self: &Arc, rx: CancellationToken, idx: usize) -> Result<()> { let pool = self.pools[idx].clone(); let pending = { let pool_meta = self.pool_meta.read().await; pool_meta.pending_buckets(idx) }; for bucket in pending.iter() { let is_decommissioned = { let pool_meta = self.pool_meta.read().await; pool_meta.is_bucket_decommissioned(idx, bucket.to_string()) }; if is_decommissioned { warn!("decommission: already done, moving on {}", bucket.to_string()); { let mut pool_meta = self.pool_meta.write().await; if pool_meta.bucket_done(idx, bucket.to_string()) && let Err(err) = pool_meta.save(self.pools.clone()).await { error!("decom pool_meta.save err {:?}", err); } } continue; } warn!("decommission: currently on bucket {}", &bucket.name); if let Err(err) = self.decommission_pool(rx.clone(), idx, pool.clone(), bucket.clone()).await { error!("decommission: decommission_pool err {:?}", &err); return Err(err); } else { warn!("decommission: decommission_pool done {}", &bucket.name); } { let mut pool_meta = self.pool_meta.write().await; if pool_meta.bucket_done(idx, bucket.to_string()) && let Err(err) = pool_meta.save(self.pools.clone()).await { error!("decom pool_meta.save err {:?}", err); } warn!("decommission: decommission_pool bucket_done {}", &bucket.name); } } Ok(()) } #[tracing::instrument(skip(self))] pub async fn start_decommission(&self, indices: Vec) -> Result<()> { if indices.is_empty() { return Err(Error::other("errInvalidArgument")); } if self.single_pool() { return Err(Error::other("errInvalidArgument")); } let decom_buckets = self.get_buckets_to_decommission().await?; for bk in decom_buckets.iter() { let _ = self.heal_bucket(&bk.name, &HealOpts::default()).await; } let meta_buckets = [ path_join(&[PathBuf::from(RUSTFS_META_BUCKET), PathBuf::from(CONFIG_PREFIX)]), path_join(&[PathBuf::from(RUSTFS_META_BUCKET), PathBuf::from(BUCKET_META_PREFIX)]), ]; for bk in meta_buckets.iter() { if let Err(err) = self .make_bucket(bk.to_string_lossy().to_string().as_str(), &MakeBucketOptions::default()) .await && !is_err_bucket_exists(&err) { error!("decommission: make bucket failed: {err}"); return Err(err); } } let mut pool_meta = self.pool_meta.write().await; for idx in indices.iter() { let pi = self.get_decommission_pool_space_info(*idx).await?; pool_meta.decommission(*idx, pi)?; pool_meta.queue_buckets(*idx, decom_buckets.clone()); } pool_meta.save(self.pools.clone()).await?; if let Some(notification_sys) = get_global_notification_sys() { notification_sys.reload_pool_meta().await; } Ok(()) } async fn get_buckets_to_decommission(&self) -> Result> { let buckets = self.list_bucket(&BucketOptions::default()).await?; let mut ret: Vec = buckets .iter() .map(|v| DecomBucketInfo { name: v.name.clone(), ..Default::default() }) .collect(); ret.push(DecomBucketInfo { name: RUSTFS_META_BUCKET.to_owned(), prefix: CONFIG_PREFIX.to_owned(), }); ret.push(DecomBucketInfo { name: RUSTFS_META_BUCKET.to_owned(), prefix: BUCKET_META_PREFIX.to_owned(), }); Ok(ret) } async fn check_after_decommission(self: &Arc, idx: usize) -> Result<()> { let buckets = self.get_buckets_to_decommission().await?; let pool = self.pools[idx].clone(); for set in &pool.disk_set { for bucket_info in &buckets { let mut lifecycle_config = None; let mut lock_retention = None; let mut replication_config = None; if bucket_info.name != RUSTFS_META_BUCKET { lifecycle_config = GLOBAL_LifecycleSys.get(&bucket_info.name).await; lock_retention = BucketObjectLockSys::get(&bucket_info.name).await; replication_config = match metadata_sys::get_replication_config(&bucket_info.name).await { Ok(config) => Some(config), Err(Error::ConfigNotFound) => None, Err(err) => return Err(err), }; } let versions_found = Arc::new(AtomicUsize::new(0)); let versions_found_cb = versions_found.clone(); let bucket_name = bucket_info.name.clone(); let lifecycle_config_cb = lifecycle_config.clone(); let lock_retention_cb = lock_retention.clone(); let replication_config_cb = replication_config.clone(); let store = Arc::clone(self); let callback: ListCallback = Arc::new(move |entry: MetaCacheEntry| { let versions_found = versions_found_cb.clone(); let bucket_name = bucket_name.clone(); let lifecycle_config = lifecycle_config_cb.clone(); let lock_retention = lock_retention_cb.clone(); let replication_config = replication_config_cb.clone(); let store = Arc::clone(&store); Box::pin(async move { if !entry.is_object() { return; } if bucket_name == RUSTFS_META_BUCKET && entry.name.contains(DATA_USAGE_CACHE_NAME) { return; } let Ok(fivs) = entry.file_info_versions(&bucket_name) else { return; }; let mut remaining = 0; for version in &fivs.versions { if version.deleted { continue; } if should_skip_lifecycle_for_decommission( Arc::clone(&store), &bucket_name, version, lifecycle_config.as_ref(), lock_retention.clone(), replication_config.clone(), false, ) .await { continue; } remaining += 1; } versions_found.fetch_add(remaining, Ordering::Relaxed); }) }); set.list_objects_to_decommission(CancellationToken::new(), bucket_info.clone(), callback) .await?; let versions_found = versions_found.load(Ordering::Relaxed); if versions_found > 0 { return Err(Error::other(format!( "at least {versions_found} object(s)/version(s) were found in bucket `{}` after decommissioning", bucket_info.name ))); } } } Ok(()) } #[tracing::instrument(skip(self, rd))] async fn decommission_object(self: Arc, pool_idx: usize, bucket: String, rd: GetObjectReader) -> Result<()> { warn!("decommission_object: start {} {}", &bucket, &rd.object_info.name); let object_info = rd.object_info.clone(); // TODO: check : use size or actual_size ? let _actual_size = object_info.get_actual_size()?; if object_info.is_multipart() { let res = match self .new_multipart_upload( &bucket, &object_info.name, &ObjectOptions { version_id: object_info.version_id.as_ref().map(|v| v.to_string()), user_defined: object_info.user_defined.clone(), src_pool_idx: pool_idx, data_movement: true, ..Default::default() }, ) .await { Ok(res) => res, Err(err) => { error!("decommission_object: new_multipart_upload err {:?}", &err); return Err(err); } }; defer!(|| async { if let Err(err) = self .abort_multipart_upload(&bucket, &object_info.name, &res.upload_id, &ObjectOptions::default()) .await { error!("decommission_object: abort_multipart_upload err {:?}", &err); } }); let mut parts = vec![CompletePart::default(); object_info.parts.len()]; let mut reader = rd.stream; for (i, part) in object_info.parts.iter().enumerate() { let mut chunk = vec![0u8; part.size]; reader.read_exact(&mut chunk).await?; let part_size = i64::try_from(part.size).map_err(|_| Error::other("part size overflow"))?; let part_actual_size = if part.actual_size > 0 { part.actual_size } else { part_size }; let index = decode_part_index(part.index.as_ref()); let mut data = put_obj_reader_from_chunk(chunk, part_size, part_actual_size, index)?; let pi = match self .put_object_part( &bucket, &object_info.name, &res.upload_id, part.number, &mut data, &ObjectOptions { preserve_etag: Some(part.etag.clone()), ..Default::default() }, ) .await { Ok(pi) => pi, Err(err) => { error!("decommission_object: put_object_part {} err {:?}", i, &err); return Err(err); } }; warn!("decommission_object: put_object_part {} done {} {}", i, &bucket, &object_info.name); parts[i] = CompletePart { part_num: pi.part_num, etag: pi.etag, ..Default::default() }; } if let Err(err) = self .clone() .complete_multipart_upload( &bucket, &object_info.name, &res.upload_id, parts, &ObjectOptions { data_movement: true, mod_time: object_info.mod_time, ..Default::default() }, ) .await { error!("decommission_object: complete_multipart_upload err {:?}", &err); return Err(err); } warn!("decommission_object: complete_multipart_upload done {} {}", &bucket, &object_info.name); return Ok(()); } let actual_size = object_info.get_actual_size()?; let index = object_info .parts .first() .and_then(|part| decode_part_index(part.index.as_ref())); let reader = IndexedDecommissionReader::new(WarpReader::new(BufReader::new(rd.stream)), index); let hrd = HashReader::new(Box::new(reader), object_info.size, actual_size, object_info.etag.clone(), None, false)?; let mut data = PutObjReader::new(hrd); if let Err(err) = self .put_object( &bucket, &object_info.name, &mut data, &ObjectOptions { src_pool_idx: pool_idx, data_movement: true, version_id: object_info.version_id.as_ref().map(|v| v.to_string()), mod_time: object_info.mod_time, user_defined: object_info.user_defined.clone(), preserve_etag: object_info.etag.clone(), ..Default::default() }, ) .await { error!("decommission_object: put_object err {:?}", &err); return Err(err); } warn!("decommission_object: put_object done {} {}", &bucket, &object_info.name); Ok(()) } } #[cfg(test)] #[allow(clippy::items_after_test_module)] mod tests { use super::*; #[test] fn determine_decommission_final_state_marks_failures_and_cancellations() { assert_eq!(determine_decommission_final_state(0, false), DecommissionFinalState::Complete); assert_eq!(determine_decommission_final_state(1, false), DecommissionFinalState::Failed); assert_eq!(determine_decommission_final_state(0, true), DecommissionFinalState::Failed); } #[test] fn remaining_versions_after_decommission_ignores_delete_markers() { let fivs = FileInfoVersions { versions: vec![ rustfs_filemeta::FileInfo { deleted: false, size: 128, ..Default::default() }, rustfs_filemeta::FileInfo { deleted: true, size: 0, ..Default::default() }, ], ..Default::default() }; assert_eq!(remaining_versions_after_decommission(&fivs), 1); } #[test] fn decommission_delete_marker_opts_preserves_replication_state() { let mod_time = OffsetDateTime::now_utc(); let version = rustfs_filemeta::FileInfo { mod_time: Some(mod_time), replication_state_internal: Some(rustfs_filemeta::ReplicationState { replica_status: rustfs_filemeta::ReplicationStatusType::Replica, delete_marker: true, replicate_decision_str: "existing".to_string(), ..Default::default() }), ..Default::default() }; let opts = decommission_delete_marker_opts(&version, Some("version-id".to_string()), 7); let replication = opts.delete_replication.expect("replication state should be preserved"); assert!(opts.versioned); assert!(opts.data_movement); assert!(opts.delete_marker); assert!(opts.skip_decommissioned); assert_eq!(opts.src_pool_idx, 7); assert_eq!(opts.version_id.as_deref(), Some("version-id")); assert_eq!(opts.mod_time, Some(mod_time)); assert_eq!(replication.replica_status, rustfs_filemeta::ReplicationStatusType::Replica); assert!(replication.delete_marker); assert_eq!(replication.replicate_decision_str, "existing"); } #[test] fn decommission_state_transitions_preserve_start_time() { let start_time = OffsetDateTime::now_utc(); let mut pool_meta = PoolMeta { version: POOL_META_VERSION, pools: vec![PoolStatus { id: 0, cmd_line: "/tmp/pool".to_string(), last_update: start_time, decommission: Some(PoolDecommissionInfo { start_time: Some(start_time), ..Default::default() }), }], dont_save: true, }; assert!(pool_meta.decommission_failed(0)); assert_eq!( pool_meta.pools[0].decommission.as_ref().and_then(|info| info.start_time), Some(start_time) ); assert!(pool_meta.decommission_complete(0)); assert_eq!( pool_meta.pools[0].decommission.as_ref().and_then(|info| info.start_time), Some(start_time) ); assert!(pool_meta.decommission_cancel(0)); assert_eq!( pool_meta.pools[0].decommission.as_ref().and_then(|info| info.start_time), Some(start_time) ); } #[test] fn pool_meta_persists_decommission_resume_queues() { let start_time = OffsetDateTime::now_utc(); let pool_meta = PoolMeta { version: POOL_META_VERSION, pools: vec![PoolStatus { id: 1, cmd_line: "/data/pool1/disk{1...4}".to_string(), last_update: start_time, decommission: Some(PoolDecommissionInfo { start_time: Some(start_time), queued_buckets: vec!["bucket-a".to_string(), "bucket-b/prefix".to_string()], decommissioned_buckets: vec!["bucket-done".to_string()], bucket: "bucket-b".to_string(), prefix: "prefix".to_string(), object: "object.txt".to_string(), items_decommissioned: 7, items_decommission_failed: 1, bytes_done: 1024, bytes_failed: 128, ..Default::default() }), }], dont_save: false, }; let mut buf = Vec::new(); PersistedPoolMeta::from(&pool_meta) .serialize(&mut Serializer::new(&mut buf)) .expect("pool meta should serialize"); let mut deserializer = Deserializer::new(Cursor::new(&buf)); let restored: PoolMeta = PersistedPoolMeta::deserialize(&mut deserializer) .expect("pool meta should deserialize") .into(); let restored_decommission = restored.pools[0] .decommission .as_ref() .expect("decommission info should survive round-trip"); assert_eq!( restored_decommission.queued_buckets, vec!["bucket-a".to_string(), "bucket-b/prefix".to_string()] ); assert_eq!(restored_decommission.decommissioned_buckets, vec!["bucket-done".to_string()]); assert_eq!(restored_decommission.bucket, "bucket-b"); assert_eq!(restored_decommission.prefix, "prefix"); assert_eq!(restored_decommission.object, "object.txt"); assert_eq!(restored_decommission.items_decommissioned, 7); assert_eq!(restored_decommission.items_decommission_failed, 1); assert_eq!(restored_decommission.bytes_done, 1024); assert_eq!(restored_decommission.bytes_failed, 128); } #[test] fn pool_meta_decode_supports_legacy_payload() { let start_time = OffsetDateTime::now_utc(); let legacy_meta = PoolMeta { version: POOL_META_VERSION, pools: vec![PoolStatus { id: 3, cmd_line: "/legacy/pool".to_string(), last_update: start_time, decommission: Some(PoolDecommissionInfo { start_time: Some(start_time), items_decommissioned: 9, items_decommission_failed: 2, bytes_done: 2048, bytes_failed: 256, queued_buckets: vec!["not-persisted".to_string()], decommissioned_buckets: vec!["not-persisted".to_string()], bucket: "not-persisted".to_string(), prefix: "not-persisted".to_string(), object: "not-persisted".to_string(), ..Default::default() }), }], dont_save: true, }; let mut legacy_payload = Vec::new(); legacy_meta .serialize(&mut Serializer::new(&mut legacy_payload)) .expect("legacy payload should serialize"); // New persisted schema has fewer top-level fields and should not decode this legacy struct payload. let persisted_decode: std::result::Result = rmp_serde::from_slice(&legacy_payload); assert!(persisted_decode.is_err()); let decoded = PoolMeta::decode_pool_meta_payload(&legacy_payload).expect("legacy payload should decode"); assert_eq!(decoded.version, POOL_META_VERSION); assert!(!decoded.dont_save, "runtime-only flag should reset on load"); assert_eq!(decoded.pools.len(), 1); assert_eq!(decoded.pools[0].id, 3); assert_eq!(decoded.pools[0].cmd_line, "/legacy/pool"); assert_eq!(decoded.pools[0].last_update, start_time); let decommission = decoded.pools[0].decommission.as_ref().expect("decommission should decode"); assert_eq!(decommission.start_time, Some(start_time)); assert_eq!(decommission.items_decommissioned, 9); assert_eq!(decommission.items_decommission_failed, 2); assert_eq!(decommission.bytes_done, 2048); assert_eq!(decommission.bytes_failed, 256); // These fields were skipped in legacy payload and should be defaulted. assert!(decommission.queued_buckets.is_empty()); assert!(decommission.decommissioned_buckets.is_empty()); assert!(decommission.bucket.is_empty()); assert!(decommission.prefix.is_empty()); assert!(decommission.object.is_empty()); } } // impl Fn(MetaCacheEntry) -> impl Future> pub type ListCallback = Arc BoxFuture<'static, ()> + Send + Sync + 'static>; impl SetDisks { #[tracing::instrument(skip(self, rx, cb_func))] async fn list_objects_to_decommission( self: &Arc, rx: CancellationToken, bucket_info: DecomBucketInfo, cb_func: ListCallback, ) -> Result<()> { let (disks, _) = self.get_online_disks_with_healing(false).await; if disks.is_empty() { return Err(Error::other("errNoDiskAvailable")); } let listing_quorum = self.set_drive_count.div_ceil(2); let resolver = MetadataResolutionParams { dir_quorum: listing_quorum, obj_quorum: listing_quorum, bucket: bucket_info.name.clone(), ..Default::default() }; let cb1 = cb_func.clone(); list_path_raw( rx, ListPathRawOptions { disks: disks.iter().cloned().map(Some).collect(), bucket: bucket_info.name.clone(), path: bucket_info.prefix.clone(), recursive: true, min_disks: listing_quorum, agreed: Some(Box::new(move |entry: MetaCacheEntry| Box::pin(cb1(entry)))), partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option]| { let resolver = resolver.clone(); let cb_func = cb_func.clone(); match entries.resolve(resolver) { Some(entry) => { warn!("decommission_pool: list_objects_to_decommission get {}", &entry.name); Box::pin(async move { cb_func(entry).await; }) } None => { warn!("decommission_pool: list_objects_to_decommission get none"); Box::pin(async {}) } } })), ..Default::default() }, ) .await?; Ok(()) } } fn is_disk_online_state(state: &str) -> bool { // The disk state strings are produced from rustfs_utils::os::get_drive_stats or DiskError::to_string(). // Conventionally, online is "ok"/"online" (may evolve). Be conservative: // - Treat empty as unknown -> include it (to avoid dropping capacity). // - Exclude explicit offline-ish states. let s = state.trim().to_lowercase(); if s.is_empty() { return true; } if s.contains("offline") { return false; } if s.contains("not found") || s.contains("disk not found") { return false; } true } #[deprecated(since = "0.1.0", note = "Use fallback_total_capacity_dedup instead")] #[allow(dead_code)] fn fallback_total_capacity(disks: &[rustfs_madmin::Disk]) -> usize { fallback_total_capacity_dedup(disks) } #[deprecated(since = "0.1.0", note = "Use fallback_free_capacity_dedup instead")] #[allow(dead_code)] fn fallback_free_capacity(disks: &[rustfs_madmin::Disk]) -> usize { fallback_free_capacity_dedup(disks) } pub fn get_total_usable_capacity(disks: &[rustfs_madmin::Disk], info: &rustfs_madmin::StorageInfo) -> usize { // If backend info is missing or inconsistent, do a safe fallback to avoid reporting nonsense. if info.backend.standard_sc_data.is_empty() { return fallback_total_capacity_dedup(disks); } let mut capacity = 0usize; let mut matched_any = false; let mut counted_disks: HashSet = HashSet::new(); for disk in disks.iter() { if disk.pool_index < 0 { continue; } let pool_idx = disk.pool_index as usize; if info.backend.standard_sc_data.len() <= pool_idx { continue; } let usable_disks_per_set = info.backend.standard_sc_data[pool_idx]; if usable_disks_per_set == 0 { continue; } if (disk.disk_index as usize) < usable_disks_per_set { // 🔧 Generate a unique identity using a combination of fields let disk_key = format!( "{}|{}|p{}s{}d{}", disk.endpoint, // Node address disk.drive_path, // mount path disk.pool_index, // Pool index disk.set_index, // Collection index disk.disk_index // Disk index ); debug!("get_total_usable_capacity disk_key: {}", disk_key); // 🔧 Only disks that have not been counted are counted towards capacity if counted_disks.insert(disk_key) { matched_any = true; capacity += disk.total_space as usize; } else { // Log duplicate disks: this likely indicates a configuration issue and should always be visible. warn!( "Duplicate disk detected in capacity calculation: {} at {}", disk.endpoint, disk.drive_path ); } } } if matched_any { capacity } else { // Even if standard_sc_data exists, it might not match disk indexes due to upstream bugs. // Fallback to summing all online disks to prevent under-reporting. fallback_total_capacity_dedup(disks) } } pub fn get_total_usable_capacity_free(disks: &[rustfs_madmin::Disk], info: &rustfs_madmin::StorageInfo) -> usize { if info.backend.standard_sc_data.is_empty() { return fallback_free_capacity_dedup(disks); } let mut capacity = 0usize; let mut matched_any = false; let mut counted_disks: HashSet = HashSet::new(); for disk in disks.iter() { if disk.pool_index < 0 { continue; } let pool_idx = disk.pool_index as usize; if info.backend.standard_sc_data.len() <= pool_idx { continue; } let usable_disks_per_set = info.backend.standard_sc_data[pool_idx]; if usable_disks_per_set == 0 { continue; } if (disk.disk_index as usize) < usable_disks_per_set { let disk_key = format!( "{}|{}|p{}s{}d{}", disk.endpoint, disk.drive_path, disk.pool_index, disk.set_index, disk.disk_index ); if counted_disks.insert(disk_key) { matched_any = true; capacity += disk.available_space as usize; } } } if matched_any { capacity } else { fallback_free_capacity_dedup(disks) } } /// Total fallback capacity calculation with deweight /// /// Replace original function: fallback_total_capacity() pub(crate) fn fallback_total_capacity_dedup(disks: &[rustfs_madmin::Disk]) -> usize { let mut counted_disks: HashSet = HashSet::new(); let mut total = 0usize; for disk in disks.iter() { // Only online disks are counted if !is_disk_online_state(&disk.state) { continue; } // Use endpoint + drive_path as a unique identifier let disk_key = format!("{}|{}", disk.endpoint, disk.drive_path); // Capacity is counted only when the disk is encountered for the first time if counted_disks.insert(disk_key) { total += disk.total_space as usize; } } total } /// Remove the heavy fallback idle capacity calculation /// /// Replace original function: fallback_free_capacity() pub(crate) fn fallback_free_capacity_dedup(disks: &[rustfs_madmin::Disk]) -> usize { let mut counted_disks: HashSet = HashSet::new(); let mut total = 0usize; for disk in disks.iter() { if !is_disk_online_state(&disk.state) { continue; } let disk_key = format!("{}|{}", disk.endpoint, disk.drive_path); if counted_disks.insert(disk_key) { total += disk.available_space as usize; } } total }