diff --git a/ecstore/src/pools.rs b/ecstore/src/pools.rs index 0e06a0e3f..ba607edf0 100644 --- a/ecstore/src/pools.rs +++ b/ecstore/src/pools.rs @@ -6,7 +6,7 @@ use crate::heal::heal_commands::HealOpts; use crate::new_object_layer_fn; use crate::store_api::{BucketOptions, MakeBucketOptions, StorageAPI, StorageDisk, StorageInfo}; use crate::store_err::{is_err_bucket_exists, StorageError}; -use crate::utils::path::path_join; +use crate::utils::path::{path_join, SLASH_SEPARATOR}; use crate::{sets::Sets, store::ECStore}; use byteorder::{ByteOrder, LittleEndian, WriteBytesExt}; use rmp_serde::{Deserializer, Serializer}; @@ -15,7 +15,7 @@ use std::io::{Cursor, Write}; use std::path::PathBuf; use std::sync::Arc; use time::OffsetDateTime; -use tracing::error; +use tracing::{error, info, warn}; pub const POOL_META_NAME: &str = "pool.bin"; pub const POOL_META_FORMAT: u16 = 1; @@ -143,6 +143,54 @@ impl PoolMeta { 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.start_time = None; + 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.start_time = None; + 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 { @@ -171,6 +219,74 @@ impl PoolMeta { } } } + pub fn pending_buckets(&self, idx: usize) -> Vec { + let mut list = Vec::new(); + + if let Some(pool) = self.pools.get(idx) { + if 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) { + if 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; + } + } + } + } +} + +fn path2_bucket_object(name: &String) -> (String, String) { + path2_bucket_object_with_base_path("", &name) +} + +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 pos = trimmed_path.find(SLASH_SEPARATOR).unwrap_or(trimmed_path.len()); + // 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)] @@ -219,6 +335,29 @@ impl PoolDecommissionInfo { } 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.clone()); + 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)] @@ -338,11 +477,122 @@ impl ECStore { Ok(()) } - async fn do_decommission_in_routine(&self, _idx: usize) { + async fn decommission_pool(&self, _idx: usize, _pool: Arc, _bucket: DecomBucketInfo) -> Result<()> { // FIXME: unimplemented!() } + async fn do_decommission_in_routine(&self, idx: usize) { + if let Err(err) = self.decommission_in_background(idx).await { + error!("decom err {:?}", &err); + if let Err(er) = self.decommission_failed(idx).await { + error!("decom failed err {:?}", &er); + } + + return; + } + + let (failed, cmd_line) = { + let pool_meta = self.pool_meta.read().await; + let failed = { + if let Some(info) = &pool_meta.pools[idx].decommission { + info.items_decommission_failed > 0 + } else { + false + } + }; + let cmd_line = pool_meta.pools[idx].cmd_line.clone(); + (failed, cmd_line) + }; + + if !failed { + warn!("Decommissioning complete for pool {}, verifying for any pending objects", cmd_line); + 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); + } + } + } + + pub async fn decommission_failed(&self, idx: usize) -> Result<()> { + if self.single_pool() { + return Err(Error::msg("errInvalidArgument")); + } + + let mut pool_meta = self.pool_meta.write().await; + if pool_meta.decommission_failed(idx) { + pool_meta.save(self.pools.clone()).await?; + // FIXME: globalNotificationSys.ReloadPoolMeta(ctx) + } + + Ok(()) + } + + pub async fn complete_decommission(&self, idx: usize) -> Result<()> { + if self.single_pool() { + return Err(Error::msg("errInvalidArgument")); + } + + let mut pool_meta = self.pool_meta.write().await; + if pool_meta.decommission_complete(idx) { + pool_meta.save(self.pools.clone()).await?; + // FIXME: globalNotificationSys.ReloadPoolMeta(ctx) + } + + Ok(()) + } + + async fn decommission_in_background(&self, 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 { + info!("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()) { + if let Err(err) = pool_meta.save(self.pools.clone()).await { + error!("decom pool_meta.save err {:?}", err); + } + } + } + continue; + } + + info!("decommission: currently on bucket {}", &bucket.name); + + if let Err(err) = self.decommission_pool(idx, pool.clone(), bucket.clone()).await { + error!("decommission: decommission_pool err {:?}", &err); + return Err(err); + } + + { + let mut pool_meta = self.pool_meta.write().await; + if pool_meta.bucket_done(idx, bucket.to_string()) { + if let Err(err) = pool_meta.save(self.pools.clone()).await { + error!("decom pool_meta.save err {:?}", err); + } + } + } + } + + Ok(()) + } + pub async fn start_decommission(&self, indices: Vec) -> Result<()> { if indices.is_empty() { return Err(Error::msg("errInvalidArgument"));