diff --git a/ecstore/src/pools.rs b/ecstore/src/pools.rs index 73f8afa29..0f977d35c 100644 --- a/ecstore/src/pools.rs +++ b/ecstore/src/pools.rs @@ -3,25 +3,35 @@ use crate::cache_value::metacache_set::{list_path_raw, AgreedFn, ListPathRawOpti use crate::config::com::{read_config, save_config, CONFIG_PREFIX}; use crate::config::error::ConfigError; use crate::disk::{MetaCacheEntries, MetaCacheEntry, MetadataResolutionParams, BUCKET_META_PREFIX, RUSTFS_META_BUCKET}; +use crate::heal::data_usage::DATA_USAGE_CACHE_NAME; use crate::heal::heal_commands::HealOpts; use crate::new_object_layer_fn; use crate::notification_sys::get_global_notification_sys; use crate::set_disk::SetDisks; -use crate::store_api::{BucketOptions, GetObjectReader, MakeBucketOptions, StorageAPI}; -use crate::store_err::{is_err_bucket_exists, StorageError}; -use crate::utils::path::{path_join, SLASH_SEPARATOR}; +use crate::store_api::{BucketOptions, GetObjectReader, MakeBucketOptions, ObjectIO, ObjectOptions, StorageAPI}; +use crate::store_err::{ + is_err_bucket_exists, is_err_data_movement_overwrite, is_err_object_not_found, is_err_version_not_found, StorageError, +}; +use crate::utils::path::{encode_dir_object, path_join, SLASH_SEPARATOR}; use crate::{sets::Sets, store::ECStore}; +use ::workers::workers::Workers; use byteorder::{ByteOrder, LittleEndian, WriteBytesExt}; use common::error::{Error, Result}; +use http::HeaderMap; use rmp_serde::{Deserializer, Serializer}; -use serde::{Deserialize, Serialize}; +use serde::{de, Deserialize, Serialize}; use std::fmt::Display; +use std::future::Future; use std::io::{Cursor, Write}; use std::path::PathBuf; +use std::pin::Pin; +use std::rc; use std::sync::Arc; -use time::OffsetDateTime; +use time::{Duration, OffsetDateTime}; use tokio::sync::broadcast::Receiver as B_Receiver; use tracing::{error, info, warn}; +use uuid::Uuid; +use workers::workers; pub const POOL_META_NAME: &str = "pool.bin"; pub const POOL_META_FORMAT: u16 = 1; @@ -277,6 +287,33 @@ impl PoolMeta { } } } + + pub fn track_current_bucket_object(&mut self, idx: usize, bucket: String, object: String) { + self.pools.get(idx).is_some_and(|v| v.decommission.is_some()); + if let Some(pool) = self.pools.get_mut(idx) { + if 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_some_and(|v| v.decommission.is_some()) { + return Err(Error::msg("InvalidArgument")); + } + + let now = OffsetDateTime::now_utc(); + + if now.unix_timestamp() - self.pools[idx].last_update.unix_timestamp() > duration.whole_seconds() as i64 { + self.pools[idx].last_update = now; + self.save(pools).await?; + + return Ok(true); + } + + Ok(false) + } } fn path2_bucket_object(name: &str) -> (String, String) { @@ -482,7 +519,7 @@ impl ECStore { false } - pub async fn decommission(&self, indices: Vec) -> Result<()> { + pub async fn decommission(&self, rx: B_Receiver, indices: Vec) -> Result<()> { if indices.is_empty() { return Err(Error::msg("errInvalidArgument")); } @@ -499,32 +536,257 @@ impl ECStore { return; }; for idx in indices.iter() { - store.do_decommission_in_routine(*idx).await; + store.do_decommission_in_routine(rx.resubscribe(), *idx).await; } }); Ok(()) } - async fn decommission_pool(&self, _idx: usize, pool: Arc, bi: DecomBucketInfo) -> Result<()> { - let mut vc = None; + async fn decommission_pool(&self, rx: B_Receiver, idx: usize, pool: Arc, bi: DecomBucketInfo) -> Result<()> { + let wk = Workers::new(pool.disk_set.len() * 2).map_err(|v| Error::from_string(v))?; + + // let mut vc = None; + // replication + let rcfg: Option = None; if bi.name != RUSTFS_META_BUCKET { - let versioning = BucketVersioningSys::get(&bi.name).await?; - vc = Some(versioning); + let _versioning = BucketVersioningSys::get(&bi.name).await?; + // vc = Some(versioning); // TODO: LifecycleSys // TODO: BucketObjectLockSys // TODO: ReplicationConfig } - for (idx, set) in pool.disk_set.iter().enumerate() {} + for (idx, set) in pool.disk_set.iter().enumerate() { + let set = set.clone(); + let wk_clone = wk.clone(); + let bucket = bi.name.clone(); + let rcfg = rcfg.clone(); - // FIXME: - unimplemented!() + let decommission_entry = |entry: MetaCacheEntry| async move { + wk_clone.give().await; + if entry.is_dir() { + 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() { + // TODO: filterLifecycle + let remaining_versions = fivs.versions.len() - expired; + if version.deleted && remaining_versions == 1 && rcfg.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 { + // TODO: other params + if let Err(err) = self + .delete_object( + bucket.as_str(), + &version.name, + ObjectOptions { + versioned: true, + version_id: version_id.clone(), + mod_time: version.mod_time, + src_pool_idx: idx, + data_movement: true, + delete_marker: true, + skip_decommissioned: 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; + 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() { + // TODO: DecomTieredObject + } + + 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; + } + }; + + if let Err(err) = self.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; + } + + failure = false; + break; + } + + if ignore { + info!("decommission_pool: ignore {}", &version.name); + continue; + } + + { + self.pool_meta.write().await.count_item(idx, decommissioned, failure); + } + + if failure { + break; + } + + decommissioned += 1; + } + + if decommissioned == fivs.versions.len() { + if 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); + } + } + + { + 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(); + if ok { + // TODO: ReloadPoolMeta + } + } + }; + + wk.clone().take().await; + + // let set = set.clone(); + // let mutrx = rx.resubscribe(); + // tokio::spawn(async move { + // loop { + // if rx.try_recv().is_ok() { + // break; + // } + + // let decommission_entry = Arc::new(move |entry: MetaCacheEntry| { + // let decommission_entry = decommission_entry.clone(); + // Box::pin(async move { + // decommission_entry(entry).await; + // }) as Pin + Send>> + // }); + // set.list_objects_to_decommission(rx, bi, decommission_entry).await.unwrap(); + // } + + // // + // }); + } + + wk.wait().await; + + Ok(()) } - async fn do_decommission_in_routine(&self, idx: usize) { - if let Err(err) = self.decommission_in_background(idx).await { + async fn do_decommission_in_routine(&self, rx: B_Receiver, idx: usize) { + if let Err(err) = self.decommission_in_background(rx, idx).await { error!("decom err {:?}", &err); if let Err(er) = self.decommission_failed(idx).await { error!("decom failed err {:?}", &er); @@ -588,7 +850,7 @@ impl ECStore { Ok(()) } - async fn decommission_in_background(&self, idx: usize) -> Result<()> { + async fn decommission_in_background(&self, rx: B_Receiver, idx: usize) -> Result<()> { let pool = self.pools[idx].clone(); let pending = { @@ -618,7 +880,10 @@ impl ECStore { info!("decommission: currently on bucket {}", &bucket.name); - if let Err(err) = self.decommission_pool(idx, pool.clone(), bucket.clone()).await { + if let Err(err) = self + .decommission_pool(rx.resubscribe(), idx, pool.clone(), bucket.clone()) + .await + { error!("decommission: decommission_pool err {:?}", &err); return Err(err); } @@ -708,7 +973,7 @@ impl ECStore { Ok(ret) } - fn decommission_object(&self, id: usize, bucket: String, rd: GetObjectReader) -> Result<()> { + async fn decommission_object(&self, id: usize, bucket: String, rd: GetObjectReader) -> Result<()> { let object_info = rd.object_info; let actual_size = object_info.get_actual_size()?; @@ -718,13 +983,17 @@ impl ECStore { } } +// impl Fn(MetaCacheEntry) -> impl Future> + +type Listcb = Arc Pin + Send>> + Send + Sync + 'static>; + impl SetDisks { // async fn list_objects_to_decommission( self: &Arc, - rx: B_Receiver, + mut rx: B_Receiver, bucket_info: DecomBucketInfo, - func: Arc, + func: Listcb, ) -> Result<()> { let (disks, _) = self.get_online_disks_with_healing(false).await; if disks.is_empty() { @@ -750,7 +1019,8 @@ impl SetDisks { min_disks: listing_quorum, partial: Some(Box::new(move |entries: MetaCacheEntries, _: &[Option]| { let resolver = resolver.clone(); - let func = func.clone(); + let func = + func.clone() as Arc Pin + Send>> + Send + Sync>; Box::pin(async move { if let Ok(Some(entry)) = entries.resolve(resolver) { func(entry); diff --git a/ecstore/src/store_err.rs b/ecstore/src/store_err.rs index b6b3532eb..be945e639 100644 --- a/ecstore/src/store_err.rs +++ b/ecstore/src/store_err.rs @@ -241,6 +241,14 @@ pub fn to_object_err(err: Error, params: Vec<&str>) -> Error { err } +pub fn is_err_data_movement_overwrite(err: &Error) -> bool { + if let Some(e) = err.downcast_ref::() { + matches!(e, StorageError::DataMovementOverwriteErr(_, _, _)) + } else { + false + } +} + pub fn is_err_read_quorum(err: &Error) -> bool { if let Some(e) = err.downcast_ref::() { matches!(e, StorageError::InsufficientReadQuorum)