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..0e06a0e3f 100644 --- a/ecstore/src/pools.rs +++ b/ecstore/src/pools.rs @@ -1,16 +1,21 @@ -use crate::config::common::{read_config, save_config}; +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; 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; pub const POOL_META_NAME: &str = "pool.bin"; pub const POOL_META_FORMAT: u16 = 1; @@ -96,7 +101,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 +112,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 +143,34 @@ impl PoolMeta { 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); + } + } + } } #[derive(Debug, Clone, Serialize, Deserialize, Default)] @@ -162,6 +194,33 @@ 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 + } +} + #[derive(Debug)] pub struct PoolSpaceInfo { pub free: usize, @@ -169,6 +228,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 +301,117 @@ 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 do_decommission_in_routine(&self, _idx: usize) { + // FIXME: + unimplemented!() + } + + 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()