diff --git a/Cargo.lock b/Cargo.lock index a65c7ab5f..953b6e14f 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -678,7 +678,7 @@ dependencies = [ "sha2", "siphasher", "tempfile", - "thiserror", + "thiserror 1.0.68", "time", "tokio", "tokio-stream", @@ -1976,7 +1976,7 @@ checksum = "a3a7c64d9bf75b1b8d981124c14c179074e8caa7dfe7b6a12e6222ddcd0c8f72" dependencies = [ "once_cell", "protobuf-support", - "thiserror", + "thiserror 1.0.68", ] [[package]] @@ -1985,7 +1985,7 @@ version = "3.7.1" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "b088fd20b938a875ea00843b6faf48579462630015c3788d397ad6a786663252" dependencies = [ - "thiserror", + "thiserror 1.0.68", ] [[package]] @@ -2018,7 +2018,7 @@ dependencies = [ "num_cpus", "once_cell", "platforms", - "thiserror", + "thiserror 1.0.68", "unescape", ] @@ -2082,7 +2082,7 @@ dependencies = [ "pin-project-lite", "s3s", "sha2", - "thiserror", + "thiserror 1.0.68", "tokio", "tracing", ] @@ -2332,7 +2332,7 @@ checksum = "f3cb5ba0dc43242ce17de99c180e96db90b235b8a9fdc9543c96d2209116bd9f" [[package]] name = "s3s" version = "0.11.0-dev" -source = "git+https://github.com/Nugine/s3s.git?rev=207170f526c75a8190e8f9afadf961909fd01d34#207170f526c75a8190e8f9afadf961909fd01d34" +source = "git+https://github.com/Nugine/s3s.git?rev=3291c0ca0284971569499cbe75bd69ef7bde8321#3291c0ca0284971569499cbe75bd69ef7bde8321" dependencies = [ "arrayvec", "async-trait", @@ -2364,8 +2364,9 @@ dependencies = [ "sha1", "sha2", "smallvec", + "std-next", "sync_wrapper 1.0.1", - "thiserror", + "thiserror 2.0.3", "time", "tokio", "tower 0.5.1", @@ -2378,12 +2379,12 @@ dependencies = [ [[package]] name = "s3s-policy" version = "0.11.0-dev" -source = "git+https://github.com/Nugine/s3s.git?rev=207170f526c75a8190e8f9afadf961909fd01d34#207170f526c75a8190e8f9afadf961909fd01d34" +source = "git+https://github.com/Nugine/s3s.git?rev=3291c0ca0284971569499cbe75bd69ef7bde8321#3291c0ca0284971569499cbe75bd69ef7bde8321" dependencies = [ "indexmap 2.6.0", "serde", "serde_json", - "thiserror", + "thiserror 2.0.3", ] [[package]] @@ -2560,6 +2561,16 @@ version = "1.2.0" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a8f112729512f8e442d81f95a8a7ddf2b7c6b8a1a6f509a95864142b30cab2d3" +[[package]] +name = "std-next" +version = "0.1.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "87854dde78837ff867561b22d873c53c25938d4b152afba2fa6fc48cb406c099" +dependencies = [ + "simdutf8", + "thiserror 2.0.3", +] + [[package]] name = "strsim" version = "0.11.1" @@ -2625,7 +2636,16 @@ version = "1.0.68" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "02dd99dc800bbb97186339685293e1cc5d9df1f8fae2d0aecd9ff1c77efea892" dependencies = [ - "thiserror-impl", + "thiserror-impl 1.0.68", +] + +[[package]] +name = "thiserror" +version = "2.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "c006c85c7651b3cf2ada4584faa36773bd07bac24acfb39f3c431b36d7e667aa" +dependencies = [ + "thiserror-impl 2.0.3", ] [[package]] @@ -2639,6 +2659,17 @@ dependencies = [ "syn", ] +[[package]] +name = "thiserror-impl" +version = "2.0.3" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "f077553d607adc1caf65430528a576c757a71ed73944b66ebb58ef2bbd243568" +dependencies = [ + "proc-macro2", + "quote", + "syn", +] + [[package]] name = "thread_local" version = "1.1.8" diff --git a/Cargo.toml b/Cargo.toml index 08539fe2c..f5f283948 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -10,7 +10,7 @@ members = [ "common/protos", "api/admin", "reader", - "common/workers", + "common/workers", ] [workspace.package] @@ -55,10 +55,10 @@ protos = { path = "./common/protos" } rand = "0.8.5" rmp = "0.8.14" rmp-serde = "1.3.0" -s3s = { git = "https://github.com/Nugine/s3s.git", rev = "207170f526c75a8190e8f9afadf961909fd01d34", default-features = true, features = [ +s3s = { git = "https://github.com/Nugine/s3s.git", rev = "3291c0ca0284971569499cbe75bd69ef7bde8321", default-features = true, features = [ "tower", ] } -s3s-policy = { git = "https://github.com/Nugine/s3s.git", rev = "207170f526c75a8190e8f9afadf961909fd01d34" } +s3s-policy = { git = "https://github.com/Nugine/s3s.git", rev = "3291c0ca0284971569499cbe75bd69ef7bde8321" } serde = { version = "1.0.214", features = ["derive"] } serde_json = "1.0.132" tempfile = "3.13.0" diff --git a/ecstore/src/bucket/versioning/mod.rs b/ecstore/src/bucket/versioning/mod.rs index 6ad2e4014..3444034e4 100644 --- a/ecstore/src/bucket/versioning/mod.rs +++ b/ecstore/src/bucket/versioning/mod.rs @@ -1,24 +1,25 @@ use s3s::dto::{BucketVersioningStatus, VersioningConfiguration}; +use crate::utils::wildcard; + pub trait VersioningApi { fn enabled(&self) -> bool; fn prefix_enabled(&self, prefix: &str) -> bool; fn prefix_suspended(&self, prefix: &str) -> bool; + fn versioned(&self, prefix: &str) -> bool; + fn suspended(&self) -> bool; } impl VersioningApi for VersioningConfiguration { fn enabled(&self) -> bool { - self.status - .as_ref() - .is_some_and(|v| v.as_str() == BucketVersioningStatus::ENABLED) + self.status == Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)) + } + fn suspended(&self) -> bool { + self.status == Some(BucketVersioningStatus::from_static(BucketVersioningStatus::SUSPENDED)) } fn prefix_enabled(&self, prefix: &str) -> bool { - if !self - .status - .as_ref() - .is_some_and(|v| v.as_str() == BucketVersioningStatus::ENABLED) - { + if self.status == Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)) { return false; } @@ -26,27 +27,56 @@ impl VersioningApi for VersioningConfiguration { return true; } - // TODO: ExcludeFolders + if let Some(exclude_folders) = self.exclude_folders { + if exclude_folders && prefix.ends_with('/') { + return false; + } + } + + if let Some(ref excluded_prefixes) = self.excluded_prefixes { + for p in excluded_prefixes.iter() { + if let Some(ref sprefix) = p.prefix { + let pattern = format!("{}*", sprefix); + if wildcard::match_simple(&pattern, prefix) { + return false; + } + } + } + } true } fn prefix_suspended(&self, prefix: &str) -> bool { - if self - .status - .as_ref() - .is_some_and(|v| v.as_str() == BucketVersioningStatus::SUSPENDED) - { + if self.status == Some(BucketVersioningStatus::from_static(BucketVersioningStatus::SUSPENDED)) { return true; } - if let Some(status) = self.status.as_ref() { - if status.as_str() == BucketVersioningStatus::ENABLED && prefix.is_empty() { + if self.status == Some(BucketVersioningStatus::from_static(BucketVersioningStatus::ENABLED)) { + if prefix.is_empty() { return false; + } - // TODO: ExcludeFolders + if let Some(exclude_folders) = self.exclude_folders { + if exclude_folders && prefix.ends_with('/') { + return true; + } + } + + if let Some(ref excluded_prefixes) = self.excluded_prefixes { + for p in excluded_prefixes.iter() { + if let Some(ref sprefix) = p.prefix { + let pattern = format!("{}*", sprefix); + if wildcard::match_simple(&pattern, prefix) { + return true; + } + } + } } } false } + fn versioned(&self, prefix: &str) -> bool { + self.prefix_enabled(prefix) || self.prefix_suspended(prefix) + } } diff --git a/ecstore/src/bucket/versioning_sys.rs b/ecstore/src/bucket/versioning_sys.rs index b464a1606..c62a829c9 100644 --- a/ecstore/src/bucket/versioning_sys.rs +++ b/ecstore/src/bucket/versioning_sys.rs @@ -7,6 +7,9 @@ use tracing::warn; pub struct BucketVersioningSys {} impl BucketVersioningSys { + pub fn new() -> Self { + Self {} + } pub async fn enabled(bucket: &str) -> bool { match Self::get(bucket).await { Ok(res) => res.enabled(), @@ -27,15 +30,15 @@ impl BucketVersioningSys { } } - // pub async fn suspended(bucket: &str) -> bool { - // match Self::get(bucket).await { - // Ok(res) => res.suspended(), - // Err(err) => { - // warn!("{:?}", err); - // false - // } - // } - // } + pub async fn suspended(bucket: &str) -> bool { + match Self::get(bucket).await { + Ok(res) => res.suspended(), + Err(err) => { + warn!("{:?}", err); + false + } + } + } pub async fn prefix_suspended(bucket: &str, prefix: &str) -> bool { match Self::get(bucket).await { diff --git a/ecstore/src/config/common.rs b/ecstore/src/config/common.rs index 7ce6ced42..6a47e4898 100644 --- a/ecstore/src/config/common.rs +++ b/ecstore/src/config/common.rs @@ -6,8 +6,7 @@ use super::{storageclass, Config, GLOBAL_StorageClass, KVS}; use crate::config::error::is_not_found; use crate::disk::RUSTFS_META_BUCKET; use crate::error::{Error, Result}; -use crate::store::ECStore; -use crate::store_api::{HTTPRangeSpec, ObjectIO, ObjectInfo, ObjectOptions, PutObjReader, StorageAPI}; +use crate::store_api::{HTTPRangeSpec, ObjectInfo, ObjectOptions, PutObjReader, StorageAPI}; use crate::store_err::is_err_object_not_found; use crate::utils::path::SLASH_SEPARATOR; use http::HeaderMap; @@ -30,13 +29,17 @@ lazy_static! { h }; } -pub async fn read_config(api: Arc, file: &str) -> Result> { +pub async fn read_config(api: Arc, file: &str) -> Result> { let (data, _obj) = read_config_with_metadata(api, file, &ObjectOptions::default()).await?; Ok(data) } -async fn read_config_with_metadata(api: Arc, file: &str, opts: &ObjectOptions) -> Result<(Vec, ObjectInfo)> { +async fn read_config_with_metadata( + api: Arc, + file: &str, + opts: &ObjectOptions, +) -> Result<(Vec, ObjectInfo)> { let range = HTTPRangeSpec::nil(); let h = HeaderMap::new(); let mut rd = api @@ -59,7 +62,7 @@ async fn read_config_with_metadata(api: Arc, file: &str, opts: &ObjectO Ok((data, rd.object_info)) } -pub async fn save_config(api: Arc, file: &str, data: &[u8]) -> Result<()> { +pub async fn save_config(api: Arc, file: &str, data: &[u8]) -> Result<()> { save_config_with_opts( api, file, @@ -72,7 +75,7 @@ pub async fn save_config(api: Arc, file: &str, data: &[u8]) -> Result<( .await } -async fn save_config_with_opts(api: Arc, file: &str, data: &[u8], opts: &ObjectOptions) -> Result<()> { +async fn save_config_with_opts(api: Arc, file: &str, data: &[u8], opts: &ObjectOptions) -> Result<()> { let _ = api .put_object( RUSTFS_META_BUCKET, @@ -88,7 +91,7 @@ fn new_server_config() -> Config { Config::new() } -async fn new_and_save_server_config(api: Arc) -> Result { +async fn new_and_save_server_config(api: Arc) -> Result { let mut cfg = new_server_config(); lookup_configs(&mut cfg, api.clone()).await; save_server_config(api, &cfg).await?; @@ -96,7 +99,7 @@ async fn new_and_save_server_config(api: Arc) -> Result { Ok(cfg) } -pub async fn read_config_without_migrate(api: Arc) -> Result { +pub async fn read_config_without_migrate(api: Arc) -> Result { let config_file = format!("{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, CONFIG_FILE); let data = match read_config(api.clone(), config_file.as_str()).await { Ok(res) => res, @@ -116,7 +119,7 @@ pub async fn read_config_without_migrate(api: Arc) -> Result { read_server_config(api, data.as_slice()).await } -async fn read_server_config(api: Arc, data: &[u8]) -> Result { +async fn read_server_config(api: Arc, data: &[u8]) -> Result { let cfg = { if data.is_empty() { let config_file = format!("{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, CONFIG_FILE); @@ -145,7 +148,7 @@ async fn read_server_config(api: Arc, data: &[u8]) -> Result { Ok(cfg.merge()) } -async fn save_server_config(api: Arc, cfg: &Config) -> Result<()> { +async fn save_server_config(api: Arc, cfg: &Config) -> Result<()> { let data = cfg.marshal()?; let config_file = format!("{}{}{}", CONFIG_PREFIX, SLASH_SEPARATOR, CONFIG_FILE); @@ -153,14 +156,14 @@ async fn save_server_config(api: Arc, cfg: &Config) -> Result<()> { save_config(api, &config_file, data.as_slice()).await } -pub async fn lookup_configs(cfg: &mut Config, api: Arc) { +pub async fn lookup_configs(cfg: &mut Config, api: Arc) { // TODO: from etcd if let Err(err) = apply_dynamic_config(cfg, api).await { error!("apply_dynamic_config err {:?}", &err); } } -async fn apply_dynamic_config(cfg: &mut Config, api: Arc) -> Result<()> { +async fn apply_dynamic_config(cfg: &mut Config, api: Arc) -> Result<()> { for key in SubSystemsDynamic.iter() { apply_dynamic_config_for_sub_sys(cfg, api.clone(), key).await?; } @@ -168,7 +171,7 @@ async fn apply_dynamic_config(cfg: &mut Config, api: Arc) -> Result<()> Ok(()) } -async fn apply_dynamic_config_for_sub_sys(cfg: &mut Config, api: Arc, subsys: &str) -> Result<()> { +async fn apply_dynamic_config_for_sub_sys(cfg: &mut Config, api: Arc, subsys: &str) -> Result<()> { let set_drive_counts = api.set_drive_counts(); if subsys == STORAGE_CLASS_SUB_SYS { let kvs = match cfg.get_value(STORAGE_CLASS_SUB_SYS, DEFAULT_KV_KEY) { diff --git a/ecstore/src/pools.rs b/ecstore/src/pools.rs index a7f95a49d..38f928e30 100644 --- a/ecstore/src/pools.rs +++ b/ecstore/src/pools.rs @@ -1,16 +1,22 @@ -use crate::config::common::{read_config, save_config}; +use crate::bucket::versioning_sys::BucketVersioningSys; +use crate::config::common::{read_config, save_config, CONFIG_PREFIX}; use crate::config::error::ConfigError; +use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET}; use crate::error::{Error, Result}; +use crate::heal::heal_commands::HealOpts; use crate::new_object_layer_fn; -use crate::store_api::{StorageAPI, StorageDisk, StorageInfo}; -use crate::store_err::StorageError; +use crate::store_api::{BucketOptions, MakeBucketOptions, StorageAPI, StorageDisk, StorageInfo}; +use crate::store_err::{is_err_bucket_exists, StorageError}; +use crate::utils::path::{path_join, SLASH_SEPARATOR}; use crate::{sets::Sets, store::ECStore}; use byteorder::{ByteOrder, LittleEndian, WriteBytesExt}; use rmp_serde::{Deserializer, Serializer}; use serde::{Deserialize, Serialize}; use std::io::{Cursor, Write}; +use std::path::PathBuf; use std::sync::Arc; use time::OffsetDateTime; +use tracing::{error, info, warn}; pub const POOL_META_NAME: &str = "pool.bin"; pub const POOL_META_FORMAT: u16 = 1; @@ -96,7 +102,7 @@ impl PoolMeta { Ok(()) } - pub async fn save(&self, _pools: Vec>) -> Result<()> { + pub async fn save(&self, pools: Vec>) -> Result<()> { if self.dont_save { return Ok(()); } @@ -107,12 +113,11 @@ impl PoolMeta { self.serialize(&mut Serializer::new(&mut buf))?; data.write_all(&buf)?; - let Some(store) = new_object_layer_fn() else { - return Err(Error::from_string("errServerNotInitialized".to_string())); - }; + for pool in pools { + save_config(pool, &POOL_META_NAME, &data).await?; + } - // FIXME: - save_config(store, &POOL_META_NAME, &data).await + Ok(()) } pub fn decommission_cancel(&mut self, idx: usize) -> bool { @@ -139,6 +144,150 @@ 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 { + if !info.complete && !info.failed && !info.canceled { + return Err(Error::msg("DecommissionAlreadyRunning")); + } + } + + let now = OffsetDateTime::now_utc(); + pool.last_update = now.clone(); + 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) { + 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)] @@ -162,6 +311,56 @@ pub struct PoolDecommissionInfo { 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.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)] pub struct PoolSpaceInfo { pub free: usize, @@ -169,6 +368,20 @@ pub struct PoolSpaceInfo { pub used: usize, } +#[derive(Debug, Default, Clone)] +pub struct DecomBucketInfo { + pub name: String, + pub prefix: String, +} + +impl ToString for DecomBucketInfo { + fn to_string(&self) -> String { + path_join(&[PathBuf::from(self.name.clone()), PathBuf::from(self.prefix.clone())]) + .to_string_lossy() + .to_string() + } +} + impl ECStore { pub async fn status(&self, idx: usize) -> Result { let space_info = self.get_decommission_pool_space_info(idx).await?; @@ -228,6 +441,235 @@ impl ECStore { 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 { + if !info.complete && !info.failed && !info.canceled { + return true; + } + } + } + + false + } + + pub async fn decommission(&self, indices: Vec) -> Result<()> { + if indices.is_empty() { + return Err(Error::msg("errInvalidArgument")); + } + + if self.single_pool() { + return Err(Error::msg("errInvalidArgument")); + } + + self.start_decommission(indices.clone()).await?; + + 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(idx.clone()).await; + } + }); + + Ok(()) + } + + async fn decommission_pool(&self, _idx: usize, _pool: Arc, bi: DecomBucketInfo) -> Result<()> { + let mut _vc = None; + + if &bi.name == RUSTFS_META_BUCKET { + let versioning = BucketVersioningSys::get(&bi.name).await?; + _vc = Some(versioning); + } + + // 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")); + } + + if self.single_pool() { + return Err(Error::msg("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 = vec![ + 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 + { + if !is_err_bucket_exists(&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.clone()).await?; + + pool_meta.decommission(idx.clone(), pi)?; + + pool_meta.queue_buckets(idx.clone(), decom_buckets.clone()); + } + + pool_meta.save(self.pools.clone()).await?; + + // FIXME: globalNotificationSys.ReloadPoolMeta(ctx) + + 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) + } } fn get_total_usable_capacity(disks: &Vec, info: &StorageInfo) -> usize { diff --git a/rustfs/src/admin/handlers.rs b/rustfs/src/admin/handlers.rs index a72a45b36..988a7dc31 100644 --- a/rustfs/src/admin/handlers.rs +++ b/rustfs/src/admin/handlers.rs @@ -308,7 +308,7 @@ impl Operation for StatusPool { let query = { if let Some(query) = req.uri.query() { let input: StatusPoolQuery = - from_bytes(query.as_bytes()).map_err(|_e| s3_error!(InvalidRequest, "get body failed"))?; + from_bytes(query.as_bytes()).map_err(|_e| s3_error!(InvalidArgument, "get body failed"))?; input } else { StatusPoolQuery::default() @@ -353,10 +353,77 @@ pub struct StartDecommission {} #[async_trait::async_trait] impl Operation for StartDecommission { // POST //pools/decommission?pool=http://server{1...4}/disk{1...4} - async fn call(&self, _req: S3Request, _params: Params<'_, '_>) -> S3Result> { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { warn!("handle StartDecommission"); - return Err(s3_error!(NotImplemented)); + let Some(endpoints) = GLOBAL_Endpoints.get() else { + return Err(s3_error!(NotImplemented)); + }; + + if endpoints.legacy() { + return Err(s3_error!(NotImplemented)); + } + + let Some(store) = new_object_layer_fn() else { + return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string())); + }; + + if store.is_decommission_running().await { + return Err(S3Error::with_message( + S3ErrorCode::InvalidRequest, + "DecommissionAlreadyRunning".to_string(), + )); + } + + // TODO: check IsRebalanceStarted + + let query = { + if let Some(query) = req.uri.query() { + let input: StatusPoolQuery = + from_bytes(query.as_bytes()).map_err(|_e| s3_error!(InvalidArgument, "get body failed"))?; + input + } else { + StatusPoolQuery::default() + } + }; + let is_byid = query.by_id.as_str() == "true"; + + let pools: Vec<&str> = query.pool.split(",").collect(); + let mut pools_indices = Vec::with_capacity(pools.len()); + + for pool in pools.iter() { + let idx = { + if is_byid { + pool.parse::() + .map_err(|_e| s3_error!(InvalidArgument, "pool parse failed"))? + } else { + let Some(idx) = endpoints.get_pool_idx(pool) else { + return Err(s3_error!(InvalidArgument, "pool parse failed")); + }; + idx + } + }; + + let mut has_found = None; + for (i, pool) in store.pools.iter().enumerate() { + if i == idx { + has_found = Some(pool.clone()); + break; + } + } + + let Some(_p) = has_found else { + return Err(s3_error!(InvalidArgument)); + }; + + pools_indices.push(idx); + } + + if !pools_indices.is_empty() { + store.decommission(pools_indices).await.map_err(to_s3_error)?; + } + + Ok(S3Response::new((StatusCode::OK, Body::default()))) } } @@ -379,7 +446,7 @@ impl Operation for CancelDecommission { let query = { if let Some(query) = req.uri.query() { let input: StatusPoolQuery = - from_bytes(query.as_bytes()).map_err(|_e| s3_error!(InvalidRequest, "get body failed"))?; + from_bytes(query.as_bytes()).map_err(|_e| s3_error!(InvalidArgument, "get body failed"))?; input } else { StatusPoolQuery::default()