diff --git a/crates/ecstore/src/store.rs b/crates/ecstore/src/store.rs index e480b229a..ed8ad8301 100644 --- a/crates/ecstore/src/store.rs +++ b/crates/ecstore/src/store.rs @@ -131,6 +131,21 @@ async fn has_xlmeta_files(path: &std::path::Path) -> bool { const MAX_UPLOADS_LIST: usize = 10000; +mod bucket; +mod heal; +mod init; +mod list; +mod multipart; +mod object; +mod peer; +mod rebalance; + +use peer::init_local_peer; +pub use peer::{ + all_local_disk, all_local_disk_path, find_local_disk, get_disk_infos, get_disk_via_endpoint, has_space_for, init_local_disks, + init_lock_clients, +}; + #[derive(Debug)] pub struct ECStore { pub id: Uuid, @@ -161,1025 +176,8 @@ pub struct ECStore { // } // } -impl ECStore { - #[allow(clippy::new_ret_no_self)] - #[instrument(level = "debug", skip(endpoint_pools))] - pub async fn new(address: SocketAddr, endpoint_pools: EndpointServerPools, ctx: CancellationToken) -> Result> { - // let layouts = DisksLayout::from_volumes(endpoints.as_slice())?; - - let mut deployment_id = None; - - // let (endpoint_pools, _) = EndpointServerPools::create_server_endpoints(address.as_str(), &layouts)?; - - let mut pools = Vec::with_capacity(endpoint_pools.as_ref().len()); - let mut disk_map = HashMap::with_capacity(endpoint_pools.as_ref().len()); - - let first_is_local = endpoint_pools.first_local(); - - let mut local_disks = Vec::new(); - - info!("ECStore new address: {}", address.to_string()); - let mut host = address.ip().to_string(); - if host.is_empty() { - host = GLOBAL_RUSTFS_HOST.read().await.to_string() - } - let mut port = address.port().to_string(); - if port.is_empty() { - port = GLOBAL_RUSTFS_PORT.read().await.to_string() - } - info!("ECStore new host: {}, port: {}", host, port); - init_local_peer(&endpoint_pools, &host, &port).await; - - // debug!("endpoint_pools: {:?}", endpoint_pools); - - let mut common_parity_drives = 0; - - for (i, pool_eps) in endpoint_pools.as_ref().iter().enumerate() { - if common_parity_drives == 0 { - let parity_drives = ec_drives_no_config(pool_eps.drives_per_set)?; - storageclass::validate_parity(parity_drives, pool_eps.drives_per_set)?; - common_parity_drives = parity_drives; - } - - // validate_parity(parity_count, pool_eps.drives_per_set)?; - - // Initialize disks without health monitoring so that remote peers - // are not immediately marked as faulty before they have a chance to - // start up. Health monitoring is enabled after format loading succeeds. - let (disks, errs) = store_init::init_disks( - &pool_eps.endpoints, - &DiskOption { - cleanup: true, - health_check: false, - }, - ) - .await; - - check_disk_fatal_errs(&errs)?; - - let fm = { - let mut times = 0; - let mut interval = 1; - loop { - match store_init::connect_load_init_formats( - first_is_local, - &disks, - pool_eps.set_count, - pool_eps.drives_per_set, - deployment_id, - ) - .await - { - Ok(fm) => break Ok(fm), - // Wrap the final error if we are giving up - Err(e) if times >= 10 => { - break Err(Error::other(format!("can not get formats after {} retries, last error: {e}", times))); - } - // Retrying so just drop the error - Err(_) => {} - } - times += 1; - if interval < 16 { - interval *= 2; - } - info!("retrying get formats after {:?}", interval); - select! { - _ = tokio::signal::ctrl_c() => { - info!("got ctrl+c, exits"); - exit(0); - } - _ = sleep(Duration::from_secs(interval)) => { - } - } - } - }?; - - // Format loading succeeded, enable health monitoring on all disks - for disk in disks.iter().flatten() { - disk.enable_health_check(); - } - - if deployment_id.is_none() { - deployment_id = Some(fm.id); - } - - if deployment_id != Some(fm.id) { - return Err(Error::other("deployment_id not same in one pool")); - } - - if deployment_id.is_some() && deployment_id.unwrap().is_nil() { - deployment_id = Some(Uuid::new_v4()); - } - - for disk in disks.iter() { - if disk.is_some() && disk.as_ref().unwrap().is_local() { - local_disks.push(disk.as_ref().unwrap().clone()); - } - } - - let sets = Sets::new(disks.clone(), pool_eps, &fm, i, common_parity_drives).await?; - pools.push(sets); - - disk_map.insert(i, disks); - } - - // Replace the local disk - if !is_dist_erasure().await { - let mut global_local_disk_map = GLOBAL_LOCAL_DISK_MAP.write().await; - for disk in local_disks { - let path = disk.endpoint().to_string(); - global_local_disk_map.insert(path, Some(disk.clone())); - } - } - - let peer_sys = S3PeerSys::new(&endpoint_pools); - let mut pool_meta = PoolMeta::new(&pools, &PoolMeta::default()); - pool_meta.dont_save = true; - - let decommission_cancelers = vec![None; pools.len()]; - let ec = Arc::new(ECStore { - id: deployment_id.unwrap(), - disk_map, - pools, - peer_sys, - pool_meta: RwLock::new(pool_meta), - rebalance_meta: RwLock::new(None), - decommission_cancelers, - }); - - // Only set it when the global deployment ID is not yet configured - if let Some(dep_id) = deployment_id - && get_global_deployment_id().is_none() - { - set_global_deployment_id(dep_id); - } - - let wait_sec = 5; - let mut exit_count = 0; - loop { - if let Err(err) = ec.init(ctx.clone()).await { - error!("init err: {}", err); - error!("retry after {} second", wait_sec); - sleep(Duration::from_secs(wait_sec)).await; - - if exit_count > 10 { - return Err(Error::other("ec init failed")); - } - - exit_count += 1; - - continue; - } - - break; - } - - set_object_layer(ec.clone()).await; - - Ok(ec) - } - - #[instrument(level = "debug", skip(self, rx))] - pub async fn init(self: &Arc, rx: CancellationToken) -> Result<()> { - GLOBAL_BOOT_TIME.get_or_init(|| async { SystemTime::now() }).await; - - if self.load_rebalance_meta().await.is_ok() { - self.start_rebalance().await; - } - - let mut meta = PoolMeta::default(); - meta.load(self.pools[0].clone(), self.pools.clone()).await?; - let update = meta.validate(self.pools.clone())?; - - if !update { - { - let mut pool_meta = self.pool_meta.write().await; - *pool_meta = meta.clone(); - } - } else { - let new_meta = PoolMeta::new(&self.pools, &meta); - new_meta.save(self.pools.clone()).await?; - { - let mut pool_meta = self.pool_meta.write().await; - *pool_meta = new_meta; - } - } - - let pools = meta.return_resumable_pools(); - let mut pool_indices = Vec::with_capacity(pools.len()); - - let endpoints = get_global_endpoints(); - - for p in pools.iter() { - if let Some(idx) = endpoints.get_pool_idx(&p.cmd_line) { - pool_indices.push(idx); - } else { - return Err(Error::other(format!( - "unexpected state present for decommission status pool({}) not found", - p.cmd_line - ))); - } - } - - if !pool_indices.is_empty() { - let idx = pool_indices[0]; - if endpoints.as_ref()[idx].endpoints.as_ref()[0].is_local { - let store = self.clone(); - - tokio::spawn(async move { - // wait 3 minutes for cluster init - tokio::time::sleep(Duration::from_secs(60 * 3)).await; - - if let Err(err) = store.decommission(rx.clone(), pool_indices.clone()).await { - if err == StorageError::DecommissionAlreadyRunning { - for i in pool_indices.iter() { - store.do_decommission_in_routine(rx.clone(), *i).await; - } - return; - } - - error!("store init decommission err: {}", err); - - // TODO: check config err - } - }); - } - } - - let num_nodes = get_global_endpoints().get_nodes().len() as u64; - init_global_bucket_monitor(num_nodes); - - init_background_expiry(self.clone()).await; - - TransitionState::init(self.clone()).await; - - if let Err(err) = GLOBAL_TierConfigMgr.write().await.init(self.clone()).await { - info!("TierConfigMgr init error: {}", err); - } - - Ok(()) - } - - pub fn init_local_disks() {} - - // pub fn local_disks(&self) -> Vec { - // self.local_disks.clone() - // } - - pub fn single_pool(&self) -> bool { - self.pools.len() == 1 - } - - // define in store_list_objects.rs - // pub async fn list_path(&self, opts: &ListPathOptions, delimiter: &str) -> Result { - // // if opts.prefix.ends_with(SLASH_SEPARATOR) { - // // return Err(Error::other("eof")); - // // } - - // let mut opts = opts.clone(); - - // if opts.base_dir.is_empty() { - // opts.base_dir = base_dir_from_prefix(&opts.prefix); - // } - - // let objects = self.list_merged(&opts, delimiter).await?; - - // let info = ListObjectsInfo { - // objects, - // ..Default::default() - // }; - // Ok(info) - // } - - // Read all entries - // define in store_list_objects.rs - // async fn list_merged(&self, opts: &ListPathOptions, delimiter: &str) -> Result> { - // let walk_opts = WalkDirOptions { - // bucket: opts.bucket.clone(), - // base_dir: opts.base_dir.clone(), - // ..Default::default() - // }; - - // // let (mut wr, mut rd) = tokio::io::duplex(1024); - - // let mut futures = Vec::new(); - - // for sets in self.pools.iter() { - // for set in sets.disk_set.iter() { - // futures.push(set.walk_dir(&walk_opts)); - // } - // } - - // let results = join_all(futures).await; - - // // let mut errs = Vec::new(); - // let mut ress = Vec::new(); - // let mut uniq = HashSet::new(); - - // for (disks_ress, _disks_errs) in results { - // for disks_res in disks_ress.iter() { - // if disks_res.is_none() { - // // TODO handle errs - // continue; - // } - // let entries = disks_res.as_ref().unwrap(); - - // for entry in entries { - // // warn!("lst_merged entry---- {}", &entry.name); - - // if !opts.prefix.is_empty() && !entry.name.starts_with(&opts.prefix) { - // continue; - // } - - // if !uniq.contains(&entry.name) { - // uniq.insert(entry.name.clone()); - // // TODO: filter - - // if opts.limit > 0 && ress.len() as i32 >= opts.limit { - // return Ok(ress); - // } - - // if entry.is_object() { - // if !delimiter.is_empty() { - // // entry.name.trim_start_matches(pat) - // } - - // let fi = entry.to_fileinfo(&opts.bucket)?; - // if let Some(f) = fi { - // ress.push(f.to_object_info(&opts.bucket, &entry.name, false)); - // } - // continue; - // } - - // if entry.is_dir() { - // ress.push(ObjectInfo { - // is_dir: true, - // bucket: opts.bucket.clone(), - // name: entry.name.clone(), - // ..Default::default() - // }); - // } - // } - // } - // } - // } - - // // warn!("list_merged errs {:?}", errs); - - // Ok(ress) - // } - - #[instrument(level = "debug", skip(self))] - async fn delete_all(&self, bucket: &str, prefix: &str) -> Result<()> { - let mut futures = Vec::new(); - for sets in self.pools.iter() { - for set in sets.disk_set.iter() { - futures.push(set.delete_all(bucket, prefix)); - // let disks = set.disks.read().await; - // let dd = disks.clone(); - // for disk in dd { - // if disk.is_none() { - // continue; - // } - // // let disk = disk.as_ref().unwrap().clone(); - // // futures.push(disk.delete( - // // bucket, - // // prefix, - // // DeleteOptions { - // // recursive: true, - // // immediate: false, - // // }, - // // )); - // } - } - } - let results = join_all(futures).await; - - let mut errs = Vec::new(); - - for res in results { - match res { - Ok(_) => errs.push(None), - Err(e) => errs.push(Some(e)), - } - } - - debug!("store delete_all errs {:?}", errs); - - Ok(()) - } - async fn delete_prefix(&self, bucket: &str, object: &str) -> Result<()> { - for pool in self.pools.iter() { - pool.delete_object( - bucket, - object, - ObjectOptions { - delete_prefix: true, - ..Default::default() - }, - ) - .await?; - } - - Ok(()) - } - - async fn get_available_pool_idx(&self, bucket: &str, object: &str, size: i64) -> Option { - // // Return a random one first - - let mut server_pools = self.get_server_pools_available_space(bucket, object, size).await; - server_pools.filter_max_used(100 - (100_f64 * DISK_RESERVE_FRACTION) as u64); - let total = server_pools.total_available(); - - if total == 0 { - return None; - } - - let mut rng = rand::rng(); - let random_u64: u64 = rng.random_range(0..total); - - let choose = random_u64 % total; - let mut at_total = 0; - - for pool in server_pools.iter() { - at_total += pool.available; - if at_total > choose && pool.available > 0 { - return Some(pool.index); - } - } - - None - } - - async fn get_server_pools_available_space(&self, bucket: &str, object: &str, size: i64) -> ServerPoolsAvailableSpace { - let mut n_sets = vec![0; self.pools.len()]; - let mut infos = vec![Vec::new(); self.pools.len()]; - - // TODO: add concurrency - for (idx, pool) in self.pools.iter().enumerate() { - if self.is_suspended(idx).await || self.is_pool_rebalancing(idx).await { - continue; - } - - n_sets[idx] = pool.set_count; - - if let Ok(disks) = pool.get_disks_by_key(object).get_disks(0, 0).await { - let disk_infos = get_disk_infos(&disks).await; - infos[idx] = disk_infos; - } - } - - let mut server_pools = vec![PoolAvailableSpace::default(); self.pools.len()]; - for (i, zinfo) in infos.iter().enumerate() { - if zinfo.is_empty() { - server_pools[i] = PoolAvailableSpace { - index: i, - ..Default::default() - }; - - continue; - } - - if !is_meta_bucketname(bucket) && !has_space_for(zinfo, size).await.unwrap_or_default() { - server_pools[i] = PoolAvailableSpace { - index: i, - ..Default::default() - }; - - continue; - } - - let mut available = 0; - let mut max_used_pct = 0; - for disk in zinfo.iter().flatten() { - if disk.total == 0 { - continue; - } - - available += disk.total - disk.used; - - let pct_used = disk.used * 100 / disk.total; - - if pct_used > max_used_pct { - max_used_pct = pct_used; - } - } - - available *= n_sets[i] as u64; - - server_pools[i] = PoolAvailableSpace { - index: i, - available, - max_used_pct, - } - } - - ServerPoolsAvailableSpace(server_pools) - } - - async fn is_suspended(&self, idx: usize) -> bool { - // TODO: LOCK - - let pool_meta = self.pool_meta.read().await; - - pool_meta.is_suspended(idx) - } - - async fn get_pool_idx(&self, bucket: &str, object: &str, size: i64) -> Result { - let idx = match self - .get_pool_idx_existing_with_opts( - bucket, - object, - &ObjectOptions { - skip_decommissioned: true, - skip_rebalancing: true, - ..Default::default() - }, - ) - .await - { - Ok(res) => res, - Err(err) => { - if !is_err_object_not_found(&err) { - return Err(err); - } - - if let Some(hit_idx) = self.get_available_pool_idx(bucket, object, size).await { - hit_idx - } else { - return Err(Error::DiskFull); - } - } - }; - - Ok(idx) - } - - async fn get_pool_idx_no_lock(&self, bucket: &str, object: &str, size: i64) -> Result { - let idx = match self.get_pool_idx_existing_no_lock(bucket, object).await { - Ok(res) => res, - Err(err) => { - if !is_err_object_not_found(&err) { - return Err(err); - } - - if let Some(idx) = self.get_available_pool_idx(bucket, object, size).await { - idx - } else { - warn!("get_pool_idx_no_lock: disk full {}/{}", bucket, object); - return Err(Error::DiskFull); - } - } - }; - - Ok(idx) - } - - async fn get_pool_idx_existing_no_lock(&self, bucket: &str, object: &str) -> Result { - self.get_pool_idx_existing_with_opts( - bucket, - object, - &ObjectOptions { - no_lock: true, - skip_decommissioned: true, - skip_rebalancing: true, - ..Default::default() - }, - ) - .await - } - - async fn get_pool_idx_existing_with_opts(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { - let (pinfo, _) = self.get_pool_info_existing_with_opts(bucket, object, opts).await?; - Ok(pinfo.index) - } - async fn get_pool_info_existing_with_opts( - &self, - bucket: &str, - object: &str, - opts: &ObjectOptions, - ) -> Result<(PoolObjInfo, Vec)> { - self.internal_get_pool_info_existing_with_opts(bucket, object, opts).await - } - - async fn internal_get_pool_info_existing_with_opts( - &self, - bucket: &str, - object: &str, - opts: &ObjectOptions, - ) -> Result<(PoolObjInfo, Vec)> { - let mut futures = Vec::new(); - for pool in self.pools.iter() { - let mut pool_opts = opts.clone(); - if !pool_opts.metadata_chg { - pool_opts.version_id = None; - } - - futures.push(async move { pool.get_object_info(bucket, object, &pool_opts).await }); - } - - let results = join_all(futures).await; - - let mut ress = Vec::new(); - - // join_all preserves the input order - for (i, res) in results.into_iter().enumerate() { - let index = i; - - match res { - Ok(r) => { - ress.push(PoolObjInfo { - index, - object_info: r, - err: None, - }); - } - Err(e) => { - ress.push(PoolObjInfo { - index, - err: Some(e), - ..Default::default() - }); - } - } - } - - ress.sort_by(|a, b| { - let at = a.object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH); - let bt = b.object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH); - - bt.cmp(&at) - }); - - let mut def_pool = PoolObjInfo::default(); - let mut has_def_pool = false; - - for pinfo in ress.iter() { - if opts.skip_decommissioned && self.is_suspended(pinfo.index).await { - continue; - } - - if opts.skip_rebalancing && self.is_pool_rebalancing(pinfo.index).await { - continue; - } - - if pinfo.err.is_none() { - return Ok((pinfo.clone(), self.pools_with_object(&ress, opts).await)); - } - - let err = pinfo.err.as_ref().unwrap(); - - if err == &Error::ErasureReadQuorum && !opts.metadata_chg { - return Ok((pinfo.clone(), self.pools_with_object(&ress, opts).await)); - } - - def_pool = pinfo.clone(); - has_def_pool = true; - // https://docs.aws.amazon.com/AmazonS3/latest/userguide/conditional-deletes.html - if is_err_object_not_found(err) - && let Err(err) = opts.precondition_check(&pinfo.object_info) - { - return Err(err.clone()); - } - - if !is_err_object_not_found(err) && !is_err_version_not_found(err) { - return Err(err.clone()); - } - - if pinfo.object_info.delete_marker && !pinfo.object_info.name.is_empty() { - return Ok((pinfo.clone(), Vec::new())); - } - } - - if opts.replication_request && opts.delete_marker && has_def_pool { - return Ok((def_pool, Vec::new())); - } - - Err(Error::ObjectNotFound(bucket.to_owned(), object.to_owned())) - } - - async fn pools_with_object(&self, pools: &[PoolObjInfo], opts: &ObjectOptions) -> Vec { - let mut errs = Vec::new(); - - for pool in pools.iter() { - if opts.skip_decommissioned && self.is_suspended(pool.index).await { - continue; - } - - if opts.skip_rebalancing && self.is_pool_rebalancing(pool.index).await { - continue; - } - - if let Some(err) = &pool.err { - if err == &Error::ErasureReadQuorum { - errs.push(PoolErr { - index: Some(pool.index), - err: Some(Error::ErasureReadQuorum), - }); - } - } else { - errs.push(PoolErr { - index: Some(pool.index), - err: None, - }); - } - } - errs - } - - async fn get_latest_object_info_with_idx( - &self, - bucket: &str, - object: &str, - opts: &ObjectOptions, - ) -> Result<(ObjectInfo, usize)> { - let mut futures = Vec::with_capacity(self.pools.len()); - for pool in self.pools.iter() { - futures.push(pool.get_object_info(bucket, object, opts)); - } - - let results = join_all(futures).await; - - struct IndexRes { - res: Option, - idx: usize, - err: Option, - } - - let mut idx_res = Vec::with_capacity(self.pools.len()); - - for (idx, result) in results.into_iter().enumerate() { - match result { - Ok(res) => { - idx_res.push(IndexRes { - res: Some(res), - idx, - err: None, - }); - } - Err(e) => { - idx_res.push(IndexRes { - res: None, - idx, - err: Some(e), - }); - } - } - } - - // TODO: test order - idx_res.sort_by(|a, b| { - let a_mod = if let Some(o1) = &a.res { - o1.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH) - } else { - OffsetDateTime::UNIX_EPOCH - }; - - let b_mod = if let Some(o2) = &b.res { - o2.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH) - } else { - OffsetDateTime::UNIX_EPOCH - }; - - if a_mod == b_mod { - return if a.idx < b.idx { Ordering::Greater } else { Ordering::Less }; - } - - b_mod.cmp(&a_mod) - }); - - for res in idx_res.into_iter() { - if let Some(obj) = res.res { - return Ok((obj, res.idx)); - } - - if let Some(err) = res.err - && !is_err_object_not_found(&err) - && !is_err_version_not_found(&err) - { - return Err(err); - } - - // TODO: delete marker - } - - let object = decode_dir_object(object); - - if opts.version_id.is_none() { - Err(StorageError::ObjectNotFound(bucket.to_owned(), object.to_owned())) - } else { - Err(StorageError::VersionNotFound( - bucket.to_owned(), - object.to_owned(), - opts.version_id.clone().unwrap_or_default(), - )) - } - } - - async fn delete_object_from_all_pools( - &self, - bucket: &str, - object: &str, - opts: &ObjectOptions, - errs: Vec, - ) -> Result { - let mut objs = Vec::new(); - let mut derrs = Vec::new(); - - for pe in errs.iter() { - if let Some(err) = &pe.err - && err == &StorageError::ErasureWriteQuorum - { - objs.push(None); - derrs.push(Some(StorageError::ErasureWriteQuorum)); - continue; - } - - if let Some(idx) = pe.index { - match self.pools[idx].delete_object(bucket, object, opts.clone()).await { - Ok(res) => { - objs.push(Some(res)); - - derrs.push(None); - } - Err(err) => { - objs.push(None); - derrs.push(Some(err)); - } - } - } - } - - if let Some(e) = &derrs[0] { - return Err(e.clone()); - } - - Ok(objs[0].as_ref().unwrap().clone()) - } - - pub async fn reload_pool_meta(&self) -> Result<()> { - let mut meta = PoolMeta::default(); - meta.load(self.pools[0].clone(), self.pools.clone()).await?; - - let mut pool_meta = self.pool_meta.write().await; - *pool_meta = meta; - // *self.pool_meta.write().unwrap() = meta; - Ok(()) - } - - /// Disk information deduplication function - /// - /// Use multiple field combinations to ensure uniqueness: - /// - endpoint (node address) - /// - drive_path (mount path) - /// - pool_index (pool index) - /// - set_index (Collection Index) - /// - disk_index (disk index) - pub(crate) fn deduplicate_disks(disks: Vec) -> Vec { - use std::collections::HashMap; - - let mut unique_disks: HashMap = HashMap::new(); - let mut duplicate_count = 0; - - for disk in disks { - // Generate a compound unique key - let key = format!( - "{}|{}|p{}s{}d{}", - disk.endpoint, disk.drive_path, disk.pool_index, disk.set_index, disk.disk_index - ); - - // Use the entry API to avoid duplicate inserts - use std::collections::hash_map::Entry; - match unique_disks.entry(key) { - Entry::Vacant(e) => { - e.insert(disk); - } - Entry::Occupied(_) => { - duplicate_count += 1; - } - } - } - - if duplicate_count > 0 { - debug!("Deduplicated {} duplicate disk entries", duplicate_count); - } - - unique_disks.into_values().collect() - } -} - -pub async fn find_local_disk(disk_path: &String) -> Option { - let disk_map = GLOBAL_LOCAL_DISK_MAP.read().await; - - if let Some(disk) = disk_map.get(disk_path) { - disk.as_ref().cloned() - } else { - None - } -} - -pub async fn get_disk_via_endpoint(endpoint: &Endpoint) -> Option { - let global_set_drives = GLOBAL_LOCAL_DISK_SET_DRIVES.read().await; - if global_set_drives.is_empty() { - return GLOBAL_LOCAL_DISK_MAP.read().await[&endpoint.to_string()].clone(); - } - global_set_drives[endpoint.pool_idx as usize][endpoint.set_idx as usize][endpoint.disk_idx as usize].clone() -} - -pub async fn all_local_disk_path() -> Vec { - let disk_map = GLOBAL_LOCAL_DISK_MAP.read().await; - disk_map.keys().cloned().collect() -} - -pub async fn all_local_disk() -> Vec { - let disk_map = GLOBAL_LOCAL_DISK_MAP.read().await; - disk_map - .values() - .filter(|v| v.is_some()) - .map(|v| v.as_ref().unwrap().clone()) - .collect() -} - // init_local_disks must succeed before the server starts -pub async fn init_local_disks(endpoint_pools: EndpointServerPools) -> Result<()> { - let opt = &DiskOption { - cleanup: true, - health_check: true, - }; - - let mut global_set_drives = GLOBAL_LOCAL_DISK_SET_DRIVES.write().await; - for pool_eps in endpoint_pools.as_ref().iter() { - let mut set_count_drives = Vec::with_capacity(pool_eps.set_count); - for _ in 0..pool_eps.set_count { - set_count_drives.push(vec![None; pool_eps.drives_per_set]); - } - - global_set_drives.push(set_count_drives); - } - - let mut global_local_disk_map = GLOBAL_LOCAL_DISK_MAP.write().await; - - for pool_eps in endpoint_pools.as_ref().iter() { - let mut set_drives = HashMap::new(); - for ep in pool_eps.endpoints.as_ref().iter() { - if !ep.is_local { - continue; - } - - let disk = new_disk(ep, opt).await?; - - let path = disk.endpoint().to_string(); - - global_local_disk_map.insert(path, Some(disk.clone())); - - set_drives.insert(ep.disk_idx, Some(disk.clone())); - - global_set_drives[ep.pool_idx as usize][ep.set_idx as usize][ep.disk_idx as usize] = Some(disk.clone()); - } - } - - Ok(()) -} - /// create unique lock clients for the endpoints and store them globally -pub fn init_lock_clients(endpoint_pools: EndpointServerPools) { - let mut unique_endpoints: HashMap = HashMap::new(); - - for pool_eps in endpoint_pools.as_ref().iter() { - for ep in pool_eps.endpoints.as_ref().iter() { - unique_endpoints.insert(ep.host_port(), ep); - } - } - - let mut clients = HashMap::new(); - let mut first_local_client_set = false; - - for (key, endpoint) in unique_endpoints { - if endpoint.is_local { - let local_client = Arc::new(LocalClient::new()) as Arc; - - // Store the first LocalClient globally for use by other modules - if !first_local_client_set { - if let Err(e) = crate::global::set_global_lock_client(local_client.clone()) { - // If already set, ignore the error (another thread may have set it) - warn!("set_global_lock_client error: {:?}", e); - } else { - first_local_client_set = true; - } - } - - clients.insert(key, local_client); - } else { - clients.insert(key, Arc::new(RemoteClient::new(endpoint.url.to_string())) as Arc); - } - } - - // Store the lock clients map globally - if crate::global::set_global_lock_clients(clients).is_err() { - error!("init_lock_clients: error setting lock clients"); - } -} - #[derive(Debug, Default)] struct PoolErr { index: Option, @@ -1241,48 +239,11 @@ impl ObjectIO for ECStore { h: HeaderMap, opts: &ObjectOptions, ) -> Result { - check_get_obj_args(bucket, object)?; - - let object = encode_dir_object(object); - - if self.single_pool() { - return self.pools[0].get_object_reader(bucket, object.as_str(), range, h, opts).await; - } - - // TODO: nslock - - let mut opts = opts.clone(); - - opts.no_lock = true; - - // TODO: check if DeleteMarker - let (_oi, idx) = self.get_latest_object_info_with_idx(bucket, &object, &opts).await?; - - self.pools[idx] - .get_object_reader(bucket, object.as_str(), range, h, &opts) - .await + self.handle_get_object_reader(bucket, object, range, h, opts).await } #[instrument(level = "debug", skip(self, data))] async fn put_object(&self, bucket: &str, object: &str, data: &mut PutObjReader, opts: &ObjectOptions) -> Result { - check_put_object_args(bucket, object)?; - - let object = encode_dir_object(object); - - if self.single_pool() { - return self.pools[0].put_object(bucket, object.as_str(), data, opts).await; - } - - let idx = self.get_pool_idx(bucket, &object, data.size()).await?; - - if opts.data_movement && idx == opts.src_pool_idx { - return Err(StorageError::DataMovementOverwriteErr( - bucket.to_owned(), - object.to_owned(), - opts.version_id.clone().unwrap_or_default(), - )); - } - - self.pools[idx].put_object(bucket, &object, data, opts).await + self.handle_put_object(bucket, object, data, opts).await } } @@ -1301,236 +262,37 @@ lazy_static! { impl StorageAPI for ECStore { #[instrument(skip(self))] async fn new_ns_lock(&self, bucket: &str, object: &str) -> Result { - self.pools[0].new_ns_lock(bucket, object).await + self.handle_new_ns_lock(bucket, object).await } #[instrument(skip(self))] async fn backend_info(&self) -> rustfs_madmin::BackendInfo { - let (standard_sc_parity, rr_sc_parity) = { - if let Some(sc) = GLOBAL_STORAGE_CLASS.get() { - let sc_parity = sc - .get_parity_for_sc(storageclass::CLASS_STANDARD) - .or(Some(self.pools[0].default_parity_count)); - - let rrs_sc_parity = sc.get_parity_for_sc(storageclass::RRS); - - (sc_parity, rrs_sc_parity) - } else { - (Some(self.pools[0].default_parity_count), None) - } - }; - - let mut standard_sc_data = Vec::new(); - let mut rr_sc_data = Vec::new(); - let mut drives_per_set = Vec::new(); - let mut total_sets = Vec::new(); - - for (idx, set_count) in self.set_drive_counts().iter().enumerate() { - if let Some(sc_parity) = standard_sc_parity { - standard_sc_data.push(set_count - sc_parity); - } - if let Some(sc_parity) = rr_sc_parity { - rr_sc_data.push(set_count - sc_parity); - } - total_sets.push(self.pools[idx].set_count); - drives_per_set.push(*set_count); - } - - rustfs_madmin::BackendInfo { - backend_type: rustfs_madmin::BackendByte::Erasure, - online_disks: rustfs_madmin::BackendDisks::new(), - offline_disks: rustfs_madmin::BackendDisks::new(), - standard_sc_data, - standard_sc_parity, - rr_sc_data, - rr_sc_parity, - total_sets, - drives_per_set, - ..Default::default() - } + self.handle_backend_info().await } #[instrument(skip(self))] async fn storage_info(&self) -> rustfs_madmin::StorageInfo { - let Some(notification_sy) = get_global_notification_sys() else { - return rustfs_madmin::StorageInfo::default(); - }; - - let mut info = notification_sy.storage_info(self).await; - - // 🔧 Defensive deduplication: This protection mechanism is retained even if the upstream is fixed - let original_count = info.disks.len(); - info.disks = Self::deduplicate_disks(info.disks); - let final_count = info.disks.len(); - - if original_count != final_count { - warn!( - "Storage info deduplication: removed {} duplicate disk entries ({} -> {})", - original_count - final_count, - original_count, - final_count - ); - } - - info + self.handle_storage_info().await } #[instrument(skip(self))] async fn local_storage_info(&self) -> rustfs_madmin::StorageInfo { - let mut futures = Vec::with_capacity(self.pools.len()); - - for pool in self.pools.iter() { - futures.push(pool.local_storage_info()) - } - - let results = join_all(futures).await; - - let mut disks = Vec::new(); - - for res in results.into_iter() { - disks.extend_from_slice(&res.disks); - } - - // 🔧 Defensive deduplication: when aggregating disks from all pools, drop duplicate - // entries that may be reported multiple times by backends; this extra layer is kept - // even if the upstream reporting is later fixed. - let original_count = disks.len(); - disks = Self::deduplicate_disks(disks); - - if original_count != disks.len() { - warn!("Local storage info deduplication: {} -> {}", original_count, disks.len()); - } - - let backend = self.backend_info().await; - rustfs_madmin::StorageInfo { backend, disks } + self.handle_local_storage_info().await } #[instrument(skip(self))] async fn make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()> { - if !is_meta_bucketname(bucket) - && let Err(err) = check_valid_bucket_name_strict(bucket) - { - return Err(StorageError::BucketNameInvalid(err.to_string())); - } - - // TODO: nslock - - if let Err(err) = self.peer_sys.make_bucket(bucket, opts).await { - let err = to_object_err(err.into(), vec![bucket]); - if !is_err_bucket_exists(&err) { - error!("make bucket failed: {err}"); - let _ = self - .delete_bucket( - bucket, - &DeleteBucketOptions { - no_lock: true, - no_recreate: true, - ..Default::default() - }, - ) - .await; - } - return Err(err); - }; - - let mut meta = BucketMetadata::new(bucket); - - meta.set_created(opts.created_at); - - if opts.lock_enabled { - meta.object_lock_config_xml = crate::bucket::utils::serialize::(&enableObjcetLockConfig)?; - meta.versioning_config_xml = crate::bucket::utils::serialize::(&enableVersioningConfig)?; - } - - if opts.versioning_enabled { - meta.versioning_config_xml = crate::bucket::utils::serialize::(&enableVersioningConfig)?; - } - - meta.save().await?; - - set_bucket_metadata(bucket.to_string(), meta).await?; - - Ok(()) + self.handle_make_bucket(bucket, opts).await } #[instrument(skip(self))] async fn get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result { - let mut info = self.peer_sys.get_bucket_info(bucket, opts).await?; - - if let Ok(sys) = metadata_sys::get(bucket).await { - info.created = Some(sys.created); - info.versioning = sys.versioning(); - info.object_locking = sys.object_locking(); - } - - Ok(info) + self.handle_get_bucket_info(bucket, opts).await } #[instrument(skip(self))] async fn list_bucket(&self, opts: &BucketOptions) -> Result> { - // TODO: opts.cached - - let mut buckets = self.peer_sys.list_bucket(opts).await?; - - if !opts.no_metadata { - for bucket in buckets.iter_mut() { - if let Ok(created) = metadata_sys::created_at(&bucket.name).await { - bucket.created = Some(created); - } - } - } - Ok(buckets) + self.handle_list_bucket(opts).await } #[instrument(skip(self))] async fn delete_bucket(&self, bucket: &str, opts: &DeleteBucketOptions) -> Result<()> { - if is_meta_bucketname(bucket) { - return Err(StorageError::BucketNameInvalid(bucket.to_string())); - } - - if let Err(err) = check_valid_bucket_name(bucket) { - return Err(StorageError::BucketNameInvalid(err.to_string())); - } - - // TODO: nslock - - // Check bucket exists before deletion (per S3 API spec) - // If bucket doesn't exist, return NoSuchBucket error - if let Err(err) = self.peer_sys.get_bucket_info(bucket, &BucketOptions::default()).await { - // Convert DiskError to StorageError for comparison - let storage_err: StorageError = err.into(); - if is_err_bucket_not_found(&storage_err) { - return Err(StorageError::BucketNotFound(bucket.to_string())); - } - return Err(to_object_err(storage_err, vec![bucket])); - } - - // Check bucket is empty before deletion (per S3 API spec) - // If bucket is not empty (contains actual objects with xl.meta files) and force - // is not set, return BucketNotEmpty error. - // Note: Empty directories (left after object deletion) should NOT count as objects. - if !opts.force { - let local_disks = all_local_disk().await; - for disk in local_disks.iter() { - // Check if bucket directory contains any xl.meta files (actual objects) - // We recursively scan for xl.meta files to determine if bucket has objects - // Use the disk's root path to construct bucket path - let bucket_path = disk.path().join(bucket); - if has_xlmeta_files(&bucket_path).await { - return Err(StorageError::BucketNotEmpty(bucket.to_string())); - } - } - } - - self.peer_sys - .delete_bucket(bucket, opts) - .await - .map_err(|e| to_object_err(e.into(), vec![bucket]))?; - - // TODO: replication opts.srdelete_op - - // Delete the metadata - self.delete_all(RUSTFS_META_BUCKET, format!("{BUCKET_META_PREFIX}/{bucket}").as_str()) - .await?; - if let Some(monitor) = get_global_bucket_monitor() { - monitor.delete_bucket(bucket); - } - Ok(()) + self.handle_delete_bucket(bucket, opts).await } // @continuation_token marker @@ -1549,7 +311,7 @@ impl StorageAPI for ECStore { start_after: Option, incl_deleted: bool, ) -> Result { - self.inner_list_objects_v2( + self.handle_list_objects_v2( bucket, prefix, continuation_token, @@ -1572,7 +334,7 @@ impl StorageAPI for ECStore { delimiter: Option, max_keys: i32, ) -> Result { - self.inner_list_object_versions(bucket, prefix, marker, version_marker, delimiter, max_keys) + self.handle_list_object_versions(bucket, prefix, marker, version_marker, delimiter, max_keys) .await } @@ -1584,25 +346,12 @@ impl StorageAPI for ECStore { result: tokio::sync::mpsc::Sender, opts: WalkOptions, ) -> Result<()> { - self.walk_internal(rx, bucket, prefix, result, opts).await + self.handle_walk(rx, bucket, prefix, result, opts).await } #[instrument(skip(self))] async fn get_object_info(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { - check_object_args(bucket, object)?; - - let object = encode_dir_object(object); - - if self.single_pool() { - return self.pools[0].get_object_info(bucket, object.as_str(), opts).await; - } - - // TODO: nslock - - let (info, _) = self.get_latest_object_info_with_idx(bucket, object.as_str(), opts).await?; - - opts.precondition_check(&info)?; - Ok(info) + self.handle_get_object_info(bucket, object, opts).await } // TODO: review @@ -1617,134 +366,12 @@ impl StorageAPI for ECStore { src_opts: &ObjectOptions, dst_opts: &ObjectOptions, ) -> Result { - check_copy_obj_args(src_bucket, src_object)?; - check_copy_obj_args(dst_bucket, dst_object)?; - - let src_object = encode_dir_object(src_object); - let dst_object = encode_dir_object(dst_object); - - let cp_src_dst_same = path_join_buf(&[src_bucket, &src_object]) == path_join_buf(&[dst_bucket, &dst_object]); - - // TODO: nslock - - let pool_idx = self.get_pool_idx_no_lock(src_bucket, &src_object, src_info.size).await?; - - if cp_src_dst_same { - if let (Some(src_vid), Some(dst_vid)) = (&src_opts.version_id, &dst_opts.version_id) - && src_vid == dst_vid - { - return self.pools[pool_idx] - .copy_object(src_bucket, &src_object, dst_bucket, &dst_object, src_info, src_opts, dst_opts) - .await; - } - - if !dst_opts.versioned && src_opts.version_id.is_none() { - return self.pools[pool_idx] - .copy_object(src_bucket, &src_object, dst_bucket, &dst_object, src_info, src_opts, dst_opts) - .await; - } - - if dst_opts.versioned && src_opts.version_id != dst_opts.version_id { - src_info.version_only = true; - return self.pools[pool_idx] - .copy_object(src_bucket, &src_object, dst_bucket, &dst_object, src_info, src_opts, dst_opts) - .await; - } - } - - let put_opts = ObjectOptions { - user_defined: src_info.user_defined.clone(), - versioned: dst_opts.versioned, - version_id: dst_opts.version_id.clone(), - no_lock: true, - mod_time: dst_opts.mod_time, - ..Default::default() - }; - - if let Some(put_object_reader) = src_info.put_object_reader.as_mut() { - return self.pools[pool_idx] - .put_object(dst_bucket, &dst_object, put_object_reader, &put_opts) - .await; - } - - Err(StorageError::InvalidArgument( - src_bucket.to_owned(), - src_object.to_owned(), - "put_object_reader is none".to_owned(), - )) + self.handle_copy_object(src_bucket, src_object, dst_bucket, dst_object, src_info, src_opts, dst_opts) + .await } #[instrument(skip(self))] async fn delete_object(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result { - check_del_obj_args(bucket, object)?; - - if opts.delete_prefix { - self.delete_prefix(bucket, object).await?; - return Ok(ObjectInfo::default()); - } - - // TODO: nslock - - let object = encode_dir_object(object); - let object = object.as_str(); - - let mut gopts = opts.clone(); - gopts.no_lock = true; - - // Determine which pool contains it - let (mut pinfo, errs) = self - .get_pool_info_existing_with_opts(bucket, object, &gopts) - .await - .map_err(|e| { - if is_err_read_quorum(&e) { - StorageError::ErasureWriteQuorum - } else { - e - } - })?; - - if pinfo.object_info.delete_marker && opts.version_id.is_none() { - pinfo.object_info.name = decode_dir_object(object); - return Ok(pinfo.object_info); - } - - if opts.data_movement && opts.src_pool_idx == pinfo.index { - return Err(StorageError::DataMovementOverwriteErr( - bucket.to_owned(), - object.to_owned(), - opts.version_id.unwrap_or_default(), - )); - } - - if opts.data_movement { - let mut obj = self.pools[pinfo.index].delete_object(bucket, object, opts).await?; - obj.name = decode_dir_object(obj.name.as_str()); - return Ok(obj); - } - - if !errs.is_empty() && !opts.versioned && !opts.version_suspended { - return self.delete_object_from_all_pools(bucket, object, &opts, errs).await; - } - - for pool in self.pools.iter() { - match pool.delete_object(bucket, object, opts.clone()).await { - Ok(res) => { - let mut obj = res; - obj.name = decode_dir_object(object); - return Ok(obj); - } - Err(err) => { - if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) { - return Err(err); - } - } - } - } - - if let Some(ver) = opts.version_id { - return Err(StorageError::VersionNotFound(bucket.to_owned(), object.to_owned(), ver)); - } - - Err(StorageError::ObjectNotFound(bucket.to_owned(), object.to_owned())) + self.handle_delete_object(bucket, object, opts).await } // TODO: review #[instrument(skip(self))] @@ -1754,180 +381,7 @@ impl StorageAPI for ECStore { objects: Vec, opts: ObjectOptions, ) -> (Vec, Vec>) { - // encode object name - let objects: Vec = objects - .iter() - .map(|v| { - let mut v = v.clone(); - v.object_name = encode_dir_object(v.object_name.as_str()); - v - }) - .collect(); - - // Default return value - let mut del_objects = vec![DeletedObject::default(); objects.len()]; - - let mut del_errs = Vec::with_capacity(objects.len()); - for _ in 0..objects.len() { - del_errs.push(None) - } - - // TODO: nslock - - let mut futures = Vec::with_capacity(self.pools.len()); - - for pool in self.pools.iter() { - futures.push(pool.delete_objects(bucket, objects.clone(), opts.clone())); - } - - let results = join_all(futures).await; - - for idx in 0..del_objects.len() { - for (dels, errs) in results.iter() { - if errs[idx].is_none() && dels[idx].found { - del_errs[idx] = None; - del_objects[idx] = dels[idx].clone(); - break; - } - - if del_errs[idx].is_none() { - del_errs[idx] = errs[idx].clone(); - del_objects[idx] = dels[idx].clone(); - } - } - } - - del_objects.iter_mut().for_each(|v| { - v.object_name = decode_dir_object(&v.object_name); - }); - - (del_objects, del_errs) - - // let mut futures = Vec::with_capacity(objects.len()); - - // for obj in objects.iter() { - // futures.push(async move { - // self.internal_get_pool_info_existing_with_opts( - // bucket, - // &obj.object_name, - // &ObjectOptions { - // no_lock: true, - // ..Default::default() - // }, - // ) - // .await - // }); - // } - - // let results = join_all(futures).await; - - // // let mut jhs = Vec::new(); - // // let semaphore = Arc::new(Semaphore::new(num_cpus::get())); - // // let pools = Arc::new(self.pools.clone()); - - // // for obj in objects.iter() { - // // let (semaphore, pools, bucket, object_name, opt) = ( - // // semaphore.clone(), - // // pools.clone(), - // // bucket.to_string(), - // // obj.object_name.to_string(), - // // ObjectOptions::default(), - // // ); - - // // let jh = tokio::spawn(async move { - // // let _permit = semaphore.acquire().await.unwrap(); - // // self.internal_get_pool_info_existing_with_opts(pools.as_ref(), &bucket, &object_name, &opt) - // // .await - // // }); - // // jhs.push(jh); - // // } - // // let mut results = Vec::new(); - // // for jh in jhs { - // // results.push(jh.await.unwrap()); - // // } - - // // Record the mapping pool_idx -> object index - // let mut pool_obj_idx_map = HashMap::new(); - // let mut orig_index_map = HashMap::new(); - - // for (i, res) in results.into_iter().enumerate() { - // match res { - // Ok((pinfo, _)) => { - // if let Some(obj) = objects.get(i) { - // if pinfo.object_info.delete_marker && obj.version_id.is_none() { - // del_objects[i] = DeletedObject { - // delete_marker: pinfo.object_info.delete_marker, - // delete_marker_version_id: pinfo.object_info.version_id.map(|v| v.to_string()), - // object_name: decode_dir_object(&pinfo.object_info.name), - // delete_marker_mtime: pinfo.object_info.mod_time, - // ..Default::default() - // }; - // continue; - // } - - // if !pool_obj_idx_map.contains_key(&pinfo.index) { - // pool_obj_idx_map.insert(pinfo.index, vec![obj.clone()]); - // } else if let Some(val) = pool_obj_idx_map.get_mut(&pinfo.index) { - // val.push(obj.clone()); - // } - - // if !orig_index_map.contains_key(&pinfo.index) { - // orig_index_map.insert(pinfo.index, vec![i]); - // } else if let Some(val) = orig_index_map.get_mut(&pinfo.index) { - // val.push(i); - // } - // } - // } - // Err(e) => { - // if !is_err_object_not_found(&e) && is_err_version_not_found(&e) { - // del_errs[i] = Some(e) - // } - - // if let Some(obj) = objects.get(i) { - // del_objects[i] = DeletedObject { - // object_name: decode_dir_object(&obj.object_name), - // version_id: obj.version_id.map(|v| v.to_string()), - // ..Default::default() - // } - // } - // } - // } - // } - - // if !pool_obj_idx_map.is_empty() { - // for (i, sets) in self.pools.iter().enumerate() { - // // Retrieve the object index for a pool idx - // if let Some(objs) = pool_obj_idx_map.get(&i) { - // // Fetch the corresponding object (should never be None) - // // let objs: Vec = obj_idxs.iter().filter_map(|&idx| objects.get(idx).cloned()).collect(); - - // if objs.is_empty() { - // continue; - // } - - // let (pdel_objs, perrs) = sets.delete_objects(bucket, objs.clone(), opts.clone()).await?; - - // // Insert simultaneously (should never be None) - // let org_indexes = orig_index_map.get(&i).unwrap(); - - // // perrs should follow the same order as obj_idxs - // for (i, err) in perrs.into_iter().enumerate() { - // let obj_idx = org_indexes[i]; - - // if err.is_some() { - // del_errs[obj_idx] = err; - // } - - // let mut dobj = pdel_objs.get(i).unwrap().clone(); - // dobj.object_name = decode_dir_object(&dobj.object_name); - - // del_objects[obj_idx] = dobj; - // } - // } - // } - // } - - // Ok((del_objects, del_errs)) + self.handle_delete_objects(bucket, objects, opts).await } #[instrument(skip(self))] @@ -1940,35 +394,8 @@ impl StorageAPI for ECStore { max_parts: usize, opts: &ObjectOptions, ) -> Result { - check_list_parts_args(bucket, object, upload_id)?; - - // TODO: nslock - - if self.single_pool() { - return self.pools[0] - .list_object_parts(bucket, object, upload_id, part_number_marker, max_parts, opts) - .await; - } - - for pool in self.pools.iter() { - if self.is_suspended(pool.pool_idx).await { - continue; - } - return match pool - .list_object_parts(bucket, object, upload_id, part_number_marker, max_parts, opts) - .await - { - Ok(res) => Ok(res), - Err(err) => { - if is_err_invalid_upload_id(&err) { - continue; - } - Err(err) - } - }; - } - - Err(StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned())) + self.handle_list_object_parts(bucket, object, upload_id, part_number_marker, max_parts, opts) + .await } #[instrument(skip(self))] @@ -1981,124 +408,27 @@ impl StorageAPI for ECStore { delimiter: Option, max_uploads: usize, ) -> Result { - check_list_multipart_args(bucket, prefix, &key_marker, &upload_id_marker, &delimiter)?; - - if prefix.is_empty() { - // TODO: return from cache - } - - if self.single_pool() { - return self.pools[0] - .list_multipart_uploads(bucket, prefix, key_marker, upload_id_marker, delimiter, max_uploads) - .await; - } - - let mut uploads = Vec::new(); - - for pool in self.pools.iter() { - if self.is_suspended(pool.pool_idx).await { - continue; - } - let res = pool - .list_multipart_uploads( - bucket, - prefix, - key_marker.clone(), - upload_id_marker.clone(), - delimiter.clone(), - max_uploads, - ) - .await?; - uploads.extend(res.uploads); - } - - Ok(ListMultipartsInfo { - key_marker, - upload_id_marker, - max_uploads, - uploads, - prefix: prefix.to_owned(), - delimiter: delimiter.to_owned(), - ..Default::default() - }) + self.handle_list_multipart_uploads(bucket, prefix, key_marker, upload_id_marker, delimiter, max_uploads) + .await } #[instrument(skip(self))] async fn new_multipart_upload(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { - check_new_multipart_args(bucket, object)?; - - if self.single_pool() { - return self.pools[0].new_multipart_upload(bucket, object, opts).await; - } - - for (idx, pool) in self.pools.iter().enumerate() { - if self.is_suspended(idx).await || self.is_pool_rebalancing(idx).await { - continue; - } - let res = pool - .list_multipart_uploads(bucket, object, None, None, None, MAX_UPLOADS_LIST) - .await?; - - if !res.uploads.is_empty() { - return self.pools[idx].new_multipart_upload(bucket, object, opts).await; - } - } - let idx = self.get_pool_idx(bucket, object, -1).await?; - if opts.data_movement && idx == opts.src_pool_idx { - return Err(StorageError::DataMovementOverwriteErr( - bucket.to_owned(), - object.to_owned(), - "".to_owned(), - )); - } - - self.pools[idx].new_multipart_upload(bucket, object, opts).await + self.handle_new_multipart_upload(bucket, object, opts).await } #[instrument(skip(self))] async fn add_partial(&self, bucket: &str, object: &str, version_id: &str) -> Result<()> { - let object = encode_dir_object(object); - - if self.single_pool() { - let _ = self.pools[0].add_partial(bucket, object.as_str(), version_id).await; - } - - let idx = self - .get_pool_idx_existing_with_opts(bucket, object.as_str(), &ObjectOptions::default()) - .await?; - - let _ = self.pools[idx].add_partial(bucket, object.as_str(), version_id).await; - Ok(()) + self.handle_add_partial(bucket, object, version_id).await } #[instrument(skip(self))] async fn transition_object(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<()> { - let object = encode_dir_object(object); - if self.single_pool() { - return self.pools[0].transition_object(bucket, &object, opts).await; - } - - //opts.skip_decommissioned = true; - //opts.no_lock = true; - let idx = self.get_pool_idx_existing_with_opts(bucket, &object, opts).await?; - - self.pools[idx].transition_object(bucket, &object, opts).await + self.handle_transition_object(bucket, object, opts).await } #[instrument(skip(self))] async fn restore_transitioned_object(self: Arc, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<()> { - let object = encode_dir_object(object); - if self.single_pool() { - return self.pools[0].clone().restore_transitioned_object(bucket, &object, opts).await; - } - - //opts.skip_decommissioned = true; - //opts.nolock = true; - let idx = self.get_pool_idx_existing_with_opts(bucket, object.as_str(), opts).await?; - - self.pools[idx] - .clone() - .restore_transitioned_object(bucket, &object, opts) - .await + self.handle_restore_transitioned_object(bucket, object, opts).await } #[instrument(skip(self))] @@ -2116,12 +446,20 @@ impl StorageAPI for ECStore { _src_opts: &ObjectOptions, _dst_opts: &ObjectOptions, ) -> Result<()> { - check_new_multipart_args(src_bucket, src_object)?; - - // TODO: PutObjectReader - // self.put_object_part(dst_bucket, dst_object, upload_id, part_id, data, opts) - - unimplemented!() + self.handle_copy_object_part( + src_bucket, + src_object, + _dst_bucket, + _dst_object, + _upload_id, + _part_id, + _start_offset, + _length, + _src_info, + _src_opts, + _dst_opts, + ) + .await } #[instrument(skip(self, data))] async fn put_object_part( @@ -2133,36 +471,8 @@ impl StorageAPI for ECStore { data: &mut PutObjReader, opts: &ObjectOptions, ) -> Result { - check_put_object_part_args(bucket, object, upload_id)?; - - if self.single_pool() { - return self.pools[0] - .put_object_part(bucket, object, upload_id, part_id, data, opts) - .await; - } - - for pool in self.pools.iter() { - if self.is_suspended(pool.pool_idx).await { - continue; - } - let err = match pool.put_object_part(bucket, object, upload_id, part_id, data, opts).await { - Ok(res) => return Ok(res), - Err(err) => { - if is_err_invalid_upload_id(&err) { - None - } else { - Some(err) - } - } - }; - - if let Some(err) = err { - error!("put_object_part err: {:?}", err); - return Err(err); - } - } - - Err(StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned())) + self.handle_put_object_part(bucket, object, upload_id, part_id, data, opts) + .await } #[instrument(skip(self))] @@ -2173,59 +483,11 @@ impl StorageAPI for ECStore { upload_id: &str, opts: &ObjectOptions, ) -> Result { - check_list_parts_args(bucket, object, upload_id)?; - if self.single_pool() { - return self.pools[0].get_multipart_info(bucket, object, upload_id, opts).await; - } - - for pool in self.pools.iter() { - if self.is_suspended(pool.pool_idx).await { - continue; - } - - return match pool.get_multipart_info(bucket, object, upload_id, opts).await { - Ok(res) => Ok(res), - Err(err) => { - if is_err_invalid_upload_id(&err) { - continue; - } - - Err(err) - } - }; - } - - Err(StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned())) + self.handle_get_multipart_info(bucket, object, upload_id, opts).await } #[instrument(skip(self))] async fn abort_multipart_upload(&self, bucket: &str, object: &str, upload_id: &str, opts: &ObjectOptions) -> Result<()> { - check_abort_multipart_args(bucket, object, upload_id)?; - - // TODO: defer DeleteUploadID - - if self.single_pool() { - return self.pools[0].abort_multipart_upload(bucket, object, upload_id, opts).await; - } - - for pool in self.pools.iter() { - if self.is_suspended(pool.pool_idx).await { - continue; - } - - let err = match pool.abort_multipart_upload(bucket, object, upload_id, opts).await { - Ok(_) => return Ok(()), - Err(err) => { - // - if is_err_invalid_upload_id(&err) { None } else { Some(err) } - } - }; - - if let Some(er) = err { - return Err(er); - } - } - - Err(StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned())) + self.handle_abort_multipart_upload(bucket, object, upload_id, opts).await } #[instrument(skip(self))] @@ -2237,165 +499,51 @@ impl StorageAPI for ECStore { uploaded_parts: Vec, opts: &ObjectOptions, ) -> Result { - check_complete_multipart_args(bucket, object, upload_id)?; - - if self.single_pool() { - return self.pools[0] - .clone() - .complete_multipart_upload(bucket, object, upload_id, uploaded_parts, opts) - .await; - } - - for pool in self.pools.iter() { - if self.is_suspended(pool.pool_idx).await { - continue; - } - - let pool = pool.clone(); - let err = match pool - .complete_multipart_upload(bucket, object, upload_id, uploaded_parts.clone(), opts) - .await - { - Ok(res) => return Ok(res), - Err(err) => { - // - if is_err_invalid_upload_id(&err) { None } else { Some(err) } - } - }; - - if let Some(er) = err { - return Err(er); - } - } - - Err(StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned())) + self.handle_complete_multipart_upload(bucket, object, upload_id, uploaded_parts, opts) + .await } #[instrument(skip(self))] async fn get_disks(&self, pool_idx: usize, set_idx: usize) -> Result>> { - if pool_idx < self.pools.len() && set_idx < self.pools[pool_idx].disk_set.len() { - self.pools[pool_idx].disk_set[set_idx].get_disks(0, 0).await - } else { - Err(Error::other(format!("pool idx {pool_idx}, set idx {set_idx}, not found"))) - } + self.handle_get_disks(pool_idx, set_idx).await } #[instrument(skip(self))] fn set_drive_counts(&self) -> Vec { - let mut counts = vec![0; self.pools.len()]; - - for (i, pool) in self.pools.iter().enumerate() { - counts[i] = pool.set_drive_count(); - } - counts + self.handle_set_drive_counts() } #[instrument(skip(self))] async fn put_object_metadata(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { - let object = encode_dir_object(object); - if self.single_pool() { - return self.pools[0].put_object_metadata(bucket, object.as_str(), opts).await; - } - - let mut opts = opts.clone(); - opts.metadata_chg = true; - - let idx = self.get_pool_idx_existing_with_opts(bucket, object.as_str(), &opts).await?; - - self.pools[idx].put_object_metadata(bucket, object.as_str(), &opts).await + self.handle_put_object_metadata(bucket, object, opts).await } #[instrument(skip(self))] async fn get_object_tags(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { - let object = encode_dir_object(object); - - if self.single_pool() { - return self.pools[0].get_object_tags(bucket, object.as_str(), opts).await; - } - - let (oi, _) = self.get_latest_object_info_with_idx(bucket, &object, opts).await?; - - Ok(oi.user_tags) + self.handle_get_object_tags(bucket, object, opts).await } #[instrument(level = "debug", skip(self))] async fn put_object_tags(&self, bucket: &str, object: &str, tags: &str, opts: &ObjectOptions) -> Result { - let object = encode_dir_object(object); - - if self.single_pool() { - return self.pools[0].put_object_tags(bucket, object.as_str(), tags, opts).await; - } - - let idx = self.get_pool_idx_existing_with_opts(bucket, object.as_str(), opts).await?; - - self.pools[idx].put_object_tags(bucket, object.as_str(), tags, opts).await + self.handle_put_object_tags(bucket, object, tags, opts).await } #[instrument(skip(self))] async fn delete_object_version(&self, bucket: &str, object: &str, fi: &FileInfo, force_del_marker: bool) -> Result<()> { - check_del_obj_args(bucket, object)?; - - let object = encode_dir_object(object); - - if self.single_pool() { - return self.pools[0] - .delete_object_version(bucket, object.as_str(), fi, force_del_marker) - .await; - } - Ok(()) + self.handle_delete_object_version(bucket, object, fi, force_del_marker).await } #[instrument(skip(self))] async fn delete_object_tags(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { - let object = encode_dir_object(object); - - if self.single_pool() { - return self.pools[0].delete_object_tags(bucket, object.as_str(), opts).await; - } - - let idx = self.get_pool_idx_existing_with_opts(bucket, object.as_str(), opts).await?; - - self.pools[idx].delete_object_tags(bucket, object.as_str(), opts).await + self.handle_delete_object_tags(bucket, object, opts).await } #[instrument(skip(self))] async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option)> { - info!("heal_format"); - let mut r = HealResultItem { - heal_item_type: HealItemType::Metadata.to_string(), - detail: "disk-format".to_string(), - ..Default::default() - }; - - let mut count_no_heal = 0; - for pool in self.pools.iter() { - let (mut result, err) = pool.heal_format(dry_run).await?; - if let Some(err) = err { - match err { - StorageError::NoHealRequired => { - count_no_heal += 1; - } - _ => { - continue; - } - } - } - r.disk_count += result.disk_count; - r.set_count += result.set_count; - r.before.drives.append(&mut result.before.drives); - r.after.drives.append(&mut result.after.drives); - } - if count_no_heal == self.pools.len() { - info!("heal format success, NoHealRequired"); - return Ok((r, Some(StorageError::NoHealRequired))); - } - info!("heal format success result: {:?}", r); - Ok((r, None)) + self.handle_heal_format(dry_run).await } #[instrument(skip(self))] async fn heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result { - let res = self.peer_sys.heal_bucket(bucket, opts).await?; - - Ok(res) + self.handle_heal_bucket(bucket, opts).await } #[instrument(skip(self))] async fn heal_object( @@ -2405,146 +553,24 @@ impl StorageAPI for ECStore { version_id: &str, opts: &HealOpts, ) -> Result<(HealResultItem, Option)> { - info!("ECStore heal_object"); - let object = encode_dir_object(object); - - let mut futures = Vec::with_capacity(self.pools.len()); - for pool in self.pools.iter() { - if self.is_suspended(pool.pool_idx).await { - continue; - } - futures.push(pool.heal_object(bucket, &object, version_id, opts)); - } - let results = join_all(futures).await; - - let mut errs = Vec::with_capacity(self.pools.len()); - let mut ress = Vec::with_capacity(self.pools.len()); - - for res in results.into_iter() { - match res { - Ok((result, err)) => { - let mut result = result; - result.object = decode_dir_object(&result.object); - ress.push(result); - errs.push(err); - } - Err(err) => { - errs.push(Some(err)); - ress.push(HealResultItem::default()); - } - } - } - - for (idx, err) in errs.iter().enumerate() { - if err.is_none() { - return Ok((ress.remove(idx), None)); - } - } - - // No pool returned a nil error, return the first non 'not found' error - for (index, err) in errs.iter().enumerate() { - return match err { - Some(err) => { - if is_err_object_not_found(err) || is_err_version_not_found(err) { - continue; - } - Ok((ress.remove(index), Some(err.clone()))) - } - None => Ok((ress.remove(index), None)), - }; - } - - // At this stage, all errors are 'not found' - if !version_id.is_empty() { - return Ok((HealResultItem::default(), Some(Error::FileVersionNotFound))); - } - - Ok((HealResultItem::default(), Some(Error::FileNotFound))) + self.handle_heal_object(bucket, object, version_id, opts).await } #[instrument(skip(self))] async fn get_pool_and_set(&self, id: &str) -> Result<(Option, Option, Option)> { - for (pool_idx, pool) in self.pools.iter().enumerate() { - for (set_idx, set) in pool.format.erasure.sets.iter().enumerate() { - for (disk_idx, disk_id) in set.iter().enumerate() { - if disk_id.to_string() == id { - return Ok((Some(pool_idx), Some(set_idx), Some(disk_idx))); - } - } - } - } - - Err(Error::DiskNotFound) + self.handle_get_pool_and_set(id).await } #[instrument(skip(self))] async fn check_abandoned_parts(&self, bucket: &str, object: &str, opts: &HealOpts) -> Result<()> { - let object = encode_dir_object(object); - if self.single_pool() { - return self.pools[0].check_abandoned_parts(bucket, &object, opts).await; - } - - let mut errs = Vec::new(); - for pool in self.pools.iter() { - //TODO: IsSuspended - if let Err(err) = pool.check_abandoned_parts(bucket, &object, opts).await { - errs.push(err); - } - } - - if !errs.is_empty() { - return Err(errs[0].clone()); - } - - Ok(()) + self.handle_check_abandoned_parts(bucket, object, opts).await } async fn verify_object_integrity(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<()> { - let get_object_reader = ::get_object_reader(self, bucket, object, None, HeaderMap::new(), opts).await?; - // Stream to sink to avoid loading entire object into memory during verification - let mut reader = get_object_reader.stream; - tokio::io::copy(&mut reader, &mut tokio::io::sink()).await?; - Ok(()) + self.handle_verify_object_integrity(bucket, object, opts).await } } -async fn init_local_peer(endpoint_pools: &EndpointServerPools, host: &String, port: &String) { - let mut peer_set = Vec::new(); - endpoint_pools.as_ref().iter().for_each(|endpoints| { - endpoints.endpoints.as_ref().iter().for_each(|endpoint| { - if endpoint.get_type() == EndpointType::Url && endpoint.is_local && endpoint.url.has_host() { - peer_set.push(endpoint.url.host_str().unwrap().to_string()); - } - }); - }); - - if peer_set.is_empty() { - if !host.is_empty() { - *GLOBAL_LOCAL_NODE_NAME.write().await = format!("{host}:{port}"); - return; - } - - *GLOBAL_LOCAL_NODE_NAME.write().await = format!("127.0.0.1:{port}"); - return; - } - - *GLOBAL_LOCAL_NODE_NAME.write().await = peer_set[0].clone(); -} - -pub async fn get_disk_infos(disks: &[Option]) -> Vec> { - let opts = &DiskInfoOptions::default(); - let mut res = vec![None; disks.len()]; - for (idx, disk_op) in disks.iter().enumerate() { - if let Some(disk) = disk_op - && let Ok(info) = disk.disk_info(opts).await - { - res[idx] = Some(info); - } - } - - res -} - #[derive(Debug, Default, Clone)] pub struct PoolAvailableSpace { pub index: usize, @@ -2598,50 +624,6 @@ impl ServerPoolsAvailableSpace { } } -pub async fn has_space_for(dis: &[Option], size: i64) -> Result { - let size = { if size < 0 { DISK_ASSUME_UNKNOWN_SIZE } else { size as u64 * 2 } }; - - let mut available = 0; - let mut total = 0; - let mut disks_num = 0; - - for disk in dis.iter().flatten() { - disks_num += 1; - total += disk.total; - available += disk.total - disk.used; - } - - if disks_num < dis.len() / 2 || disks_num == 0 { - return Err(Error::other(format!( - "not enough online disks to calculate the available space,need {}, found {}", - (dis.len() / 2) + 1, - disks_num, - ))); - } - - let per_disk = size / disks_num as u64; - - for disk in dis.iter().flatten() { - if !is_erasure_sd().await && disk.free_inodes < DISK_MIN_INODES && disk.used_inodes > 0 { - return Ok(false); - } - - if disk.free <= per_disk { - return Ok(false); - } - } - - if available < size { - return Ok(false); - } - - available -= size; - - let want = total as f64 * (1.0 - DISK_FILL_FRACTION); - - Ok(available > want as u64) -} - #[cfg(test)] mod tests { use super::*; diff --git a/crates/ecstore/src/store/bucket.rs b/crates/ecstore/src/store/bucket.rs new file mode 100644 index 000000000..f40dbfc7f --- /dev/null +++ b/crates/ecstore/src/store/bucket.rs @@ -0,0 +1,150 @@ +// 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 super::*; + +impl ECStore { + #[instrument(skip(self))] + pub(super) async fn handle_make_bucket(&self, bucket: &str, opts: &MakeBucketOptions) -> Result<()> { + if !is_meta_bucketname(bucket) + && let Err(err) = check_valid_bucket_name_strict(bucket) + { + return Err(StorageError::BucketNameInvalid(err.to_string())); + } + + // TODO: nslock + + if let Err(err) = self.peer_sys.make_bucket(bucket, opts).await { + let err = to_object_err(err.into(), vec![bucket]); + if !is_err_bucket_exists(&err) { + error!("make bucket failed: {err}"); + let _ = self + .delete_bucket( + bucket, + &DeleteBucketOptions { + no_lock: true, + no_recreate: true, + ..Default::default() + }, + ) + .await; + } + return Err(err); + }; + + let mut meta = BucketMetadata::new(bucket); + + meta.set_created(opts.created_at); + + if opts.lock_enabled { + meta.object_lock_config_xml = crate::bucket::utils::serialize::(&enableObjcetLockConfig)?; + meta.versioning_config_xml = crate::bucket::utils::serialize::(&enableVersioningConfig)?; + } + + if opts.versioning_enabled { + meta.versioning_config_xml = crate::bucket::utils::serialize::(&enableVersioningConfig)?; + } + + meta.save().await?; + + set_bucket_metadata(bucket.to_string(), meta).await?; + + Ok(()) + } + + #[instrument(skip(self))] + pub(super) async fn handle_get_bucket_info(&self, bucket: &str, opts: &BucketOptions) -> Result { + let mut info = self.peer_sys.get_bucket_info(bucket, opts).await?; + + if let Ok(sys) = metadata_sys::get(bucket).await { + info.created = Some(sys.created); + info.versioning = sys.versioning(); + info.object_locking = sys.object_locking(); + } + + Ok(info) + } + + #[instrument(skip(self))] + pub(super) async fn handle_list_bucket(&self, opts: &BucketOptions) -> Result> { + // TODO: opts.cached + + let mut buckets = self.peer_sys.list_bucket(opts).await?; + + if !opts.no_metadata { + for bucket in buckets.iter_mut() { + if let Ok(created) = metadata_sys::created_at(&bucket.name).await { + bucket.created = Some(created); + } + } + } + Ok(buckets) + } + + #[instrument(skip(self))] + pub(super) async fn handle_delete_bucket(&self, bucket: &str, opts: &DeleteBucketOptions) -> Result<()> { + if is_meta_bucketname(bucket) { + return Err(StorageError::BucketNameInvalid(bucket.to_string())); + } + + if let Err(err) = check_valid_bucket_name(bucket) { + return Err(StorageError::BucketNameInvalid(err.to_string())); + } + + // TODO: nslock + + // Check bucket exists before deletion (per S3 API spec) + // If bucket doesn't exist, return NoSuchBucket error + if let Err(err) = self.peer_sys.get_bucket_info(bucket, &BucketOptions::default()).await { + // Convert DiskError to StorageError for comparison + let storage_err: StorageError = err.into(); + if is_err_bucket_not_found(&storage_err) { + return Err(StorageError::BucketNotFound(bucket.to_string())); + } + return Err(to_object_err(storage_err, vec![bucket])); + } + + // Check bucket is empty before deletion (per S3 API spec) + // If bucket is not empty (contains actual objects with xl.meta files) and force + // is not set, return BucketNotEmpty error. + // Note: Empty directories (left after object deletion) should NOT count as objects. + if !opts.force { + let local_disks = all_local_disk().await; + for disk in local_disks.iter() { + // Check if bucket directory contains any xl.meta files (actual objects) + // We recursively scan for xl.meta files to determine if bucket has objects + // Use the disk's root path to construct bucket path + let bucket_path = disk.path().join(bucket); + if has_xlmeta_files(&bucket_path).await { + return Err(StorageError::BucketNotEmpty(bucket.to_string())); + } + } + } + + self.peer_sys + .delete_bucket(bucket, opts) + .await + .map_err(|e| to_object_err(e.into(), vec![bucket]))?; + + // TODO: replication opts.srdelete_op + + // Delete the metadata + self.delete_all(RUSTFS_META_BUCKET, format!("{BUCKET_META_PREFIX}/{bucket}").as_str()) + .await?; + if let Some(monitor) = get_global_bucket_monitor() { + monitor.delete_bucket(bucket); + } + Ok(()) + } +} diff --git a/crates/ecstore/src/store/heal.rs b/crates/ecstore/src/store/heal.rs new file mode 100644 index 000000000..5e524efab --- /dev/null +++ b/crates/ecstore/src/store/heal.rs @@ -0,0 +1,146 @@ +// 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 super::*; + +impl ECStore { + #[instrument(skip(self))] + pub(super) async fn handle_heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option)> { + info!("heal_format"); + let mut r = HealResultItem { + heal_item_type: HealItemType::Metadata.to_string(), + detail: "disk-format".to_string(), + ..Default::default() + }; + + let mut count_no_heal = 0; + for pool in self.pools.iter() { + let (mut result, err) = pool.heal_format(dry_run).await?; + if let Some(err) = err { + match err { + StorageError::NoHealRequired => { + count_no_heal += 1; + } + _ => { + continue; + } + } + } + r.disk_count += result.disk_count; + r.set_count += result.set_count; + r.before.drives.append(&mut result.before.drives); + r.after.drives.append(&mut result.after.drives); + } + if count_no_heal == self.pools.len() { + info!("heal format success, NoHealRequired"); + return Ok((r, Some(StorageError::NoHealRequired))); + } + info!("heal format success result: {:?}", r); + Ok((r, None)) + } + + #[instrument(skip(self))] + pub(super) async fn handle_heal_bucket(&self, bucket: &str, opts: &HealOpts) -> Result { + let res = self.peer_sys.heal_bucket(bucket, opts).await?; + + Ok(res) + } + + #[instrument(skip(self))] + pub(super) async fn handle_heal_object( + &self, + bucket: &str, + object: &str, + version_id: &str, + opts: &HealOpts, + ) -> Result<(HealResultItem, Option)> { + info!("ECStore heal_object"); + let object = encode_dir_object(object); + + let mut futures = Vec::with_capacity(self.pools.len()); + for pool in self.pools.iter() { + if self.is_suspended(pool.pool_idx).await { + continue; + } + futures.push(pool.heal_object(bucket, &object, version_id, opts)); + } + let results = join_all(futures).await; + + let mut errs = Vec::with_capacity(self.pools.len()); + let mut ress = Vec::with_capacity(self.pools.len()); + + for res in results.into_iter() { + match res { + Ok((result, err)) => { + let mut result = result; + result.object = decode_dir_object(&result.object); + ress.push(result); + errs.push(err); + } + Err(err) => { + errs.push(Some(err)); + ress.push(HealResultItem::default()); + } + } + } + + for (idx, err) in errs.iter().enumerate() { + if err.is_none() { + return Ok((ress.remove(idx), None)); + } + } + + // No pool returned a nil error, return the first non 'not found' error + for (index, err) in errs.iter().enumerate() { + return match err { + Some(err) => { + if is_err_object_not_found(err) || is_err_version_not_found(err) { + continue; + } + Ok((ress.remove(index), Some(err.clone()))) + } + None => Ok((ress.remove(index), None)), + }; + } + + // At this stage, all errors are 'not found' + if !version_id.is_empty() { + return Ok((HealResultItem::default(), Some(Error::FileVersionNotFound))); + } + + Ok((HealResultItem::default(), Some(Error::FileNotFound))) + } + + #[instrument(skip(self))] + pub(super) async fn handle_check_abandoned_parts(&self, bucket: &str, object: &str, opts: &HealOpts) -> Result<()> { + let object = encode_dir_object(object); + if self.single_pool() { + return self.pools[0].check_abandoned_parts(bucket, &object, opts).await; + } + + let mut errs = Vec::new(); + for pool in self.pools.iter() { + //TODO: IsSuspended + if let Err(err) = pool.check_abandoned_parts(bucket, &object, opts).await { + errs.push(err); + } + } + + if !errs.is_empty() { + return Err(errs[0].clone()); + } + + Ok(()) + } +} diff --git a/crates/ecstore/src/store/init.rs b/crates/ecstore/src/store/init.rs new file mode 100644 index 000000000..364b7a1e4 --- /dev/null +++ b/crates/ecstore/src/store/init.rs @@ -0,0 +1,281 @@ +// 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 super::*; + +impl ECStore { + #[allow(clippy::new_ret_no_self)] + #[instrument(level = "debug", skip(endpoint_pools))] + pub async fn new(address: SocketAddr, endpoint_pools: EndpointServerPools, ctx: CancellationToken) -> Result> { + // let layouts = DisksLayout::from_volumes(endpoints.as_slice())?; + + let mut deployment_id = None; + + // let (endpoint_pools, _) = EndpointServerPools::create_server_endpoints(address.as_str(), &layouts)?; + + let mut pools = Vec::with_capacity(endpoint_pools.as_ref().len()); + let mut disk_map = HashMap::with_capacity(endpoint_pools.as_ref().len()); + + let first_is_local = endpoint_pools.first_local(); + + let mut local_disks = Vec::new(); + + info!("ECStore new address: {}", address.to_string()); + let mut host = address.ip().to_string(); + if host.is_empty() { + host = GLOBAL_RUSTFS_HOST.read().await.to_string() + } + let mut port = address.port().to_string(); + if port.is_empty() { + port = GLOBAL_RUSTFS_PORT.read().await.to_string() + } + info!("ECStore new host: {}, port: {}", host, port); + init_local_peer(&endpoint_pools, &host, &port).await; + + // debug!("endpoint_pools: {:?}", endpoint_pools); + + let mut common_parity_drives = 0; + + for (i, pool_eps) in endpoint_pools.as_ref().iter().enumerate() { + if common_parity_drives == 0 { + let parity_drives = ec_drives_no_config(pool_eps.drives_per_set)?; + storageclass::validate_parity(parity_drives, pool_eps.drives_per_set)?; + common_parity_drives = parity_drives; + } + + // validate_parity(parity_count, pool_eps.drives_per_set)?; + + // Initialize disks without health monitoring so that remote peers + // are not immediately marked as faulty before they have a chance to + // start up. Health monitoring is enabled after format loading succeeds. + let (disks, errs) = store_init::init_disks( + &pool_eps.endpoints, + &DiskOption { + cleanup: true, + health_check: false, + }, + ) + .await; + + check_disk_fatal_errs(&errs)?; + + let fm = { + let mut times = 0; + let mut interval = 1; + loop { + match store_init::connect_load_init_formats( + first_is_local, + &disks, + pool_eps.set_count, + pool_eps.drives_per_set, + deployment_id, + ) + .await + { + Ok(fm) => break Ok(fm), + // Wrap the final error if we are giving up + Err(e) if times >= 10 => { + break Err(Error::other(format!("can not get formats after {} retries, last error: {e}", times))); + } + // Retrying so just drop the error + Err(_) => {} + } + times += 1; + if interval < 16 { + interval *= 2; + } + info!("retrying get formats after {:?}", interval); + select! { + _ = tokio::signal::ctrl_c() => { + info!("got ctrl+c, exits"); + exit(0); + } + _ = sleep(Duration::from_secs(interval)) => { + } + } + } + }?; + + // Format loading succeeded, enable health monitoring on all disks + for disk in disks.iter().flatten() { + disk.enable_health_check(); + } + + if deployment_id.is_none() { + deployment_id = Some(fm.id); + } + + if deployment_id != Some(fm.id) { + return Err(Error::other("deployment_id not same in one pool")); + } + + if deployment_id.is_some() && deployment_id.unwrap().is_nil() { + deployment_id = Some(Uuid::new_v4()); + } + + for disk in disks.iter() { + if disk.is_some() && disk.as_ref().unwrap().is_local() { + local_disks.push(disk.as_ref().unwrap().clone()); + } + } + + let sets = Sets::new(disks.clone(), pool_eps, &fm, i, common_parity_drives).await?; + pools.push(sets); + + disk_map.insert(i, disks); + } + + // Replace the local disk + if !is_dist_erasure().await { + let mut global_local_disk_map = GLOBAL_LOCAL_DISK_MAP.write().await; + for disk in local_disks { + let path = disk.endpoint().to_string(); + global_local_disk_map.insert(path, Some(disk.clone())); + } + } + + let peer_sys = S3PeerSys::new(&endpoint_pools); + let mut pool_meta = PoolMeta::new(&pools, &PoolMeta::default()); + pool_meta.dont_save = true; + + let decommission_cancelers = vec![None; pools.len()]; + let ec = Arc::new(ECStore { + id: deployment_id.unwrap(), + disk_map, + pools, + peer_sys, + pool_meta: RwLock::new(pool_meta), + rebalance_meta: RwLock::new(None), + decommission_cancelers, + }); + + // Only set it when the global deployment ID is not yet configured + if let Some(dep_id) = deployment_id + && get_global_deployment_id().is_none() + { + set_global_deployment_id(dep_id); + } + + let wait_sec = 5; + let mut exit_count = 0; + loop { + if let Err(err) = ec.init(ctx.clone()).await { + error!("init err: {}", err); + error!("retry after {} second", wait_sec); + sleep(Duration::from_secs(wait_sec)).await; + + if exit_count > 10 { + return Err(Error::other("ec init failed")); + } + + exit_count += 1; + + continue; + } + + break; + } + + set_object_layer(ec.clone()).await; + + Ok(ec) + } + + #[instrument(level = "debug", skip(self, rx))] + pub async fn init(self: &Arc, rx: CancellationToken) -> Result<()> { + GLOBAL_BOOT_TIME.get_or_init(|| async { SystemTime::now() }).await; + + if self.load_rebalance_meta().await.is_ok() { + self.start_rebalance().await; + } + + let mut meta = PoolMeta::default(); + meta.load(self.pools[0].clone(), self.pools.clone()).await?; + let update = meta.validate(self.pools.clone())?; + + if !update { + { + let mut pool_meta = self.pool_meta.write().await; + *pool_meta = meta.clone(); + } + } else { + let new_meta = PoolMeta::new(&self.pools, &meta); + new_meta.save(self.pools.clone()).await?; + { + let mut pool_meta = self.pool_meta.write().await; + *pool_meta = new_meta; + } + } + + let pools = meta.return_resumable_pools(); + let mut pool_indices = Vec::with_capacity(pools.len()); + + let endpoints = get_global_endpoints(); + + for p in pools.iter() { + if let Some(idx) = endpoints.get_pool_idx(&p.cmd_line) { + pool_indices.push(idx); + } else { + return Err(Error::other(format!( + "unexpected state present for decommission status pool({}) not found", + p.cmd_line + ))); + } + } + + if !pool_indices.is_empty() { + let idx = pool_indices[0]; + if endpoints.as_ref()[idx].endpoints.as_ref()[0].is_local { + let store = self.clone(); + + tokio::spawn(async move { + // wait 3 minutes for cluster init + tokio::time::sleep(Duration::from_secs(60 * 3)).await; + + if let Err(err) = store.decommission(rx.clone(), pool_indices.clone()).await { + if err == StorageError::DecommissionAlreadyRunning { + for i in pool_indices.iter() { + store.do_decommission_in_routine(rx.clone(), *i).await; + } + return; + } + + error!("store init decommission err: {}", err); + + // TODO: check config err + } + }); + } + } + + let num_nodes = get_global_endpoints().get_nodes().len() as u64; + init_global_bucket_monitor(num_nodes); + + init_background_expiry(self.clone()).await; + + TransitionState::init(self.clone()).await; + + if let Err(err) = GLOBAL_TierConfigMgr.write().await.init(self.clone()).await { + info!("TierConfigMgr init error: {}", err); + } + + Ok(()) + } + + pub fn init_local_disks() {} + + pub fn single_pool(&self) -> bool { + self.pools.len() == 1 + } +} diff --git a/crates/ecstore/src/store/list.rs b/crates/ecstore/src/store/list.rs new file mode 100644 index 000000000..ec4ccb38e --- /dev/null +++ b/crates/ecstore/src/store/list.rs @@ -0,0 +1,68 @@ +// 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 super::*; + +impl ECStore { + #[instrument(skip(self))] + #[allow(clippy::too_many_arguments)] + pub(super) async fn handle_list_objects_v2( + self: Arc, + bucket: &str, + prefix: &str, + continuation_token: Option, + delimiter: Option, + max_keys: i32, + fetch_owner: bool, + start_after: Option, + incl_deleted: bool, + ) -> Result { + self.inner_list_objects_v2( + bucket, + prefix, + continuation_token, + delimiter, + max_keys, + fetch_owner, + start_after, + incl_deleted, + ) + .await + } + + #[instrument(skip(self))] + pub(super) async fn handle_list_object_versions( + self: Arc, + bucket: &str, + prefix: &str, + marker: Option, + version_marker: Option, + delimiter: Option, + max_keys: i32, + ) -> Result { + self.inner_list_object_versions(bucket, prefix, marker, version_marker, delimiter, max_keys) + .await + } + + pub(super) async fn handle_walk( + self: Arc, + rx: CancellationToken, + bucket: &str, + prefix: &str, + result: tokio::sync::mpsc::Sender, + opts: WalkOptions, + ) -> Result<()> { + self.walk_internal(rx, bucket, prefix, result, opts).await + } +} diff --git a/crates/ecstore/src/store/multipart.rs b/crates/ecstore/src/store/multipart.rs new file mode 100644 index 000000000..dfeb1c143 --- /dev/null +++ b/crates/ecstore/src/store/multipart.rs @@ -0,0 +1,326 @@ +// 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 super::*; + +impl ECStore { + #[instrument(skip(self))] + pub(super) async fn handle_list_object_parts( + &self, + bucket: &str, + object: &str, + upload_id: &str, + part_number_marker: Option, + max_parts: usize, + opts: &ObjectOptions, + ) -> Result { + check_list_parts_args(bucket, object, upload_id)?; + + // TODO: nslock + + if self.single_pool() { + return self.pools[0] + .list_object_parts(bucket, object, upload_id, part_number_marker, max_parts, opts) + .await; + } + + for pool in self.pools.iter() { + if self.is_suspended(pool.pool_idx).await { + continue; + } + return match pool + .list_object_parts(bucket, object, upload_id, part_number_marker, max_parts, opts) + .await + { + Ok(res) => Ok(res), + Err(err) => { + if is_err_invalid_upload_id(&err) { + continue; + } + Err(err) + } + }; + } + + Err(StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned())) + } + + #[instrument(skip(self))] + pub(super) async fn handle_list_multipart_uploads( + &self, + bucket: &str, + prefix: &str, + key_marker: Option, + upload_id_marker: Option, + delimiter: Option, + max_uploads: usize, + ) -> Result { + check_list_multipart_args(bucket, prefix, &key_marker, &upload_id_marker, &delimiter)?; + + if prefix.is_empty() { + // TODO: return from cache + } + + if self.single_pool() { + return self.pools[0] + .list_multipart_uploads(bucket, prefix, key_marker, upload_id_marker, delimiter, max_uploads) + .await; + } + + let mut uploads = Vec::new(); + + for pool in self.pools.iter() { + if self.is_suspended(pool.pool_idx).await { + continue; + } + let res = pool + .list_multipart_uploads( + bucket, + prefix, + key_marker.clone(), + upload_id_marker.clone(), + delimiter.clone(), + max_uploads, + ) + .await?; + uploads.extend(res.uploads); + } + + Ok(ListMultipartsInfo { + key_marker, + upload_id_marker, + max_uploads, + uploads, + prefix: prefix.to_owned(), + delimiter: delimiter.to_owned(), + ..Default::default() + }) + } + + #[instrument(skip(self))] + pub(super) async fn handle_new_multipart_upload( + &self, + bucket: &str, + object: &str, + opts: &ObjectOptions, + ) -> Result { + check_new_multipart_args(bucket, object)?; + + if self.single_pool() { + return self.pools[0].new_multipart_upload(bucket, object, opts).await; + } + + for (idx, pool) in self.pools.iter().enumerate() { + if self.is_suspended(idx).await || self.is_pool_rebalancing(idx).await { + continue; + } + let res = pool + .list_multipart_uploads(bucket, object, None, None, None, MAX_UPLOADS_LIST) + .await?; + + if !res.uploads.is_empty() { + return self.pools[idx].new_multipart_upload(bucket, object, opts).await; + } + } + let idx = self.get_pool_idx(bucket, object, -1).await?; + if opts.data_movement && idx == opts.src_pool_idx { + return Err(StorageError::DataMovementOverwriteErr( + bucket.to_owned(), + object.to_owned(), + "".to_owned(), + )); + } + + self.pools[idx].new_multipart_upload(bucket, object, opts).await + } + + #[instrument(skip(self))] + #[allow(clippy::too_many_arguments)] + pub(super) async fn handle_copy_object_part( + &self, + src_bucket: &str, + src_object: &str, + _dst_bucket: &str, + _dst_object: &str, + _upload_id: &str, + _part_id: usize, + _start_offset: i64, + _length: i64, + _src_info: &ObjectInfo, + _src_opts: &ObjectOptions, + _dst_opts: &ObjectOptions, + ) -> Result<()> { + check_new_multipart_args(src_bucket, src_object)?; + + // TODO: PutObjectReader + // self.put_object_part(dst_bucket, dst_object, upload_id, part_id, data, opts) + + Err(StorageError::NotImplemented) + } + + #[instrument(skip(self, data))] + pub(super) async fn handle_put_object_part( + &self, + bucket: &str, + object: &str, + upload_id: &str, + part_id: usize, + data: &mut PutObjReader, + opts: &ObjectOptions, + ) -> Result { + check_put_object_part_args(bucket, object, upload_id)?; + + if self.single_pool() { + return self.pools[0] + .put_object_part(bucket, object, upload_id, part_id, data, opts) + .await; + } + + for pool in self.pools.iter() { + if self.is_suspended(pool.pool_idx).await { + continue; + } + let err = match pool.put_object_part(bucket, object, upload_id, part_id, data, opts).await { + Ok(res) => return Ok(res), + Err(err) => { + if is_err_invalid_upload_id(&err) { + None + } else { + Some(err) + } + } + }; + + if let Some(err) = err { + error!("put_object_part err: {:?}", err); + return Err(err); + } + } + + Err(StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned())) + } + + #[instrument(skip(self))] + pub(super) async fn handle_get_multipart_info( + &self, + bucket: &str, + object: &str, + upload_id: &str, + opts: &ObjectOptions, + ) -> Result { + check_list_parts_args(bucket, object, upload_id)?; + if self.single_pool() { + return self.pools[0].get_multipart_info(bucket, object, upload_id, opts).await; + } + + for pool in self.pools.iter() { + if self.is_suspended(pool.pool_idx).await { + continue; + } + + return match pool.get_multipart_info(bucket, object, upload_id, opts).await { + Ok(res) => Ok(res), + Err(err) => { + if is_err_invalid_upload_id(&err) { + continue; + } + + Err(err) + } + }; + } + + Err(StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned())) + } + + #[instrument(skip(self))] + pub(super) async fn handle_abort_multipart_upload( + &self, + bucket: &str, + object: &str, + upload_id: &str, + opts: &ObjectOptions, + ) -> Result<()> { + check_abort_multipart_args(bucket, object, upload_id)?; + + // TODO: defer DeleteUploadID + + if self.single_pool() { + return self.pools[0].abort_multipart_upload(bucket, object, upload_id, opts).await; + } + + for pool in self.pools.iter() { + if self.is_suspended(pool.pool_idx).await { + continue; + } + + let err = match pool.abort_multipart_upload(bucket, object, upload_id, opts).await { + Ok(_) => return Ok(()), + Err(err) => { + // + if is_err_invalid_upload_id(&err) { None } else { Some(err) } + } + }; + + if let Some(er) = err { + return Err(er); + } + } + + Err(StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned())) + } + + #[instrument(skip(self))] + pub(super) async fn handle_complete_multipart_upload( + self: Arc, + bucket: &str, + object: &str, + upload_id: &str, + uploaded_parts: Vec, + opts: &ObjectOptions, + ) -> Result { + check_complete_multipart_args(bucket, object, upload_id)?; + + if self.single_pool() { + return self.pools[0] + .clone() + .complete_multipart_upload(bucket, object, upload_id, uploaded_parts, opts) + .await; + } + + for pool in self.pools.iter() { + if self.is_suspended(pool.pool_idx).await { + continue; + } + + let pool = pool.clone(); + let err = match pool + .complete_multipart_upload(bucket, object, upload_id, uploaded_parts.clone(), opts) + .await + { + Ok(res) => return Ok(res), + Err(err) => { + // + if is_err_invalid_upload_id(&err) { None } else { Some(err) } + } + }; + + if let Some(er) = err { + return Err(er); + } + } + + Err(StorageError::InvalidUploadID(bucket.to_owned(), object.to_owned(), upload_id.to_owned())) + } +} diff --git a/crates/ecstore/src/store/object.rs b/crates/ecstore/src/store/object.rs new file mode 100644 index 000000000..9d56bde1a --- /dev/null +++ b/crates/ecstore/src/store/object.rs @@ -0,0 +1,567 @@ +// 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 super::*; + +impl ECStore { + #[instrument(level = "debug", skip(self))] + pub(super) async fn handle_get_object_reader( + &self, + bucket: &str, + object: &str, + range: Option, + h: HeaderMap, + opts: &ObjectOptions, + ) -> Result { + check_get_obj_args(bucket, object)?; + + let object = encode_dir_object(object); + + if self.single_pool() { + return self.pools[0].get_object_reader(bucket, object.as_str(), range, h, opts).await; + } + + // TODO: nslock + + let mut opts = opts.clone(); + + opts.no_lock = true; + + // TODO: check if DeleteMarker + let (_oi, idx) = self.get_latest_object_info_with_idx(bucket, &object, &opts).await?; + + self.pools[idx] + .get_object_reader(bucket, object.as_str(), range, h, &opts) + .await + } + + #[instrument(level = "debug", skip(self, data))] + pub(super) async fn handle_put_object( + &self, + bucket: &str, + object: &str, + data: &mut PutObjReader, + opts: &ObjectOptions, + ) -> Result { + check_put_object_args(bucket, object)?; + + let object = encode_dir_object(object); + + if self.single_pool() { + return self.pools[0].put_object(bucket, object.as_str(), data, opts).await; + } + + let idx = self.get_pool_idx(bucket, &object, data.size()).await?; + + if opts.data_movement && idx == opts.src_pool_idx { + return Err(StorageError::DataMovementOverwriteErr( + bucket.to_owned(), + object.to_owned(), + opts.version_id.clone().unwrap_or_default(), + )); + } + + self.pools[idx].put_object(bucket, &object, data, opts).await + } + + #[instrument(skip(self))] + pub(super) async fn handle_get_object_info(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { + check_object_args(bucket, object)?; + + let object = encode_dir_object(object); + + if self.single_pool() { + return self.pools[0].get_object_info(bucket, object.as_str(), opts).await; + } + + // TODO: nslock + + let (info, _) = self.get_latest_object_info_with_idx(bucket, object.as_str(), opts).await?; + + opts.precondition_check(&info)?; + Ok(info) + } + + #[instrument(skip(self))] + #[allow(clippy::too_many_arguments)] + pub(super) async fn handle_copy_object( + &self, + src_bucket: &str, + src_object: &str, + dst_bucket: &str, + dst_object: &str, + src_info: &mut ObjectInfo, + src_opts: &ObjectOptions, + dst_opts: &ObjectOptions, + ) -> Result { + check_copy_obj_args(src_bucket, src_object)?; + check_copy_obj_args(dst_bucket, dst_object)?; + + let src_object = encode_dir_object(src_object); + let dst_object = encode_dir_object(dst_object); + + let cp_src_dst_same = path_join_buf(&[src_bucket, &src_object]) == path_join_buf(&[dst_bucket, &dst_object]); + + // TODO: nslock + + let pool_idx = self.get_pool_idx_no_lock(src_bucket, &src_object, src_info.size).await?; + + if cp_src_dst_same { + if let (Some(src_vid), Some(dst_vid)) = (&src_opts.version_id, &dst_opts.version_id) + && src_vid == dst_vid + { + return self.pools[pool_idx] + .copy_object(src_bucket, &src_object, dst_bucket, &dst_object, src_info, src_opts, dst_opts) + .await; + } + + if !dst_opts.versioned && src_opts.version_id.is_none() { + return self.pools[pool_idx] + .copy_object(src_bucket, &src_object, dst_bucket, &dst_object, src_info, src_opts, dst_opts) + .await; + } + + if dst_opts.versioned && src_opts.version_id != dst_opts.version_id { + src_info.version_only = true; + return self.pools[pool_idx] + .copy_object(src_bucket, &src_object, dst_bucket, &dst_object, src_info, src_opts, dst_opts) + .await; + } + } + + let put_opts = ObjectOptions { + user_defined: src_info.user_defined.clone(), + versioned: dst_opts.versioned, + version_id: dst_opts.version_id.clone(), + no_lock: true, + mod_time: dst_opts.mod_time, + ..Default::default() + }; + + if let Some(put_object_reader) = src_info.put_object_reader.as_mut() { + return self.pools[pool_idx] + .put_object(dst_bucket, &dst_object, put_object_reader, &put_opts) + .await; + } + + Err(StorageError::InvalidArgument( + src_bucket.to_owned(), + src_object.to_owned(), + "put_object_reader is none".to_owned(), + )) + } + + #[instrument(skip(self))] + pub(super) async fn handle_delete_object(&self, bucket: &str, object: &str, opts: ObjectOptions) -> Result { + check_del_obj_args(bucket, object)?; + + if opts.delete_prefix { + self.delete_prefix(bucket, object).await?; + return Ok(ObjectInfo::default()); + } + + // TODO: nslock + + let object = encode_dir_object(object); + let object = object.as_str(); + + let mut gopts = opts.clone(); + gopts.no_lock = true; + + // Determine which pool contains it + let (mut pinfo, errs) = self + .get_pool_info_existing_with_opts(bucket, object, &gopts) + .await + .map_err(|e| { + if is_err_read_quorum(&e) { + StorageError::ErasureWriteQuorum + } else { + e + } + })?; + + if pinfo.object_info.delete_marker && opts.version_id.is_none() { + pinfo.object_info.name = decode_dir_object(object); + return Ok(pinfo.object_info); + } + + if opts.data_movement && opts.src_pool_idx == pinfo.index { + return Err(StorageError::DataMovementOverwriteErr( + bucket.to_owned(), + object.to_owned(), + opts.version_id.unwrap_or_default(), + )); + } + + if opts.data_movement { + let mut obj = self.pools[pinfo.index].delete_object(bucket, object, opts).await?; + obj.name = decode_dir_object(obj.name.as_str()); + return Ok(obj); + } + + if !errs.is_empty() && !opts.versioned && !opts.version_suspended { + return self.delete_object_from_all_pools(bucket, object, &opts, errs).await; + } + + for pool in self.pools.iter() { + match pool.delete_object(bucket, object, opts.clone()).await { + Ok(res) => { + let mut obj = res; + obj.name = decode_dir_object(object); + return Ok(obj); + } + Err(err) => { + if !is_err_object_not_found(&err) && !is_err_version_not_found(&err) { + return Err(err); + } + } + } + } + + if let Some(ver) = opts.version_id { + return Err(StorageError::VersionNotFound(bucket.to_owned(), object.to_owned(), ver)); + } + + Err(StorageError::ObjectNotFound(bucket.to_owned(), object.to_owned())) + } + + #[instrument(skip(self))] + pub(super) async fn handle_delete_objects( + &self, + bucket: &str, + objects: Vec, + opts: ObjectOptions, + ) -> (Vec, Vec>) { + // encode object name + let objects: Vec = objects + .iter() + .map(|v| { + let mut v = v.clone(); + v.object_name = encode_dir_object(v.object_name.as_str()); + v + }) + .collect(); + + // Default return value + let mut del_objects = vec![DeletedObject::default(); objects.len()]; + + let mut del_errs = Vec::with_capacity(objects.len()); + for _ in 0..objects.len() { + del_errs.push(None) + } + + // TODO: nslock + + let mut futures = Vec::with_capacity(self.pools.len()); + + for pool in self.pools.iter() { + futures.push(pool.delete_objects(bucket, objects.clone(), opts.clone())); + } + + let results = join_all(futures).await; + + for idx in 0..del_objects.len() { + for (dels, errs) in results.iter() { + if errs[idx].is_none() && dels[idx].found { + del_errs[idx] = None; + del_objects[idx] = dels[idx].clone(); + break; + } + + if del_errs[idx].is_none() { + del_errs[idx] = errs[idx].clone(); + del_objects[idx] = dels[idx].clone(); + } + } + } + + del_objects.iter_mut().for_each(|v| { + v.object_name = decode_dir_object(&v.object_name); + }); + + (del_objects, del_errs) + + // let mut futures = Vec::with_capacity(objects.len()); + + // for obj in objects.iter() { + // futures.push(async move { + // self.internal_get_pool_info_existing_with_opts( + // bucket, + // &obj.object_name, + // &ObjectOptions { + // no_lock: true, + // ..Default::default() + // }, + // ) + // .await + // }); + // } + + // let results = join_all(futures).await; + + // // let mut jhs = Vec::new(); + // // let semaphore = Arc::new(Semaphore::new(num_cpus::get())); + // // let pools = Arc::new(self.pools.clone()); + + // // for obj in objects.iter() { + // // let (semaphore, pools, bucket, object_name, opt) = ( + // // semaphore.clone(), + // // pools.clone(), + // // bucket.to_string(), + // // obj.object_name.to_string(), + // // ObjectOptions::default(), + // // ); + + // // let jh = tokio::spawn(async move { + // // let _permit = semaphore.acquire().await.unwrap(); + // // self.internal_get_pool_info_existing_with_opts(pools.as_ref(), &bucket, &object_name, &opt) + // // .await + // // }); + // // jhs.push(jh); + // // } + // // let mut results = Vec::new(); + // // for jh in jhs { + // // results.push(jh.await.unwrap()); + // // } + + // // Record the mapping pool_idx -> object index + // let mut pool_obj_idx_map = HashMap::new(); + // let mut orig_index_map = HashMap::new(); + + // for (i, res) in results.into_iter().enumerate() { + // match res { + // Ok((pinfo, _)) => { + // if let Some(obj) = objects.get(i) { + // if pinfo.object_info.delete_marker && obj.version_id.is_none() { + // del_objects[i] = DeletedObject { + // delete_marker: pinfo.object_info.delete_marker, + // delete_marker_version_id: pinfo.object_info.version_id.map(|v| v.to_string()), + // object_name: decode_dir_object(&pinfo.object_info.name), + // delete_marker_mtime: pinfo.object_info.mod_time, + // ..Default::default() + // }; + // continue; + // } + + // if !pool_obj_idx_map.contains_key(&pinfo.index) { + // pool_obj_idx_map.insert(pinfo.index, vec![obj.clone()]); + // } else if let Some(val) = pool_obj_idx_map.get_mut(&pinfo.index) { + // val.push(obj.clone()); + // } + + // if !orig_index_map.contains_key(&pinfo.index) { + // orig_index_map.insert(pinfo.index, vec![i]); + // } else if let Some(val) = orig_index_map.get_mut(&pinfo.index) { + // val.push(i); + // } + // } + // } + // Err(e) => { + // if !is_err_object_not_found(&e) && is_err_version_not_found(&e) { + // del_errs[i] = Some(e) + // } + + // if let Some(obj) = objects.get(i) { + // del_objects[i] = DeletedObject { + // object_name: decode_dir_object(&obj.object_name), + // version_id: obj.version_id.map(|v| v.to_string()), + // ..Default::default() + // } + // } + // } + // } + // } + + // if !pool_obj_idx_map.is_empty() { + // for (i, sets) in self.pools.iter().enumerate() { + // // Retrieve the object index for a pool idx + // if let Some(objs) = pool_obj_idx_map.get(&i) { + // // Fetch the corresponding object (should never be None) + // // let objs: Vec = obj_idxs.iter().filter_map(|&idx| objects.get(idx).cloned()).collect(); + + // if objs.is_empty() { + // continue; + // } + + // let (pdel_objs, perrs) = sets.delete_objects(bucket, objs.clone(), opts.clone()).await?; + + // // Insert simultaneously (should never be None) + // let org_indexes = orig_index_map.get(&i).unwrap(); + + // // perrs should follow the same order as obj_idxs + // for (i, err) in perrs.into_iter().enumerate() { + // let obj_idx = org_indexes[i]; + + // if err.is_some() { + // del_errs[obj_idx] = err; + // } + + // let mut dobj = pdel_objs.get(i).unwrap().clone(); + // dobj.object_name = decode_dir_object(&dobj.object_name); + + // del_objects[obj_idx] = dobj; + // } + // } + // } + // } + + // Ok((del_objects, del_errs)) + } + + #[instrument(skip(self))] + pub(super) async fn handle_add_partial(&self, bucket: &str, object: &str, version_id: &str) -> Result<()> { + let object = encode_dir_object(object); + + if self.single_pool() { + let _ = self.pools[0].add_partial(bucket, object.as_str(), version_id).await; + return Ok(()); + } + + let idx = self + .get_pool_idx_existing_with_opts(bucket, object.as_str(), &ObjectOptions::default()) + .await?; + + let _ = self.pools[idx].add_partial(bucket, object.as_str(), version_id).await; + Ok(()) + } + + #[instrument(skip(self))] + pub(super) async fn handle_transition_object(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<()> { + let object = encode_dir_object(object); + if self.single_pool() { + return self.pools[0].transition_object(bucket, &object, opts).await; + } + + //opts.skip_decommissioned = true; + //opts.no_lock = true; + let idx = self.get_pool_idx_existing_with_opts(bucket, &object, opts).await?; + + self.pools[idx].transition_object(bucket, &object, opts).await + } + + #[instrument(skip(self))] + pub(super) async fn handle_restore_transitioned_object( + self: Arc, + bucket: &str, + object: &str, + opts: &ObjectOptions, + ) -> Result<()> { + let object = encode_dir_object(object); + if self.single_pool() { + return self.pools[0].clone().restore_transitioned_object(bucket, &object, opts).await; + } + + //opts.skip_decommissioned = true; + //opts.nolock = true; + let idx = self.get_pool_idx_existing_with_opts(bucket, object.as_str(), opts).await?; + + self.pools[idx] + .clone() + .restore_transitioned_object(bucket, &object, opts) + .await + } + + #[instrument(skip(self))] + pub(super) async fn handle_put_object_metadata( + &self, + bucket: &str, + object: &str, + opts: &ObjectOptions, + ) -> Result { + let object = encode_dir_object(object); + if self.single_pool() { + return self.pools[0].put_object_metadata(bucket, object.as_str(), opts).await; + } + + let mut opts = opts.clone(); + opts.metadata_chg = true; + + let idx = self.get_pool_idx_existing_with_opts(bucket, object.as_str(), &opts).await?; + + self.pools[idx].put_object_metadata(bucket, object.as_str(), &opts).await + } + + #[instrument(skip(self))] + pub(super) async fn handle_get_object_tags(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { + let object = encode_dir_object(object); + + if self.single_pool() { + return self.pools[0].get_object_tags(bucket, object.as_str(), opts).await; + } + + let (oi, _) = self.get_latest_object_info_with_idx(bucket, &object, opts).await?; + + Ok(oi.user_tags) + } + + #[instrument(level = "debug", skip(self))] + pub(super) async fn handle_put_object_tags( + &self, + bucket: &str, + object: &str, + tags: &str, + opts: &ObjectOptions, + ) -> Result { + let object = encode_dir_object(object); + + if self.single_pool() { + return self.pools[0].put_object_tags(bucket, object.as_str(), tags, opts).await; + } + + let idx = self.get_pool_idx_existing_with_opts(bucket, object.as_str(), opts).await?; + + self.pools[idx].put_object_tags(bucket, object.as_str(), tags, opts).await + } + + #[instrument(skip(self))] + pub(super) async fn handle_delete_object_version( + &self, + bucket: &str, + object: &str, + fi: &FileInfo, + force_del_marker: bool, + ) -> Result<()> { + check_del_obj_args(bucket, object)?; + + let object = encode_dir_object(object); + + if self.single_pool() { + return self.pools[0] + .delete_object_version(bucket, object.as_str(), fi, force_del_marker) + .await; + } + Err(StorageError::NotImplemented) + } + + #[instrument(skip(self))] + pub(super) async fn handle_delete_object_tags(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result { + let object = encode_dir_object(object); + + if self.single_pool() { + return self.pools[0].delete_object_tags(bucket, object.as_str(), opts).await; + } + + let idx = self.get_pool_idx_existing_with_opts(bucket, object.as_str(), opts).await?; + + self.pools[idx].delete_object_tags(bucket, object.as_str(), opts).await + } + + pub(super) async fn handle_verify_object_integrity(&self, bucket: &str, object: &str, opts: &ObjectOptions) -> Result<()> { + let get_object_reader = ::get_object_reader(self, bucket, object, None, HeaderMap::new(), opts).await?; + // Stream to sink to avoid loading entire object into memory during verification + let mut reader = get_object_reader.stream; + tokio::io::copy(&mut reader, &mut tokio::io::sink()).await?; + Ok(()) + } +} diff --git a/crates/ecstore/src/store/peer.rs b/crates/ecstore/src/store/peer.rs new file mode 100644 index 000000000..9c33e683b --- /dev/null +++ b/crates/ecstore/src/store/peer.rs @@ -0,0 +1,213 @@ +// 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 super::*; + +pub async fn find_local_disk(disk_path: &String) -> Option { + let disk_map = GLOBAL_LOCAL_DISK_MAP.read().await; + + if let Some(disk) = disk_map.get(disk_path) { + disk.as_ref().cloned() + } else { + None + } +} + +pub async fn get_disk_via_endpoint(endpoint: &Endpoint) -> Option { + let global_set_drives = GLOBAL_LOCAL_DISK_SET_DRIVES.read().await; + if global_set_drives.is_empty() { + return GLOBAL_LOCAL_DISK_MAP + .read() + .await + .get(&endpoint.to_string()) + .cloned() + .unwrap_or(None); + } + global_set_drives + .get(endpoint.pool_idx as usize) + .and_then(|sets| sets.get(endpoint.set_idx as usize)) + .and_then(|disks| disks.get(endpoint.disk_idx as usize)) + .cloned() + .unwrap_or(None) +} + +pub async fn all_local_disk_path() -> Vec { + let disk_map = GLOBAL_LOCAL_DISK_MAP.read().await; + disk_map.keys().cloned().collect() +} + +pub async fn all_local_disk() -> Vec { + let disk_map = GLOBAL_LOCAL_DISK_MAP.read().await; + disk_map + .values() + .filter(|v| v.is_some()) + .map(|v| v.as_ref().unwrap().clone()) + .collect() +} + +pub async fn init_local_disks(endpoint_pools: EndpointServerPools) -> Result<()> { + let opt = &DiskOption { + cleanup: true, + health_check: true, + }; + + let mut global_set_drives = GLOBAL_LOCAL_DISK_SET_DRIVES.write().await; + for pool_eps in endpoint_pools.as_ref().iter() { + let mut set_count_drives = Vec::with_capacity(pool_eps.set_count); + for _ in 0..pool_eps.set_count { + set_count_drives.push(vec![None; pool_eps.drives_per_set]); + } + + global_set_drives.push(set_count_drives); + } + + let mut global_local_disk_map = GLOBAL_LOCAL_DISK_MAP.write().await; + + for pool_eps in endpoint_pools.as_ref().iter() { + for ep in pool_eps.endpoints.as_ref().iter() { + if !ep.is_local { + continue; + } + + let disk = new_disk(ep, opt).await?; + + let path = disk.endpoint().to_string(); + + global_local_disk_map.insert(path, Some(disk.clone())); + + global_set_drives[ep.pool_idx as usize][ep.set_idx as usize][ep.disk_idx as usize] = Some(disk.clone()); + } + } + + Ok(()) +} + +pub fn init_lock_clients(endpoint_pools: EndpointServerPools) { + let mut unique_endpoints: HashMap = HashMap::new(); + + for pool_eps in endpoint_pools.as_ref().iter() { + for ep in pool_eps.endpoints.as_ref().iter() { + unique_endpoints.insert(ep.host_port(), ep); + } + } + + let mut clients = HashMap::new(); + let mut first_local_client_set = false; + + for (key, endpoint) in unique_endpoints { + if endpoint.is_local { + let local_client = Arc::new(LocalClient::new()) as Arc; + + // Store the first LocalClient globally for use by other modules + if !first_local_client_set { + if let Err(e) = crate::global::set_global_lock_client(local_client.clone()) { + // If already set, ignore the error (another thread may have set it) + warn!("set_global_lock_client error: {:?}", e); + } else { + first_local_client_set = true; + } + } + + clients.insert(key, local_client); + } else { + clients.insert(key, Arc::new(RemoteClient::new(endpoint.url.to_string())) as Arc); + } + } + + // Store the lock clients map globally + if crate::global::set_global_lock_clients(clients).is_err() { + error!("init_lock_clients: error setting lock clients"); + } +} + +pub(super) async fn init_local_peer(endpoint_pools: &EndpointServerPools, host: &String, port: &String) { + let mut peer_set = Vec::new(); + endpoint_pools.as_ref().iter().for_each(|endpoints| { + endpoints.endpoints.as_ref().iter().for_each(|endpoint| { + if endpoint.get_type() == EndpointType::Url && endpoint.is_local && endpoint.url.has_host() { + peer_set.push(endpoint.url.host_str().unwrap().to_string()); + } + }); + }); + + if peer_set.is_empty() { + if !host.is_empty() { + *GLOBAL_LOCAL_NODE_NAME.write().await = format!("{host}:{port}"); + return; + } + + *GLOBAL_LOCAL_NODE_NAME.write().await = format!("127.0.0.1:{port}"); + return; + } + + *GLOBAL_LOCAL_NODE_NAME.write().await = peer_set[0].clone(); +} + +pub async fn get_disk_infos(disks: &[Option]) -> Vec> { + let opts = &DiskInfoOptions::default(); + let mut res = vec![None; disks.len()]; + for (idx, disk_op) in disks.iter().enumerate() { + if let Some(disk) = disk_op + && let Ok(info) = disk.disk_info(opts).await + { + res[idx] = Some(info); + } + } + + res +} + +pub async fn has_space_for(dis: &[Option], size: i64) -> Result { + let size = { if size < 0 { DISK_ASSUME_UNKNOWN_SIZE } else { size as u64 * 2 } }; + + let mut available = 0; + let mut total = 0; + let mut disks_num = 0; + + for disk in dis.iter().flatten() { + disks_num += 1; + total += disk.total; + available += disk.total - disk.used; + } + + if disks_num < dis.len() / 2 || disks_num == 0 { + return Err(Error::other(format!( + "not enough online disks to calculate the available space,need {}, found {}", + (dis.len() / 2) + 1, + disks_num, + ))); + } + + let per_disk = size / disks_num as u64; + + for disk in dis.iter().flatten() { + if !is_erasure_sd().await && disk.free_inodes < DISK_MIN_INODES && disk.used_inodes > 0 { + return Ok(false); + } + + if disk.free <= per_disk { + return Ok(false); + } + } + + if available < size { + return Ok(false); + } + + available -= size; + + let want = total as f64 * (1.0 - DISK_FILL_FRACTION); + + Ok(available > want as u64) +} diff --git a/crates/ecstore/src/store/rebalance.rs b/crates/ecstore/src/store/rebalance.rs new file mode 100644 index 000000000..1915f573d --- /dev/null +++ b/crates/ecstore/src/store/rebalance.rs @@ -0,0 +1,700 @@ +// 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 super::*; + +impl ECStore { + #[instrument(level = "debug", skip(self))] + pub(super) async fn delete_all(&self, bucket: &str, prefix: &str) -> Result<()> { + let mut futures = Vec::new(); + for sets in self.pools.iter() { + for set in sets.disk_set.iter() { + futures.push(set.delete_all(bucket, prefix)); + // let disks = set.disks.read().await; + // let dd = disks.clone(); + // for disk in dd { + // if disk.is_none() { + // continue; + // } + // // let disk = disk.as_ref().unwrap().clone(); + // // futures.push(disk.delete( + // // bucket, + // // prefix, + // // DeleteOptions { + // // recursive: true, + // // immediate: false, + // // }, + // // )); + // } + } + } + let results = join_all(futures).await; + + let mut errs = Vec::new(); + + for res in results { + match res { + Ok(_) => errs.push(None), + Err(e) => errs.push(Some(e)), + } + } + + debug!("store delete_all errs {:?}", errs); + + Ok(()) + } + + pub(super) async fn delete_prefix(&self, bucket: &str, object: &str) -> Result<()> { + for pool in self.pools.iter() { + pool.delete_object( + bucket, + object, + ObjectOptions { + delete_prefix: true, + ..Default::default() + }, + ) + .await?; + } + + Ok(()) + } + + async fn get_available_pool_idx(&self, bucket: &str, object: &str, size: i64) -> Option { + // // Return a random one first + + let mut server_pools = self.get_server_pools_available_space(bucket, object, size).await; + server_pools.filter_max_used(100 - (100_f64 * DISK_RESERVE_FRACTION) as u64); + let total = server_pools.total_available(); + + if total == 0 { + return None; + } + + let mut rng = rand::rng(); + let random_u64: u64 = rng.random_range(0..total); + + let choose = random_u64 % total; + let mut at_total = 0; + + for pool in server_pools.iter() { + at_total += pool.available; + if at_total > choose && pool.available > 0 { + return Some(pool.index); + } + } + + None + } + + async fn get_server_pools_available_space(&self, bucket: &str, object: &str, size: i64) -> ServerPoolsAvailableSpace { + let mut n_sets = vec![0; self.pools.len()]; + let mut infos = vec![Vec::new(); self.pools.len()]; + + // TODO: add concurrency + for (idx, pool) in self.pools.iter().enumerate() { + if self.is_suspended(idx).await || self.is_pool_rebalancing(idx).await { + continue; + } + + n_sets[idx] = pool.set_count; + + if let Ok(disks) = pool.get_disks_by_key(object).get_disks(0, 0).await { + let disk_infos = get_disk_infos(&disks).await; + infos[idx] = disk_infos; + } + } + + let mut server_pools = vec![PoolAvailableSpace::default(); self.pools.len()]; + for (i, zinfo) in infos.iter().enumerate() { + if zinfo.is_empty() { + server_pools[i] = PoolAvailableSpace { + index: i, + ..Default::default() + }; + + continue; + } + + if !is_meta_bucketname(bucket) && !has_space_for(zinfo, size).await.unwrap_or_default() { + server_pools[i] = PoolAvailableSpace { + index: i, + ..Default::default() + }; + + continue; + } + + let mut available = 0; + let mut max_used_pct = 0; + for disk in zinfo.iter().flatten() { + if disk.total == 0 { + continue; + } + + available += disk.total - disk.used; + + let pct_used = disk.used * 100 / disk.total; + + if pct_used > max_used_pct { + max_used_pct = pct_used; + } + } + + available *= n_sets[i] as u64; + + server_pools[i] = PoolAvailableSpace { + index: i, + available, + max_used_pct, + } + } + + ServerPoolsAvailableSpace(server_pools) + } + + pub(super) async fn is_suspended(&self, idx: usize) -> bool { + // TODO: LOCK + + let pool_meta = self.pool_meta.read().await; + + pool_meta.is_suspended(idx) + } + + pub(super) async fn get_pool_idx(&self, bucket: &str, object: &str, size: i64) -> Result { + let idx = match self + .get_pool_idx_existing_with_opts( + bucket, + object, + &ObjectOptions { + skip_decommissioned: true, + skip_rebalancing: true, + ..Default::default() + }, + ) + .await + { + Ok(res) => res, + Err(err) => { + if !is_err_object_not_found(&err) { + return Err(err); + } + + if let Some(hit_idx) = self.get_available_pool_idx(bucket, object, size).await { + hit_idx + } else { + return Err(Error::DiskFull); + } + } + }; + + Ok(idx) + } + + pub(super) async fn get_pool_idx_no_lock(&self, bucket: &str, object: &str, size: i64) -> Result { + let idx = match self.get_pool_idx_existing_no_lock(bucket, object).await { + Ok(res) => res, + Err(err) => { + if !is_err_object_not_found(&err) { + return Err(err); + } + + if let Some(idx) = self.get_available_pool_idx(bucket, object, size).await { + idx + } else { + warn!("get_pool_idx_no_lock: disk full {}/{}", bucket, object); + return Err(Error::DiskFull); + } + } + }; + + Ok(idx) + } + + async fn get_pool_idx_existing_no_lock(&self, bucket: &str, object: &str) -> Result { + self.get_pool_idx_existing_with_opts( + bucket, + object, + &ObjectOptions { + no_lock: true, + skip_decommissioned: true, + skip_rebalancing: true, + ..Default::default() + }, + ) + .await + } + + pub(super) async fn get_pool_idx_existing_with_opts( + &self, + bucket: &str, + object: &str, + opts: &ObjectOptions, + ) -> Result { + let (pinfo, _) = self.get_pool_info_existing_with_opts(bucket, object, opts).await?; + Ok(pinfo.index) + } + + pub(super) async fn get_pool_info_existing_with_opts( + &self, + bucket: &str, + object: &str, + opts: &ObjectOptions, + ) -> Result<(PoolObjInfo, Vec)> { + self.internal_get_pool_info_existing_with_opts(bucket, object, opts).await + } + + async fn internal_get_pool_info_existing_with_opts( + &self, + bucket: &str, + object: &str, + opts: &ObjectOptions, + ) -> Result<(PoolObjInfo, Vec)> { + let mut futures = Vec::new(); + for pool in self.pools.iter() { + let mut pool_opts = opts.clone(); + if !pool_opts.metadata_chg { + pool_opts.version_id = None; + } + + futures.push(async move { pool.get_object_info(bucket, object, &pool_opts).await }); + } + + let results = join_all(futures).await; + + let mut ress = Vec::new(); + + // join_all preserves the input order + for (i, res) in results.into_iter().enumerate() { + let index = i; + + match res { + Ok(r) => { + ress.push(PoolObjInfo { + index, + object_info: r, + err: None, + }); + } + Err(e) => { + ress.push(PoolObjInfo { + index, + err: Some(e), + ..Default::default() + }); + } + } + } + + ress.sort_by(|a, b| { + let at = a.object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH); + let bt = b.object_info.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH); + + bt.cmp(&at) + }); + + let mut def_pool = PoolObjInfo::default(); + let mut has_def_pool = false; + + for pinfo in ress.iter() { + if opts.skip_decommissioned && self.is_suspended(pinfo.index).await { + continue; + } + + if opts.skip_rebalancing && self.is_pool_rebalancing(pinfo.index).await { + continue; + } + + if pinfo.err.is_none() { + return Ok((pinfo.clone(), self.pools_with_object(&ress, opts).await)); + } + + let err = pinfo.err.as_ref().unwrap(); + + if err == &Error::ErasureReadQuorum && !opts.metadata_chg { + return Ok((pinfo.clone(), self.pools_with_object(&ress, opts).await)); + } + + def_pool = pinfo.clone(); + has_def_pool = true; + // https://docs.aws.amazon.com/AmazonS3/latest/userguide/conditional-deletes.html + if is_err_object_not_found(err) + && let Err(err) = opts.precondition_check(&pinfo.object_info) + { + return Err(err.clone()); + } + + if !is_err_object_not_found(err) && !is_err_version_not_found(err) { + return Err(err.clone()); + } + + if pinfo.object_info.delete_marker && !pinfo.object_info.name.is_empty() { + return Ok((pinfo.clone(), Vec::new())); + } + } + + if opts.replication_request && opts.delete_marker && has_def_pool { + return Ok((def_pool, Vec::new())); + } + + Err(Error::ObjectNotFound(bucket.to_owned(), object.to_owned())) + } + + async fn pools_with_object(&self, pools: &[PoolObjInfo], opts: &ObjectOptions) -> Vec { + let mut errs = Vec::new(); + + for pool in pools.iter() { + if opts.skip_decommissioned && self.is_suspended(pool.index).await { + continue; + } + + if opts.skip_rebalancing && self.is_pool_rebalancing(pool.index).await { + continue; + } + + if let Some(err) = &pool.err { + if err == &Error::ErasureReadQuorum { + errs.push(PoolErr { + index: Some(pool.index), + err: Some(Error::ErasureReadQuorum), + }); + } + } else { + errs.push(PoolErr { + index: Some(pool.index), + err: None, + }); + } + } + errs + } + + pub(super) async fn get_latest_object_info_with_idx( + &self, + bucket: &str, + object: &str, + opts: &ObjectOptions, + ) -> Result<(ObjectInfo, usize)> { + let mut futures = Vec::with_capacity(self.pools.len()); + for pool in self.pools.iter() { + futures.push(pool.get_object_info(bucket, object, opts)); + } + + let results = join_all(futures).await; + + struct IndexRes { + res: Option, + idx: usize, + err: Option, + } + + let mut idx_res = Vec::with_capacity(self.pools.len()); + + for (idx, result) in results.into_iter().enumerate() { + match result { + Ok(res) => { + idx_res.push(IndexRes { + res: Some(res), + idx, + err: None, + }); + } + Err(e) => { + idx_res.push(IndexRes { + res: None, + idx, + err: Some(e), + }); + } + } + } + + // TODO: test order + idx_res.sort_by(|a, b| { + let a_mod = if let Some(o1) = &a.res { + o1.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH) + } else { + OffsetDateTime::UNIX_EPOCH + }; + + let b_mod = if let Some(o2) = &b.res { + o2.mod_time.unwrap_or(OffsetDateTime::UNIX_EPOCH) + } else { + OffsetDateTime::UNIX_EPOCH + }; + + if a_mod == b_mod { + return if a.idx < b.idx { Ordering::Greater } else { Ordering::Less }; + } + + b_mod.cmp(&a_mod) + }); + + for res in idx_res.into_iter() { + if let Some(obj) = res.res { + return Ok((obj, res.idx)); + } + + if let Some(err) = res.err + && !is_err_object_not_found(&err) + && !is_err_version_not_found(&err) + { + return Err(err); + } + + // TODO: delete marker + } + + let object = decode_dir_object(object); + + if opts.version_id.is_none() { + Err(StorageError::ObjectNotFound(bucket.to_owned(), object.to_owned())) + } else { + Err(StorageError::VersionNotFound( + bucket.to_owned(), + object.to_owned(), + opts.version_id.clone().unwrap_or_default(), + )) + } + } + + pub(super) async fn delete_object_from_all_pools( + &self, + bucket: &str, + object: &str, + opts: &ObjectOptions, + errs: Vec, + ) -> Result { + let mut objs = Vec::new(); + let mut derrs = Vec::new(); + + for pe in errs.iter() { + if let Some(err) = &pe.err + && err == &StorageError::ErasureWriteQuorum + { + objs.push(None); + derrs.push(Some(StorageError::ErasureWriteQuorum)); + continue; + } + + if let Some(idx) = pe.index { + match self.pools[idx].delete_object(bucket, object, opts.clone()).await { + Ok(res) => { + objs.push(Some(res)); + + derrs.push(None); + } + Err(err) => { + objs.push(None); + derrs.push(Some(err)); + } + } + } + } + + if let Some(e) = &derrs[0] { + return Err(e.clone()); + } + + Ok(objs[0].as_ref().unwrap().clone()) + } + + pub async fn reload_pool_meta(&self) -> Result<()> { + let mut meta = PoolMeta::default(); + meta.load(self.pools[0].clone(), self.pools.clone()).await?; + + let mut pool_meta = self.pool_meta.write().await; + *pool_meta = meta; + // *self.pool_meta.write().unwrap() = meta; + Ok(()) + } + + /// Disk information deduplication function + /// + /// Use multiple field combinations to ensure uniqueness: + /// - endpoint (node address) + /// - drive_path (mount path) + /// - pool_index (pool index) + /// - set_index (Collection Index) + /// - disk_index (disk index) + pub(crate) fn deduplicate_disks(disks: Vec) -> Vec { + use std::collections::HashMap; + use std::collections::hash_map::Entry; + + let mut unique_disks: HashMap = HashMap::new(); + let mut duplicate_count = 0; + + for disk in disks { + let key = format!( + "{}|{}|p{}s{}d{}", + disk.endpoint, disk.drive_path, disk.pool_index, disk.set_index, disk.disk_index + ); + + match unique_disks.entry(key) { + Entry::Vacant(entry) => { + entry.insert(disk); + } + Entry::Occupied(_) => { + duplicate_count += 1; + } + } + } + + if duplicate_count > 0 { + debug!("Deduplicated {} duplicate disk entries", duplicate_count); + } + + unique_disks.into_values().collect() + } + + #[instrument(skip(self))] + pub(super) async fn handle_new_ns_lock(&self, bucket: &str, object: &str) -> Result { + self.pools[0].new_ns_lock(bucket, object).await + } + + #[instrument(skip(self))] + pub(super) async fn handle_backend_info(&self) -> rustfs_madmin::BackendInfo { + let (standard_sc_parity, rr_sc_parity) = { + if let Some(sc) = GLOBAL_STORAGE_CLASS.get() { + let sc_parity = sc + .get_parity_for_sc(storageclass::CLASS_STANDARD) + .or(Some(self.pools[0].default_parity_count)); + + let rrs_sc_parity = sc.get_parity_for_sc(storageclass::RRS); + + (sc_parity, rrs_sc_parity) + } else { + (Some(self.pools[0].default_parity_count), None) + } + }; + + let mut standard_sc_data = Vec::new(); + let mut rr_sc_data = Vec::new(); + let mut drives_per_set = Vec::new(); + let mut total_sets = Vec::new(); + + for (idx, set_count) in self.set_drive_counts().iter().enumerate() { + if let Some(sc_parity) = standard_sc_parity { + standard_sc_data.push(set_count - sc_parity); + } + if let Some(sc_parity) = rr_sc_parity { + rr_sc_data.push(set_count - sc_parity); + } + total_sets.push(self.pools[idx].set_count); + drives_per_set.push(*set_count); + } + + rustfs_madmin::BackendInfo { + backend_type: rustfs_madmin::BackendByte::Erasure, + online_disks: rustfs_madmin::BackendDisks::new(), + offline_disks: rustfs_madmin::BackendDisks::new(), + standard_sc_data, + standard_sc_parity, + rr_sc_data, + rr_sc_parity, + total_sets, + drives_per_set, + ..Default::default() + } + } + + #[instrument(skip(self))] + pub(super) async fn handle_storage_info(&self) -> rustfs_madmin::StorageInfo { + let Some(notification_sy) = get_global_notification_sys() else { + return rustfs_madmin::StorageInfo::default(); + }; + + let mut info = notification_sy.storage_info(self).await; + + // 🔧 Defensive deduplication: This protection mechanism is retained even if the upstream is fixed + let original_count = info.disks.len(); + info.disks = Self::deduplicate_disks(info.disks); + let final_count = info.disks.len(); + + if original_count != final_count { + warn!( + "Storage info deduplication: removed {} duplicate disk entries ({} -> {})", + original_count - final_count, + original_count, + final_count + ); + } + + info + } + + #[instrument(skip(self))] + pub(super) async fn handle_local_storage_info(&self) -> rustfs_madmin::StorageInfo { + let mut futures = Vec::with_capacity(self.pools.len()); + + for pool in self.pools.iter() { + futures.push(pool.local_storage_info()) + } + + let results = join_all(futures).await; + + let mut disks = Vec::new(); + + for res in results.into_iter() { + disks.extend_from_slice(&res.disks); + } + + // 🔧 Defensive deduplication: when aggregating disks from all pools, drop duplicate + // entries that may be reported multiple times by backends; this extra layer is kept + // even if the upstream reporting is later fixed. + let original_count = disks.len(); + disks = Self::deduplicate_disks(disks); + + if original_count != disks.len() { + warn!("Local storage info deduplication: {} -> {}", original_count, disks.len()); + } + + let backend = self.backend_info().await; + rustfs_madmin::StorageInfo { backend, disks } + } + + #[instrument(skip(self))] + pub(super) async fn handle_get_disks(&self, pool_idx: usize, set_idx: usize) -> Result>> { + if pool_idx < self.pools.len() && set_idx < self.pools[pool_idx].disk_set.len() { + self.pools[pool_idx].disk_set[set_idx].get_disks(0, 0).await + } else { + Err(Error::other(format!("pool idx {pool_idx}, set idx {set_idx}, not found"))) + } + } + + #[instrument(skip(self))] + pub(super) fn handle_set_drive_counts(&self) -> Vec { + let mut counts = vec![0; self.pools.len()]; + + for (i, pool) in self.pools.iter().enumerate() { + counts[i] = pool.set_drive_count(); + } + counts + } + + #[instrument(skip(self))] + pub(super) async fn handle_get_pool_and_set(&self, id: &str) -> Result<(Option, Option, Option)> { + for (pool_idx, pool) in self.pools.iter().enumerate() { + for (set_idx, set) in pool.format.erasure.sets.iter().enumerate() { + for (disk_idx, disk_id) in set.iter().enumerate() { + if disk_id.to_string() == id { + return Ok((Some(pool_idx), Some(set_idx), Some(disk_idx))); + } + } + } + } + + Err(Error::DiskNotFound) + } +}