diff --git a/Cargo.lock b/Cargo.lock index 953b6e14f..4d4c88789 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -52,6 +52,12 @@ dependencies = [ "memchr", ] +[[package]] +name = "android-tzdata" +version = "0.1.1" +source = "registry+https://github.com/rust-lang/crates.io-index" +checksum = "e999941b234f3131b00bc13c22d06e8c5ff726d1b6318ac7eb276997bbb4fef0" + [[package]] name = "android_system_properties" version = "0.1.5" @@ -374,7 +380,13 @@ version = "0.4.38" source = "registry+https://github.com/rust-lang/crates.io-index" checksum = "a21f936df1771bf62b77f047b726c4625ff2e8aa607c01ec06e5a05bd8463401" dependencies = [ + "android-tzdata", + "iana-time-zone", + "js-sys", "num-traits", + "serde", + "wasm-bindgen", + "windows-targets", ] [[package]] @@ -645,6 +657,7 @@ dependencies = [ "byteorder", "bytes", "bytesize", + "chrono", "common", "crc32fast", "flatbuffers", diff --git a/common/protos/src/generated/proto_gen/node_service.rs b/common/protos/src/generated/proto_gen/node_service.rs index 9dc50d8cc..42353e86c 100644 --- a/common/protos/src/generated/proto_gen/node_service.rs +++ b/common/protos/src/generated/proto_gen/node_service.rs @@ -1037,7 +1037,6 @@ pub struct LoadRebalanceMetaResponse { } #[derive(Clone, Copy, PartialEq, ::prost::Message)] pub struct LoadTransitionTierConfigRequest {} - #[derive(Clone, PartialEq, ::prost::Message)] pub struct LoadTransitionTierConfigResponse { #[prost(bool, tag = "1")] diff --git a/ecstore/Cargo.toml b/ecstore/Cargo.toml index a52874e5c..23c0ff875 100644 --- a/ecstore/Cargo.toml +++ b/ecstore/Cargo.toml @@ -14,6 +14,7 @@ backon.workspace = true blake2 = "0.10.6" bytes.workspace = true common.workspace = true +chrono = { version = "0.4.38", features = ["serde"] } reader.workspace = true glob = "0.3.1" thiserror.workspace = true diff --git a/ecstore/src/admin_server_info.rs b/ecstore/src/admin_server_info.rs index 8c140b6cc..d0224bd47 100644 --- a/ecstore/src/admin_server_info.rs +++ b/ecstore/src/admin_server_info.rs @@ -119,11 +119,11 @@ pub async fn get_local_server_property() -> ServerProperties { network.insert(node_name, ITEM_ONLINE.to_string()); continue; } - if !network.contains_key(&node_name) { + if let std::collections::hash_map::Entry::Vacant(e) = network.entry(node_name) { if is_server_resolvable(endpoint).await.is_err() { - network.insert(node_name, ITEM_OFFLINE.to_string()); + e.insert(ITEM_OFFLINE.to_string()); } else { - network.insert(node_name, ITEM_ONLINE.to_string()); + e.insert(ITEM_ONLINE.to_string()); } } } diff --git a/ecstore/src/bitrot.rs b/ecstore/src/bitrot.rs index 198c6a226..738fed600 100644 --- a/ecstore/src/bitrot.rs +++ b/ecstore/src/bitrot.rs @@ -15,7 +15,7 @@ use std::{ collections::HashMap, io::{Cursor, Read}, }; -use tracing::error; +use tracing::{error, info}; use tokio::{ spawn, @@ -219,7 +219,9 @@ pub fn bitrot_verify( if algo != BitrotAlgorithm::HighwayHash256S { let mut h = algo.new_hasher(); h.update(r.get_ref()); - if h.finalize() != want { + let hash = h.finalize(); + if hash != want { + info!("bitrot_verify except: {:?}, got: {:?}", want, hash); return Err(Error::new(DiskError::FileCorrupt)); } @@ -229,7 +231,11 @@ pub fn bitrot_verify( let mut hash_buf = vec![0; h.size()]; let mut left = want_size; - if left != bitrot_shard_file_size(part_size, shard_size, algo) { + if left != bitrot_shard_file_size(part_size, shard_size, algo.clone()) { + info!( + "bitrot_shard_file_size failed, left: {}, part_size: {}, shard_size: {}, algo: {:?}", + left, part_size, shard_size, algo + ); return Err(Error::new(DiskError::FileCorrupt)); } @@ -246,7 +252,9 @@ pub fn bitrot_verify( let read = r.read(&mut buf)?; h.update(buf); left -= read; + let hash = h.clone().finalize(); if h.clone().finalize() != hash_buf[0..n] { + info!("bitrot_verify except: {:?}, got: {:?}", hash_buf[0..n].to_vec(), hash); return Err(Error::new(DiskError::FileCorrupt)); } } diff --git a/ecstore/src/bucket/versioning_sys.rs b/ecstore/src/bucket/versioning_sys.rs index c62a829c9..c893b6fdb 100644 --- a/ecstore/src/bucket/versioning_sys.rs +++ b/ecstore/src/bucket/versioning_sys.rs @@ -6,6 +6,12 @@ use tracing::warn; pub struct BucketVersioningSys {} +impl Default for BucketVersioningSys { + fn default() -> Self { + Self::new() + } +} + impl BucketVersioningSys { pub fn new() -> Self { Self {} diff --git a/ecstore/src/cache_value/metacache_set.rs b/ecstore/src/cache_value/metacache_set.rs index 928849cf6..79601ac87 100644 --- a/ecstore/src/cache_value/metacache_set.rs +++ b/ecstore/src/cache_value/metacache_set.rs @@ -139,7 +139,7 @@ pub async fn list_path_raw(mut rx: B_Receiver, opts: ListPathRawOptions) - }); } - let errs: Vec> = Vec::with_capacity(readers.len()); + let errs: Vec> = vec![None; readers.len()]; loop { let mut current = MetaCacheEntry::default(); let (mut at_eof, mut has_err, mut agree) = (0, 0, 0); diff --git a/ecstore/src/disk/error.rs b/ecstore/src/disk/error.rs index 0bce48455..0902e5d05 100644 --- a/ecstore/src/disk/error.rs +++ b/ecstore/src/disk/error.rs @@ -419,15 +419,18 @@ pub fn convert_access_error(e: io::Error, per_err: DiskError) -> Error { } pub fn is_all_not_found(errs: &[Option]) -> bool { - for err in errs.iter().flatten() { - if let Some(err) = err.downcast_ref::() { - match err { - DiskError::FileNotFound | DiskError::VolumeNotFound | &DiskError::FileVersionNotFound => { - continue; + for err in errs.iter() { + if let Some(err) = err { + if let Some(err) = err.downcast_ref::() { + match err { + DiskError::FileNotFound | DiskError::VolumeNotFound | &DiskError::FileVersionNotFound => { + continue; + } + _ => return false, } - _ => return false, } } + return false; } !errs.is_empty() diff --git a/ecstore/src/disk/local.rs b/ecstore/src/disk/local.rs index 46dee48a6..40e05fb84 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -1172,6 +1172,7 @@ impl DiskAPI for LocalDisk { check_path_length(file_path.to_string_lossy().to_string().as_str())?; // TODO: writeAllDirect io.copy + info!("file_path: {:?}", file_path); if let Some(parent) = file_path.parent() { os::make_dir_all(parent, &volume_dir).await?; } @@ -1374,6 +1375,7 @@ impl DiskAPI for LocalDisk { let src_volume_dir = self.get_bucket_path(src_volume)?; if !skip_access_checks(src_volume) { if let Err(e) = utils::fs::access(&src_volume_dir).await { + info!("access checks failed, src_volume_dir: {:?}, err: {}", src_volume_dir, e.to_string()); return Err(convert_access_error(e, DiskError::VolumeAccessDenied)); } } @@ -1381,6 +1383,7 @@ impl DiskAPI for LocalDisk { let dst_volume_dir = self.get_bucket_path(dst_volume)?; if !skip_access_checks(dst_volume) { if let Err(e) = utils::fs::access(&dst_volume_dir).await { + info!("access checks failed, dst_volume_dir: {:?}, err: {}", dst_volume_dir, e.to_string()); return Err(convert_access_error(e, DiskError::VolumeAccessDenied)); } } @@ -1429,6 +1432,7 @@ impl DiskAPI for LocalDisk { return Err(os_err_to_file_err(e)); } + info!("read xl.meta failed, dst_file_path: {:?}, err: {:?}", dst_file_path, e); None } }; @@ -1490,13 +1494,15 @@ impl DiskAPI for LocalDisk { err } })?; - if let Some((src_data_path, dst_data_path)) = has_data_dir_path.as_ref() { let no_inline = fi.data.is_none() && fi.size > 0; if no_inline { if let Err(err) = os::rename_all(&src_data_path, &dst_data_path, &skip_parent).await { let _ = self.delete_file(&dst_volume_dir, dst_data_path, false, false).await; - + info!( + "rename all failed src_data_path: {:?}, dst_data_path: {:?}, err: {:?}", + src_data_path, dst_data_path, err + ); return Err({ if let Some(e) = err.to_io_err() { os_err_to_file_err(e) @@ -1521,6 +1527,7 @@ impl DiskAPI for LocalDisk { ) .await { + info!("write_all_private failed err: {:?}", err); return Err({ if let Some(e) = err.to_io_err() { os_err_to_file_err(e) @@ -1536,6 +1543,7 @@ impl DiskAPI for LocalDisk { if let Some((_, dst_data_path)) = has_data_dir_path.as_ref() { let _ = self.delete_file(&dst_volume_dir, dst_data_path, false, false).await; } + info!("rename all failed err: {:?}", err); return Err({ if let Some(e) = err.to_io_err() { os_err_to_file_err(e) @@ -1834,7 +1842,7 @@ impl DiskAPI for LocalDisk { if let Some(old_data_dir) = opts.old_data_dir { if opts.undo_write { let src_path = file_path.join(Path::new( - format!("{}{}{}", old_data_dir.to_string(), SLASH_SEPARATOR, STORAGE_FORMAT_FILE_BACKUP).as_str(), + format!("{}{}{}", old_data_dir, SLASH_SEPARATOR, STORAGE_FORMAT_FILE_BACKUP).as_str(), )); let dst_path = file_path.join(Path::new(format!("{}{}{}", path, SLASH_SEPARATOR, STORAGE_FORMAT_FILE).as_str())); return rename_all(src_path, dst_path, file_path).await; @@ -1924,6 +1932,7 @@ impl DiskAPI for LocalDisk { } async fn delete_volume(&self, volume: &str) -> Result<()> { + info!("delete_volume, volume: {}", volume); let p = self.get_bucket_path(volume)?; // TODO: 不能用递归删除，如果目录下面有文件，返回errVolumeNotEmpty diff --git a/ecstore/src/disk/mod.rs b/ecstore/src/disk/mod.rs index 10ceeafb7..0703e59dc 100644 --- a/ecstore/src/disk/mod.rs +++ b/ecstore/src/disk/mod.rs @@ -330,6 +330,7 @@ impl DiskAPI for Disk { } async fn delete_volume(&self, volume: &str) -> Result<()> { + info!("delete_volume, volume: {}", volume); match self { Disk::Local(local_disk) => local_disk.delete_volume(volume).await, Disk::Remote(remote_disk) => remote_disk.delete_volume(volume).await, diff --git a/ecstore/src/disk/os.rs b/ecstore/src/disk/os.rs index 322d5f258..fd5e043d6 100644 --- a/ecstore/src/disk/os.rs +++ b/ecstore/src/disk/os.rs @@ -4,6 +4,7 @@ use std::{ }; use tokio::fs; +use tracing::info; use crate::{ disk::error::{is_sys_err_not_dir, is_sys_err_path_not_found, os_is_not_exist}, @@ -138,7 +139,16 @@ pub async fn reliable_rename( if let Some(parent) = dst_file_path.as_ref().parent() { reliable_mkdir_all(parent, base_dir.as_ref()).await?; } - + // need remove dst path + if let Err(err) = utils::fs::remove_all(dst_file_path.as_ref()).await { + info!( + "reliable_rename rm dst failed. src_file_path: {:?}, dst_file_path: {:?}, base_dir: {:?}, err: {:?}", + src_file_path.as_ref(), + dst_file_path.as_ref(), + base_dir.as_ref(), + err + ); + } let mut i = 0; loop { if let Err(e) = utils::fs::rename(src_file_path.as_ref(), dst_file_path.as_ref()).await { @@ -146,7 +156,13 @@ pub async fn reliable_rename( i += 1; continue; } - + info!( + "reliable_rename failed. src_file_path: {:?}, dst_file_path: {:?}, base_dir: {:?}, err: {:?}", + src_file_path.as_ref(), + dst_file_path.as_ref(), + base_dir.as_ref(), + e + ); return Err(e); } diff --git a/ecstore/src/erasure.rs b/ecstore/src/erasure.rs index c990cce5b..9691a763a 100644 --- a/ecstore/src/erasure.rs +++ b/ecstore/src/erasure.rs @@ -10,6 +10,7 @@ use std::fmt::Debug; use std::io::ErrorKind; use tokio::io::DuplexStream; use tokio::io::{AsyncReadExt, AsyncWriteExt}; +use tracing::info; use tracing::warn; // use tracing::debug; use uuid::Uuid; @@ -425,6 +426,12 @@ impl Erasure { total_length: usize, _prefer: &[bool], ) -> Result<()> { + info!( + "Erasure heal, writers len: {}, readers len: {}, total_length: {}", + writers.len(), + readers.len(), + total_length + ); if writers.len() != self.parity_shards + self.data_shards { return Err(Error::from_string("invalid argument")); } @@ -437,14 +444,14 @@ impl Erasure { } let mut errs = Vec::new(); - for _ in start_block..=end_block { + for _ in start_block..end_block { let mut bufs = reader.read().await?; if self.parity_shards > 0 { self.encoder.as_ref().unwrap().reconstruct(&mut bufs)?; } - let shards = bufs.into_iter().flatten().collect::>(); + let shards: Vec> = bufs.into_iter().flatten().collect::>(); if shards.len() != self.parity_shards + self.data_shards { return Err(Error::from_string("can not reconstruct data")); } @@ -455,7 +462,10 @@ impl Erasure { } match w.as_mut().unwrap().write(shards[i].as_ref()).await { Ok(_) => {} - Err(e) => errs.push(e), + Err(e) => { + info!("write failed, err: {:?}", e); + errs.push(e); + } } } } diff --git a/ecstore/src/global.rs b/ecstore/src/global.rs index fdb7f8444..69e3bc218 100644 --- a/ecstore/src/global.rs +++ b/ecstore/src/global.rs @@ -28,8 +28,9 @@ lazy_static! { pub static ref GLOBAL_LOCAL_DISK_SET_DRIVES: Arc> = Arc::new(RwLock::new(Vec::new())); pub static ref GLOBAL_Endpoints: OnceLock = OnceLock::new(); pub static ref GLOBAL_RootDiskThreshold: RwLock = RwLock::new(0); - pub static ref GLOBAL_BackgroundHealRoutine: Arc> = HealRoutine::new(); - pub static ref GLOBAL_BackgroundHealState: Arc> = AllHealState::new(false); + pub static ref GLOBAL_BackgroundHealRoutine: Arc = HealRoutine::new(); + pub static ref GLOBAL_BackgroundHealState: Arc = AllHealState::new(false); + pub static ref GLOBAL_ALlHealState: Arc = AllHealState::new(false); static ref globalDeploymentIDPtr: RwLock = RwLock::new(Uuid::nil()); } @@ -50,11 +51,7 @@ pub fn set_global_endpoints(eps: Vec) { } pub fn new_object_layer_fn() -> Option> { - if let Some(ec) = GLOBAL_OBJECT_API.get() { - Some(ec.clone()) - } else { - None - } + GLOBAL_OBJECT_API.get().map(|ec| ec.clone()) } pub async fn set_object_layer(o: Arc) { diff --git a/ecstore/src/heal/background_heal_ops.rs b/ecstore/src/heal/background_heal_ops.rs index f90bab544..fa9a1f63d 100644 --- a/ecstore/src/heal/background_heal_ops.rs +++ b/ecstore/src/heal/background_heal_ops.rs @@ -37,9 +37,8 @@ pub async fn init_auto_heal() { init_background_healing().await; if let Ok(v) = env::var("_RUSTFS_AUTO_DRIVE_HEALING") { if v == "on" { + info!("start monitor local disks and heal"); GLOBAL_BackgroundHealState - .write() - .await .push_heal_local_disks(&get_local_disks_to_heal().await) .await; tokio::spawn(async { @@ -50,18 +49,14 @@ pub async fn init_auto_heal() { } async fn init_background_healing() { - let bg_seq = Arc::new(RwLock::new(new_bg_heal_sequence())); - for _ in 0..GLOBAL_BackgroundHealRoutine.read().await.workers { + let bg_seq = Arc::new(new_bg_heal_sequence()); + for _ in 0..GLOBAL_BackgroundHealRoutine.workers { let bg_seq_clone = bg_seq.clone(); tokio::spawn(async { - GLOBAL_BackgroundHealRoutine.write().await.add_worker(bg_seq_clone).await; + GLOBAL_BackgroundHealRoutine.add_worker(bg_seq_clone).await; }); } - let _ = GLOBAL_BackgroundHealState - .write() - .await - .launch_new_heal_sequence(bg_seq) - .await; + let _ = GLOBAL_BackgroundHealState.launch_new_heal_sequence(bg_seq).await; } pub async fn get_local_disks_to_heal() -> Vec { @@ -94,7 +89,7 @@ async fn monitor_local_disks_and_heal() { loop { interval.tick().await; - let heal_disks = GLOBAL_BackgroundHealState.read().await.get_heal_local_disk_endpoints().await; + let heal_disks = GLOBAL_BackgroundHealState.get_heal_local_disk_endpoints().await; if heal_disks.is_empty() { interval.reset(); continue; @@ -114,23 +109,15 @@ async fn monitor_local_disks_and_heal() { let disk_clone = disk.clone(); tokio::spawn(async move { GLOBAL_BackgroundHealState - .write() - .await .set_disk_healing_status(disk_clone.clone(), true) .await; if heal_fresh_disk(&disk_clone).await.is_err() { GLOBAL_BackgroundHealState - .write() - .await .set_disk_healing_status(disk_clone.clone(), false) .await; return; } - GLOBAL_BackgroundHealState - .write() - .await - .pop_heal_local_disks(&[disk_clone]) - .await; + GLOBAL_BackgroundHealState.pop_heal_local_disks(&[disk_clone]).await; }); } interval.reset(); @@ -317,6 +304,7 @@ impl HealTask { } } +#[derive(Debug)] pub struct HealResult { pub result: HealResultItem, pub err: Option, @@ -324,12 +312,12 @@ pub struct HealResult { pub struct HealRoutine { pub tasks_tx: Sender, - tasks_rx: Receiver, + tasks_rx: RwLock>, workers: usize, } impl HealRoutine { - pub fn new() -> Arc> { + pub fn new() -> Arc { let mut workers = num_cpus::get() / 2; if let Ok(env_heal_workers) = env::var("_RUSTFS_HEAL_WORKERS") { if let Ok(num_healers) = env_heal_workers.parse::() { @@ -342,19 +330,20 @@ impl HealRoutine { } let (tx, rx) = mpsc::channel(100); - Arc::new(RwLock::new(Self { + Arc::new(Self { tasks_tx: tx, - tasks_rx: rx, + tasks_rx: RwLock::new(rx), workers, - })) + }) } - pub async fn add_worker(&mut self, bgseq: Arc>) { + pub async fn add_worker(&self, bgseq: Arc) { loop { let mut d_res = HealResultItem::default(); let d_err: Option; - match self.tasks_rx.recv().await { + match self.tasks_rx.write().await.recv().await { Some(task) => { + info!("got task: {:?}", task); if task.bucket == NOP_HEAL { d_err = Some(Error::from_string("skip file")); } else if task.bucket == SLASH_SEPARATOR { @@ -388,6 +377,7 @@ impl HealRoutine { } } } + info!("task finished, task: {:?}", task); if let Some(resp_tx) = task.resp_tx { let _ = resp_tx .send(HealResult { @@ -399,13 +389,16 @@ impl HealRoutine { // when respCh is not set caller is not waiting but we // update the relevant metrics for them if d_err.is_none() { - bgseq.write().await.count_healed(d_res.heal_item_type); + bgseq.count_healed(d_res.heal_item_type).await; } else { - bgseq.write().await.count_failed(d_res.heal_item_type); + bgseq.count_failed(d_res.heal_item_type).await; } } } - None => return, + None => { + info!("add_worker, tasks_rx was closed, return"); + return; + } } } } diff --git a/ecstore/src/heal/data_scanner.rs b/ecstore/src/heal/data_scanner.rs index d45949371..61efee453 100644 --- a/ecstore/src/heal/data_scanner.rs +++ b/ecstore/src/heal/data_scanner.rs @@ -511,9 +511,7 @@ impl FolderScanner { let this_hash = hash_path(&folder.name); let was_compacted = into.compacted; - // do not really loop - let mut times = 1; - while times > 0 { + 'outer: { let mut abandoned_children: DataUsageHashMap = if !into.compacted { self.old_cache.find_children_copy(this_hash.clone()) } else { @@ -622,7 +620,7 @@ impl FolderScanner { } if found_objects && *GLOBAL_IsErasure.read().await { // If we found an object in erasure mode, we skip subdirs (only datadirs)... - break; + break 'outer; } let should_compact = self.new_cache.info.name != folder.name @@ -714,20 +712,16 @@ impl FolderScanner { // Scan for healing if abandoned_children.is_empty() || !self.should_heal().await { - break; + break 'outer; } if self.disks.is_empty() || self.disks_quorum == 0 { - break; + break 'outer; } - let (bg_seq, found) = GLOBAL_BackgroundHealState - .read() - .await - .get_heal_sequence_by_token(BG_HEALING_UUID) - .await; + let (bg_seq, found) = GLOBAL_BackgroundHealState.get_heal_sequence_by_token(BG_HEALING_UUID).await; if !found { - break; + break 'outer; } let bg_seq = bg_seq.unwrap(); @@ -750,8 +744,6 @@ impl FolderScanner { if bucket != resolver.bucket { bg_seq .clone() - .write() - .await .queue_heal_task( HealSource { bucket: bucket.clone(), @@ -820,8 +812,6 @@ impl FolderScanner { Ok(fiv) => fiv, Err(_) => { if let Err(err) = bg_seq_partial - .write() - .await .queue_heal_task( HealSource { bucket: bucket_partial.clone(), @@ -851,8 +841,6 @@ impl FolderScanner { let (mut success_versions, mut fail_versions) = (0, 0); for ver in fiv.versions.iter() { match bg_seq_partial - .write() - .await .queue_heal_task( HealSource { bucket: bucket_partial.clone(), @@ -901,7 +889,6 @@ impl FolderScanner { scan(&this, into, self).await; } } - times -= 1; } if !was_compacted { self.new_cache.replace_hashed(&this_hash, &Some(folder.parent.clone()), into); diff --git a/ecstore/src/heal/heal_commands.rs b/ecstore/src/heal/heal_commands.rs index 06f97fa13..c5eb3ea1e 100644 --- a/ecstore/src/heal/heal_commands.rs +++ b/ecstore/src/heal/heal_commands.rs @@ -1,5 +1,6 @@ use std::{path::Path, time::SystemTime}; +use chrono::{DateTime, Utc}; use lazy_static::lazy_static; use serde::{Deserialize, Serialize}; use time::OffsetDateTime; @@ -44,45 +45,71 @@ lazy_static! { #[derive(Clone, Copy, Debug, Default, Serialize, Deserialize)] pub struct HealOpts { pub recursive: bool, + #[serde(rename = "dryRun")] pub dry_run: bool, pub remove: bool, pub recreate: bool, + #[serde(rename = "scanMode")] pub scan_mode: HealScanMode, + #[serde(rename = "updateParity")] pub update_parity: bool, + #[serde(rename = "nolock")] pub no_lock: bool, pub pool: Option, pub set: Option, } -#[derive(Clone, Debug, Default)] +#[derive(Clone, Debug, Default, Serialize, Deserialize)] pub struct HealDriveInfo { pub uuid: String, pub endpoint: String, pub state: String, } -#[derive(Clone, Debug, Default)] +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub struct Infos { + #[serde(rename = "drives")] + pub drives: Vec, +} + +#[derive(Clone, Debug, Default, Serialize, Deserialize)] pub struct HealResultItem { + #[serde(rename = "resultId")] pub result_index: usize, + #[serde(rename = "type")] pub heal_item_type: HealItemType, + #[serde(rename = "bucket")] pub bucket: String, + #[serde(rename = "object")] pub object: String, + #[serde(rename = "versionId")] pub version_id: String, + #[serde(rename = "detail")] pub detail: String, + #[serde(rename = "parityBlocks")] pub parity_blocks: usize, + #[serde(rename = "dataBlocks")] pub data_blocks: usize, + #[serde(rename = "diskCount")] pub disk_count: usize, + #[serde(rename = "setCount")] pub set_count: usize, - pub before: Vec, - pub after: Vec, + #[serde(rename = "before")] + pub before: Infos, + #[serde(rename = "after")] + pub after: Infos, + #[serde(rename = "objectSize")] pub object_size: usize, } #[derive(Debug, Serialize, Deserialize)] pub struct HealStartSuccess { + #[serde(rename = "clientToken")] pub client_token: String, + #[serde(rename = "clientAddress")] pub client_address: String, - pub start_time: SystemTime, + #[serde(rename = "startTime")] + pub start_time: DateTime, } impl Default for HealStartSuccess { @@ -90,7 +117,7 @@ impl Default for HealStartSuccess { Self { client_token: Default::default(), client_address: Default::default(), - start_time: SystemTime::now(), + start_time: Utc::now(), } } } @@ -265,7 +292,7 @@ impl HealingTracker { let htracker_bytes = self.marshal_msg()?; - GLOBAL_BackgroundHealState.write().await.update_heal_status(self).await; + GLOBAL_BackgroundHealState.update_heal_status(self).await; if let Some(disk) = &self.disk { let file_path = Path::new(BUCKET_META_PREFIX).join(HEALING_TRACKER_FILENAME); diff --git a/ecstore/src/heal/heal_ops.rs b/ecstore/src/heal/heal_ops.rs index 22b376e97..73d7f82f7 100644 --- a/ecstore/src/heal/heal_ops.rs +++ b/ecstore/src/heal/heal_ops.rs @@ -7,6 +7,7 @@ use super::{ HEAL_ITEM_BUCKET_METADATA, }, }; +use crate::store_api::StorageAPI; use crate::{ config::common::CONFIG_PREFIX, disk::RUSTFS_META_BUCKET, @@ -27,13 +28,16 @@ use crate::{ }; use crate::{ heal::heal_commands::{HEAL_ITEM_BUCKET, HEAL_ITEM_OBJECT}, - store_api::StorageAPI, + utils::path::path_join, }; +use chrono::Utc; +use futures::join; use lazy_static::lazy_static; +use serde::{Deserialize, Serialize}; use std::{ collections::HashMap, future::Future, - path::Path, + path::PathBuf, pin::Pin, sync::Arc, time::{Duration, SystemTime, UNIX_EPOCH}, @@ -41,13 +45,14 @@ use std::{ use tokio::{ select, spawn, sync::{ - broadcast::{self, Receiver, Sender}, + broadcast, mpsc::{self, Receiver as M_Receiver, Sender as M_Sender}, + watch::{self, Receiver as W_Receiver, Sender as W_Sender}, RwLock, }, time::{interval, sleep}, }; -use tracing::info; +use tracing::{error, info}; use uuid::Uuid; type HealStatusSummary = String; @@ -73,7 +78,7 @@ pub const NOP_HEAL: &str = ""; lazy_static! {} -#[derive(Clone, Debug, Default)] +#[derive(Clone, Debug, Default, Serialize, Deserialize)] pub struct HealSequenceStatus { pub summary: HealStatusSummary, pub failure_detail: String, @@ -91,7 +96,7 @@ pub struct HealSource { pub opts: Option, } -#[derive(Clone, Debug)] +#[derive(Debug)] pub struct HealSequence { pub bucket: String, pub object: String, @@ -103,17 +108,17 @@ pub struct HealSequence { pub force_started: bool, pub setting: HealOpts, pub current_status: Arc>, - pub last_sent_result_index: usize, - pub scanned_items_map: ItemsMap, - pub healed_items_map: ItemsMap, - pub heal_failed_items_map: ItemsMap, - pub last_heal_activity: u64, + pub last_sent_result_index: RwLock, + pub scanned_items_map: RwLock, + pub healed_items_map: RwLock, + pub heal_failed_items_map: RwLock, + pub last_heal_activity: RwLock, traverse_and_heal_done_tx: Arc>>>, traverse_and_heal_done_rx: Arc>>>, - tx: Arc>>, - rx: Arc>>, + tx: W_Sender, + rx: W_Receiver, } pub fn new_bg_heal_sequence() -> HealSequence { @@ -133,9 +138,35 @@ pub fn new_bg_heal_sequence() -> HealSequence { ..Default::default() })), report_progress: false, - scanned_items_map: HashMap::new(), - healed_items_map: HashMap::new(), - heal_failed_items_map: HashMap::new(), + scanned_items_map: HashMap::new().into(), + healed_items_map: HashMap::new().into(), + heal_failed_items_map: HashMap::new().into(), + ..Default::default() + } +} + +pub fn new_heal_sequence(bucket: &str, obj_prefix: &str, client_addr: &str, hs: HealOpts, force_start: bool) -> HealSequence { + let client_token = Uuid::new_v4().to_string(); + let (tx, rx) = mpsc::channel(10); + HealSequence { + bucket: bucket.to_string(), + object: obj_prefix.to_string(), + report_progress: true, + start_time: SystemTime::now(), + client_token, + client_address: client_addr.to_string(), + force_started: force_start, + setting: hs, + current_status: Arc::new(RwLock::new(HealSequenceStatus { + summary: HEAL_NOT_STARTED_STATUS.to_string(), + heal_setting: hs, + ..Default::default() + })), + traverse_and_heal_done_tx: Arc::new(RwLock::new(tx)), + traverse_and_heal_done_rx: Arc::new(RwLock::new(rx)), + scanned_items_map: HashMap::new().into(), + healed_items_map: HashMap::new().into(), + heal_failed_items_map: HashMap::new().into(), ..Default::default() } } @@ -143,7 +174,7 @@ pub fn new_bg_heal_sequence() -> HealSequence { impl Default for HealSequence { fn default() -> Self { let (h_tx, h_rx) = mpsc::channel(1); - let (tx, rx) = broadcast::channel(1); + let (tx, rx) = watch::channel(false); Self { bucket: Default::default(), object: Default::default(), @@ -159,11 +190,11 @@ impl Default for HealSequence { scanned_items_map: Default::default(), healed_items_map: Default::default(), heal_failed_items_map: Default::default(), - last_heal_activity: Default::default(), + last_heal_activity: RwLock::new(SystemTime::now()), traverse_and_heal_done_tx: Arc::new(RwLock::new(h_tx)), traverse_and_heal_done_rx: Arc::new(RwLock::new(h_rx)), - tx: Arc::new(RwLock::new(tx)), - rx: Arc::new(RwLock::new(rx)), + tx, + rx, } } } @@ -191,49 +222,40 @@ impl HealSequence { } impl HealSequence { - pub fn get_scanned_items_count(&self) -> usize { - self.scanned_items_map.values().sum() + pub async fn get_scanned_items_count(&self) -> usize { + self.scanned_items_map.read().await.values().sum() } - pub fn _get_scanned_items_map(&self) -> ItemsMap { - self.scanned_items_map.clone() + async fn _get_scanned_items_map(&self) -> ItemsMap { + self.scanned_items_map.read().await.clone() } - pub fn _get_healed_items_map(&self) -> ItemsMap { - self.healed_items_map.clone() + async fn _get_healed_items_map(&self) -> ItemsMap { + self.healed_items_map.read().await.clone() } - pub fn _get_heal_failed_items_map(&self) -> ItemsMap { - self.heal_failed_items_map.clone() + async fn _get_heal_failed_items_map(&self) -> ItemsMap { + self.heal_failed_items_map.read().await.clone() } - pub fn count_failed(&mut self, heal_type: HealItemType) { - *self.heal_failed_items_map.entry(heal_type).or_insert(0) += 1; - self.last_heal_activity = SystemTime::now() - .duration_since(UNIX_EPOCH) - .expect("Time went backwards") - .as_secs(); + pub async fn count_failed(&self, heal_type: HealItemType) { + *self.heal_failed_items_map.write().await.entry(heal_type).or_insert(0) += 1; + *self.last_heal_activity.write().await = SystemTime::now(); } - pub fn count_scanned(&mut self, heal_type: HealItemType) { - *self.scanned_items_map.entry(heal_type).or_insert(0) += 1; - self.last_heal_activity = SystemTime::now() - .duration_since(UNIX_EPOCH) - .expect("Time went backwards") - .as_secs(); + pub async fn count_scanned(&self, heal_type: HealItemType) { + *self.scanned_items_map.write().await.entry(heal_type).or_insert(0) += 1; + *self.last_heal_activity.write().await = SystemTime::now(); } - pub fn count_healed(&mut self, heal_type: HealItemType) { - *self.healed_items_map.entry(heal_type).or_insert(0) += 1; - self.last_heal_activity = SystemTime::now() - .duration_since(UNIX_EPOCH) - .expect("Time went backwards") - .as_secs(); + pub async fn count_healed(&self, heal_type: HealItemType) { + *self.healed_items_map.write().await.entry(heal_type).or_insert(0) += 1; + *self.last_heal_activity.write().await = SystemTime::now(); } async fn is_quitting(&self) -> bool { - let mut w = self.rx.write().await; - if w.try_recv().is_ok() { + if let Ok(true) = self.rx.has_changed() { + info!("quited"); return true; } false @@ -248,8 +270,7 @@ impl HealSequence { } async fn stop(&self) { - let w = self.tx.write().await; - let _ = w.send(true); + let _ = self.tx.send(true); } async fn push_heal_result_item(&self, r: &HealResultItem) -> Result<()> { @@ -284,7 +305,7 @@ impl HealSequence { if items_len > 0 { r.result_index = 1 + current_status_w.items[items_len - 1].result_index; } else { - r.result_index = 1 + self.last_sent_result_index; + r.result_index = 1 + *self.last_sent_result_index.read().await; } current_status_w.items.push(r); @@ -292,19 +313,20 @@ impl HealSequence { Ok(()) } - pub async fn queue_heal_task(&mut self, source: HealSource, heal_type: HealItemType) -> Result<()> { + pub async fn queue_heal_task(&self, source: HealSource, heal_type: HealItemType) -> Result<()> { let mut task = HealTask::new(&source.bucket, &source.object, &source.version_id, &self.setting); + info!("queue_heal_task, {:?}", task); if let Some(opts) = source.opts { task.opts = opts; } else { task.opts.scan_mode = HEAL_UNKNOWN_SCAN; } - self.count_scanned(heal_type.clone()); + self.count_scanned(heal_type.clone()).await; if source.no_wait { let task_str = format!("{:?}", task); - if GLOBAL_BackgroundHealRoutine.read().await.tasks_tx.try_send(task).is_ok() { + if GLOBAL_BackgroundHealRoutine.tasks_tx.try_send(task).is_ok() { info!("Task in the queue: {:?}", task_str); } return Ok(()); @@ -314,8 +336,10 @@ impl HealSequence { task.resp_tx = Some(resp_tx); let task_str = format!("{:?}", task); - if GLOBAL_BackgroundHealRoutine.read().await.tasks_tx.try_send(task).is_ok() { + if GLOBAL_BackgroundHealRoutine.tasks_tx.try_send(task).is_ok() { info!("Task in the queue: {:?}", task_str); + } else { + error!("push task to queue failed"); } let count_ok_drives = |drivers: &[HealDriveInfo]| { let mut count = 0; @@ -330,9 +354,9 @@ impl HealSequence { match resp_rx.recv().await { Some(mut res) => { if res.err.is_none() { - self.count_healed(heal_type.clone()); + self.count_healed(heal_type.clone()).await; } else { - self.count_failed(heal_type.clone()); + self.count_failed(heal_type.clone()).await; } if !self.report_progress { if let Some(err) = res.err { @@ -350,7 +374,7 @@ impl HealSequence { } if res.result.parity_blocks > 0 && res.result.data_blocks > 0 && res.result.data_blocks > res.result.parity_blocks { - let got = count_ok_drives(&res.result.after); + let got = count_ok_drives(&res.result.after.drives); if got < res.result.parity_blocks { res.result.detail = format!( "quorum loss - expected {} minimum, got drive states in OK {}", @@ -358,55 +382,62 @@ impl HealSequence { ); } } + + info!("queue_heal_task, HealResult: {:?}", res); self.push_heal_result_item(&res.result).await } None => Ok(()), } } - async fn heal_disk_meta(h: Arc>) -> Result<()> { + async fn heal_disk_meta(h: Arc) -> Result<()> { HealSequence::heal_rustfs_sys_meta(h, CONFIG_PREFIX).await } - async fn heal_items(h: Arc>, buckets_only: bool) -> Result<()> { - if h.read().await.client_token == *BG_HEALING_UUID { + async fn heal_items(h: Arc, buckets_only: bool) -> Result<()> { + if h.client_token == *BG_HEALING_UUID { return Ok(()); } - Self::heal_disk_meta(h.clone()).await?; - let bucket = h.read().await.bucket.clone(); - Self::heal_bucket(h.clone(), &bucket, buckets_only).await + let bucket = h.bucket.clone(); + let task1 = Self::heal_disk_meta(h.clone()); + let task2 = Self::heal_bucket(h.clone(), &bucket, buckets_only); + let results = join!(task1, task2); + results.0?; + results.1?; + + Ok(()) } - async fn traverse_and_heal(h: Arc>) { + async fn traverse_and_heal(h: Arc) { let buckets_only = false; let result = match Self::heal_items(h.clone(), buckets_only).await { Ok(_) => None, Err(err) => Some(err), }; - let _ = h.read().await.traverse_and_heal_done_tx.read().await.send(result).await; + let _ = h.traverse_and_heal_done_tx.read().await.send(result).await; } - async fn heal_rustfs_sys_meta(h: Arc>, meta_prefix: &str) -> Result<()> { + async fn heal_rustfs_sys_meta(h: Arc, meta_prefix: &str) -> Result<()> { + info!("heal_rustfs_sys_meta, h: {:?}", h); let Some(store) = new_object_layer_fn() else { return Err(Error::msg("errServerNotInitialized")) }; - let setting = h.read().await.setting; + let setting = h.setting; store .heal_objects(RUSTFS_META_BUCKET, meta_prefix, &setting, h.clone(), true) .await } async fn is_done(&self) -> bool { - let mut rx_w = self.rx.write().await; - if let Ok(true) = rx_w.recv().await { + if let Ok(true) = self.rx.has_changed() { return true; } false } - pub async fn heal_bucket(hs: Arc>, bucket: &str, bucket_only: bool) -> Result<()> { + pub async fn heal_bucket(hs: Arc, bucket: &str, bucket_only: bool) -> Result<()> { + info!("heal_bucket, hs: {:?}", hs); let (object, setting) = { - let mut hs_w = hs.write().await; - hs_w.queue_heal_task( + hs.queue_heal_task( HealSource { bucket: bucket.to_string(), ..Default::default() @@ -419,37 +450,38 @@ impl HealSequence { return Ok(()); } - if !hs_w.setting.recursive { - if !hs_w.object.is_empty() { - HealSequence::heal_object(hs.clone(), bucket, &hs_w.object, "", hs_w.setting.scan_mode).await?; + if !hs.setting.recursive { + if !hs.object.is_empty() { + HealSequence::heal_object(hs.clone(), bucket, &hs.object, "", hs.setting.scan_mode).await?; } return Ok(()); } - (hs_w.object.clone(), hs_w.setting) + (hs.object.clone(), hs.setting) }; let Some(store) = new_object_layer_fn() else { return Err(Error::msg("errServerNotInitialized")) }; store.heal_objects(bucket, &object, &setting, hs.clone(), false).await } pub async fn heal_object( - hs: Arc>, + hs: Arc, bucket: &str, object: &str, version_id: &str, _scan_mode: HealScanMode, ) -> Result<()> { - let mut hs_w = hs.write().await; - if hs_w.is_quitting().await { + info!("heal_object"); + if hs.is_quitting().await { + info!("heal_object hs is quitting"); return Err(Error::from_string(ERR_HEAL_STOP_SIGNALLED)); } - let setting = hs_w.setting; - hs_w.queue_heal_task( + info!("will queue task"); + hs.queue_heal_task( HealSource { bucket: bucket.to_string(), object: object.to_string(), version_id: version_id.to_string(), - opts: Some(setting), + opts: Some(hs.setting), ..Default::default() }, HEAL_ITEM_OBJECT.to_string(), @@ -460,18 +492,17 @@ impl HealSequence { } pub async fn heal_meta_object( - hs: Arc>, + hs: Arc, bucket: &str, object: &str, version_id: &str, _scan_mode: HealScanMode, ) -> Result<()> { - let mut hs_w = hs.write().await; - if hs_w.is_quitting().await { + if hs.is_quitting().await { return Err(Error::from_string(ERR_HEAL_STOP_SIGNALLED)); } - hs_w.queue_heal_task( + hs.queue_heal_task( HealSource { bucket: bucket.to_string(), object: object.to_string(), @@ -486,10 +517,9 @@ impl HealSequence { } } -pub async fn heal_sequence_start(h: Arc>) { - let r = h.read().await; +pub async fn heal_sequence_start(h: Arc) { { - let mut current_status_w = r.current_status.write().await; + let mut current_status_w = h.current_status.write().await; current_status_w.summary = HEAL_RUNNING_STATUS.to_string(); current_status_w.start_time = SystemTime::now() .duration_since(UNIX_EPOCH) @@ -503,16 +533,15 @@ pub async fn heal_sequence_start(h: Arc>) { }); let h_clone_1 = h.clone(); - let mut x = r.traverse_and_heal_done_rx.write().await; + let mut x = h.traverse_and_heal_done_rx.write().await; select! { - _ = r.is_done() => { - *(r.end_time.write().await) = SystemTime::now(); - let mut current_status_w = r.current_status.write().await; + _ = h.is_done() => { + *(h.end_time.write().await) = SystemTime::now(); + let mut current_status_w = h.current_status.write().await; current_status_w.summary = HEAL_FINISHED_STATUS.to_string(); spawn(async move { - let binding = h_clone_1.read().await; - let mut rx_w = binding.traverse_and_heal_done_rx.write().await; + let mut rx_w = h_clone_1.traverse_and_heal_done_rx.write().await; rx_w.recv().await; }); } @@ -520,12 +549,12 @@ pub async fn heal_sequence_start(h: Arc>) { if let Some(err) = result { match err { Some(err) => { - let mut current_status_w = r.current_status.write().await; + let mut current_status_w = h.current_status.write().await; (current_status_w).summary = HEAL_STOPPED_STATUS.to_string(); (current_status_w).failure_detail = err.to_string(); }, None => { - let mut current_status_w = r.current_status.write().await; + let mut current_status_w = h.current_status.write().await; (current_status_w).summary = HEAL_FINISHED_STATUS.to_string(); } } @@ -539,14 +568,14 @@ pub async fn heal_sequence_start(h: Arc>) { pub struct AllHealState { mu: RwLock, - heal_seq_map: HashMap>>, - heal_local_disks: HashMap, - heal_status: HashMap, + heal_seq_map: RwLock>>, + heal_local_disks: RwLock>, + heal_status: RwLock>, } impl AllHealState { - pub fn new(cleanup: bool) -> Arc> { - let hstate = Arc::new(RwLock::new(AllHealState::default())); + pub fn new(cleanup: bool) -> Arc { + let hstate = Arc::new(AllHealState::default()); let (_, mut rx) = broadcast::channel(1); if cleanup { let hstate_clone = hstate.clone(); @@ -559,7 +588,7 @@ impl AllHealState { } } _ = sleep(Duration::from_secs(5 * 60)) => { - hstate_clone.write().await.periodic_heal_seqs_clean().await; + hstate_clone.periodic_heal_seqs_clean().await; } } } @@ -569,10 +598,10 @@ impl AllHealState { hstate } - pub async fn pop_heal_local_disks(&mut self, heal_local_disks: &[Endpoint]) { + pub async fn pop_heal_local_disks(&self, heal_local_disks: &[Endpoint]) { let _ = self.mu.write().await; - self.heal_local_disks.retain(|k, _| { + self.heal_local_disks.write().await.retain(|k, _| { if heal_local_disks.contains(k) { return false; } @@ -580,7 +609,7 @@ impl AllHealState { }); let heal_local_disks = heal_local_disks.iter().map(|s| s.to_string()).collect::>(); - self.heal_status.retain(|_, v| { + self.heal_status.write().await.retain(|_, v| { if heal_local_disks.contains(&v.endpoint) { return false; } @@ -589,18 +618,57 @@ impl AllHealState { }); } - pub async fn update_heal_status(&mut self, tracker: &HealingTracker) { + pub async fn pop_heal_status_json(&self, heal_path: &str, client_token: &str) -> Result> { + match self.get_heal_sequence(heal_path).await { + Some(h) => { + if client_token != h.client_token { + info!("err heal invalid client token"); + return Err(Error::from_string("err heal invalid client token")); + } + let num_items = h.current_status.read().await.items.len(); + let mut last_result_index = *h.last_sent_result_index.read().await; + if num_items > 0 { + if let Some(item) = h.current_status.read().await.items.last() { + last_result_index = item.result_index; + } + } + *h.last_sent_result_index.write().await = last_result_index; + let data = h.current_status.read().await.clone(); + match serde_json::to_vec(&data) { + Ok(b) => { + h.current_status.write().await.items.clear(); + Ok(b) + } + Err(e) => { + h.current_status.write().await.items.clear(); + info!("json encode err, e: {}", e); + Err(Error::msg(e.to_string())) + } + } + } + None => serde_json::to_vec(&HealSequenceStatus { + summary: HEAL_FINISHED_STATUS.to_string(), + ..Default::default() + }) + .map_err(|e| { + info!("json encode err, e: {}", e); + Error::msg(e.to_string()) + }), + } + } + + pub async fn update_heal_status(&self, tracker: &HealingTracker) { let _ = self.mu.write().await; let _ = tracker.mu.read().await; - self.heal_status.insert(tracker.id.clone(), tracker.clone()); + self.heal_status.write().await.insert(tracker.id.clone(), tracker.clone()); } pub async fn get_local_healing_disks(&self) -> HashMap { let _ = self.mu.read().await; let mut dst = HashMap::new(); - for v in self.heal_status.values() { + for v in self.heal_status.read().await.values() { dst.insert(v.endpoint.clone(), v.to_healing_disk().await); } @@ -611,7 +679,7 @@ impl AllHealState { let _ = self.mu.read().await; let mut endpoints = Vec::new(); - self.heal_local_disks.iter().for_each(|(k, v)| { + self.heal_local_disks.read().await.iter().for_each(|(k, v)| { if !v { endpoints.push(k.clone()); } @@ -620,42 +688,40 @@ impl AllHealState { Endpoints::from(endpoints) } - pub async fn set_disk_healing_status(&mut self, ep: Endpoint, healing: bool) { + pub async fn set_disk_healing_status(&self, ep: Endpoint, healing: bool) { let _ = self.mu.write().await; - self.heal_local_disks.insert(ep, healing); + self.heal_local_disks.write().await.insert(ep, healing); } - pub async fn push_heal_local_disks(&mut self, heal_local_disks: &[Endpoint]) { + pub async fn push_heal_local_disks(&self, heal_local_disks: &[Endpoint]) { let _ = self.mu.write().await; - heal_local_disks.iter().for_each(|heal_local_disk| { - self.heal_local_disks.insert(heal_local_disk.clone(), false); - }); + for heal_local_disk in heal_local_disks.iter() { + self.heal_local_disks.write().await.insert(heal_local_disk.clone(), false); + } } - pub async fn periodic_heal_seqs_clean(&mut self) { + pub async fn periodic_heal_seqs_clean(&self) { let _ = self.mu.write().await; let now = SystemTime::now(); let mut keys_to_reomve = Vec::new(); - for (k, v) in self.heal_seq_map.iter() { - let r = v.read().await; - if r.has_ended().await && now.duration_since(*(r.end_time.read().await)).unwrap() > KEEP_HEAL_SEQ_STATE_DURATION { + for (k, v) in self.heal_seq_map.read().await.iter() { + if v.has_ended().await && now.duration_since(*(v.end_time.read().await)).unwrap() > KEEP_HEAL_SEQ_STATE_DURATION { keys_to_reomve.push(k.clone()) } } for key in keys_to_reomve.iter() { - self.heal_seq_map.remove(key); + self.heal_seq_map.write().await.remove(key); } } - pub async fn get_heal_sequence_by_token(&self, token: &str) -> (Option>>, bool) { + pub async fn get_heal_sequence_by_token(&self, token: &str) -> (Option>, bool) { let _ = self.mu.read().await; - for v in self.heal_seq_map.values() { - let r = v.read().await; - if r.client_token == token { + for v in self.heal_seq_map.read().await.values() { + if v.client_token == token { return (Some(v.clone()), true); } } @@ -663,16 +729,15 @@ impl AllHealState { (None, false) } - pub async fn get_heal_sequence(&self, path: &str) -> Option>> { + pub async fn get_heal_sequence(&self, path: &str) -> Option> { let _ = self.mu.read().await; - self.heal_seq_map.get(path).cloned() + self.heal_seq_map.read().await.get(path).cloned() } - pub async fn stop_heal_sequence(&mut self, path: &str) -> Result> { + pub async fn stop_heal_sequence(&self, path: &str) -> Result> { let mut hsp = HealStopSuccess::default(); if let Some(he) = self.get_heal_sequence(path).await { - let he = he.read().await; let client_token = he.client_token.clone(); if *GLOBAL_IsDistErasure.read().await { // TODO: proxy @@ -680,7 +745,7 @@ impl AllHealState { hsp.client_token = client_token; hsp.client_address = he.client_address.clone(); - hsp.start_time = he.start_time; + hsp.start_time = Utc::now(); he.stop().await; @@ -693,7 +758,7 @@ impl AllHealState { } let _ = self.mu.write().await; - self.heal_seq_map.remove(path); + self.heal_seq_map.write().await.remove(path); } else { hsp.client_token = "unknown".to_string(); } @@ -712,22 +777,24 @@ impl AllHealState { // `keepHealSeqStateDuration`. This function also launches a // background routine to clean up heal results after the // aforementioned duration. - pub async fn launch_new_heal_sequence(&mut self, heal_sequence: Arc>) -> Result> { - let r = heal_sequence.read().await; - let path = Path::new(&r.bucket).join(r.object.clone()); + pub async fn launch_new_heal_sequence(&self, heal_sequence: Arc) -> Result> { + let path = path_join(&[ + PathBuf::from(heal_sequence.bucket.clone()), + PathBuf::from(heal_sequence.object.clone()), + ]); let path_s = path.to_str().unwrap(); - if r.force_started { + if heal_sequence.force_started { self.stop_heal_sequence(path_s).await?; } else if let Some(hs) = self.get_heal_sequence(path_s).await { - if !hs.read().await.has_ended().await { - return Err(Error::from_string(format!("Heal is already running on the given path (use force-start option to stop and start afresh). The heal was started by IP {} at {:?}, token is {}", r.client_address, r.start_time, r.client_token))); + if !hs.has_ended().await { + return Err(Error::from_string(format!("Heal is already running on the given path (use force-start option to stop and start afresh). The heal was started by IP {} at {:?}, token is {}", heal_sequence.client_address, heal_sequence.start_time, heal_sequence.client_token))); } } let _ = self.mu.write().await; - for (k, v) in self.heal_seq_map.iter() { - if !v.read().await.has_ended().await && (has_profix(k, path_s) || has_profix(path_s, k)) { + for (k, v) in self.heal_seq_map.read().await.iter() { + if (has_profix(k, path_s) || has_profix(path_s, k)) && !v.has_ended().await { return Err(Error::from_string(format!( "The provided heal sequence path overlaps with an existing heal path: {}", k @@ -735,14 +802,17 @@ impl AllHealState { } } - self.heal_seq_map.insert(path_s.to_string(), heal_sequence.clone()); + self.heal_seq_map + .write() + .await + .insert(path_s.to_string(), heal_sequence.clone()); - let client_token = r.client_token.clone(); + let client_token = heal_sequence.client_token.clone(); if *GLOBAL_IsDistErasure.read().await { // TODO: proxy } - if r.client_token == BG_HEALING_UUID { + if heal_sequence.client_token == BG_HEALING_UUID { // For background heal do nothing, do not spawn an unnecessary goroutine. } else { let heal_sequence_clone = heal_sequence.clone(); @@ -753,8 +823,9 @@ impl AllHealState { let b = serde_json::to_vec(&HealStartSuccess { client_token, - client_address: r.client_address.clone(), - start_time: r.start_time, + client_address: heal_sequence.client_address.clone(), + // start_time: Utc::now(), + start_time: heal_sequence.start_time.into(), })?; Ok(b) } diff --git a/ecstore/src/peer.rs b/ecstore/src/peer.rs index 369b47dfb..61737fc04 100644 --- a/ecstore/src/peer.rs +++ b/ecstore/src/peer.rs @@ -8,9 +8,9 @@ use regex::Regex; use std::{collections::HashMap, fmt::Debug, sync::Arc}; use tokio::sync::RwLock; use tonic::Request; -use tracing::warn; +use tracing::{info, warn}; -use crate::disk::error::{is_all_buckets_not_found, is_all_not_found}; +use crate::disk::error::is_all_buckets_not_found; use crate::disk::{DiskAPI, DiskStore}; use crate::global::GLOBAL_LOCAL_DISK_MAP; use crate::heal::heal_commands::{ @@ -103,14 +103,14 @@ impl S3PeerSys { } if !opts.recreate { - opts.remove = is_all_not_found(&pool_errs); - opts.recursive = !opts.remove; + opts.remove = is_all_buckets_not_found(&pool_errs); + opts.recreate = !opts.remove; } let mut futures = Vec::new(); let heal_bucket_results = Arc::new(RwLock::new(vec![HealResultItem::default(); self.clients.len()])); for (idx, client) in self.clients.iter().enumerate() { - let opts_clone = opts; + let opts_clone = opts.clone(); let heal_bucket_results_clone = heal_bucket_results.clone(); futures.push(async move { match client.heal_bucket(bucket, &opts_clone).await { @@ -716,21 +716,23 @@ pub async fn heal_bucket_local(bucket: &str, opts: &HealOpts) -> Result { + info!("will call delete_volume, volume: {}", bucket); let _ = disk.delete_volume(&bucket).await; None } @@ -752,6 +754,7 @@ pub async fn heal_bucket_local(bucket: &str, opts: &HealOpts) -> Result { as_clone.write().await[idx] = DRIVE_STATE_OK.to_string(); @@ -770,7 +773,7 @@ pub async fn heal_bucket_local(bucket: &str, opts: &HealOpts) -> Result (String, String) { - path2_bucket_object_with_base_path("", &name) + path2_bucket_object_with_base_path("", name) } fn path2_bucket_object_with_base_path(base_path: &str, path: &str) -> (String, String) { @@ -342,7 +342,7 @@ impl PoolDecommissionInfo { let mut found = None; for (i, b) in self.queued_buckets.iter().enumerate() { if b == bucket { - found = Some(i.clone()); + found = Some(i); break; } } @@ -471,7 +471,7 @@ impl ECStore { return; }; for idx in indices.iter() { - store.do_decommission_in_routine(idx.clone()).await; + store.do_decommission_in_routine(*idx).await; } }); @@ -481,7 +481,7 @@ impl ECStore { async fn decommission_pool(&self, _idx: usize, _pool: Arc, bi: DecomBucketInfo) -> Result<()> { let mut _vc = None; - if &bi.name == RUSTFS_META_BUCKET { + if bi.name == RUSTFS_META_BUCKET { let versioning = BucketVersioningSys::get(&bi.name).await?; _vc = Some(versioning); } @@ -518,10 +518,8 @@ impl ECStore { if let Err(er) = self.decommission_failed(idx).await { error!("decom failed err {:?}", &er); } - } else { - if let Err(er) = self.complete_decommission(idx).await { - error!("decom complete err {:?}", &er); - } + } else if let Err(er) = self.complete_decommission(idx).await { + error!("decom complete err {:?}", &er); } } @@ -616,7 +614,7 @@ impl ECStore { let _ = self.heal_bucket(&bk.name, &HealOpts::default()).await; } - let meta_buckets = vec![ + let meta_buckets = [ path_join(&[PathBuf::from(RUSTFS_META_BUCKET), PathBuf::from(CONFIG_PREFIX)]), path_join(&[PathBuf::from(RUSTFS_META_BUCKET), PathBuf::from(BUCKET_META_PREFIX)]), ]; @@ -634,11 +632,11 @@ impl ECStore { let mut pool_meta = self.pool_meta.write().await; for idx in indices.iter() { - let pi = self.get_decommission_pool_space_info(idx.clone()).await?; + let pi = self.get_decommission_pool_space_info(*idx).await?; - pool_meta.decommission(idx.clone(), pi)?; + pool_meta.decommission(*idx, pi)?; - pool_meta.queue_buckets(idx.clone(), decom_buckets.clone()); + pool_meta.queue_buckets(*idx, decom_buckets.clone()); } pool_meta.save(self.pools.clone()).await?; @@ -672,7 +670,7 @@ impl ECStore { } } -fn get_total_usable_capacity(disks: &Vec, info: &StorageInfo) -> usize { +fn get_total_usable_capacity(disks: &[StorageDisk], info: &StorageInfo) -> usize { let mut capacity = 0; for disk in disks.iter() { if disk.pool_index < 0 || info.backend.standard_sc_data.len() <= disk.pool_index as usize { @@ -685,7 +683,7 @@ fn get_total_usable_capacity(disks: &Vec, info: &StorageInfo) -> us capacity } -fn get_total_usable_capacity_free(disks: &Vec, info: &StorageInfo) -> usize { +fn get_total_usable_capacity_free(disks: &[StorageDisk], info: &StorageInfo) -> usize { let mut capacity = 0; for disk in disks.iter() { if disk.pool_index < 0 || info.backend.standard_sc_data.len() <= disk.pool_index as usize { diff --git a/ecstore/src/sets.rs b/ecstore/src/sets.rs index 8dd190839..8997eb101 100644 --- a/ecstore/src/sets.rs +++ b/ecstore/src/sets.rs @@ -587,12 +587,12 @@ impl StorageAPI for Sets { ..Default::default() }; let before_derives = formats_to_drives_info(&self.endpoints.endpoints, &formats, &errs); - res.before = vec![HealDriveInfo::default(); before_derives.len()]; - res.after = vec![HealDriveInfo::default(); before_derives.len()]; + res.before.drives = vec![HealDriveInfo::default(); before_derives.len()]; + res.after.drives = vec![HealDriveInfo::default(); before_derives.len()]; for v in before_derives.iter() { - res.before.push(v.clone()); - res.after.push(v.clone()); + res.before.drives.push(v.clone()); + res.after.drives.push(v.clone()); } if DiskError::UnformattedDisk.count_errs(&errs) == 0 { return Ok((res, Some(Error::new(DiskError::NoHealRequired)))); @@ -609,8 +609,8 @@ impl StorageAPI for Sets { for (i, set) in new_format_sets.iter().enumerate() { for (j, fm) in set.iter().enumerate() { if let Some(fm) = fm { - res.after[i * self.set_drive_count + j].uuid = fm.erasure.this.to_string(); - res.after[i * self.set_drive_count + j].state = DRIVE_STATE_OK.to_string(); + res.after.drives[i * self.set_drive_count + j].uuid = fm.erasure.this.to_string(); + res.after.drives[i * self.set_drive_count + j].state = DRIVE_STATE_OK.to_string(); tmp_new_formats[i * self.set_drive_count + j] = Some(fm.clone()); } } @@ -662,7 +662,7 @@ impl StorageAPI for Sets { _bucket: &str, _prefix: &str, _opts: &HealOpts, - _hs: Arc>, + _hs: Arc, _is_meta: bool, ) -> Result<()> { unimplemented!() diff --git a/ecstore/src/store.rs b/ecstore/src/store.rs index 9787a4ad9..a89ca1b99 100644 --- a/ecstore/src/store.rs +++ b/ecstore/src/store.rs @@ -635,7 +635,7 @@ impl ECStore { Err(to_object_err(Error::new(DiskError::FileNotFound), vec![bucket, object])) } - async fn pools_with_object(&self, pools: &Vec, opts: &ObjectOptions) -> Vec { + async fn pools_with_object(&self, pools: &[PoolObjInfo], opts: &ObjectOptions) -> Vec { let mut errs = Vec::new(); let pool_meta = self.pool_meta.read().await; for pool in pools.iter() { @@ -1183,7 +1183,7 @@ impl StorageAPI for ECStore { } let backend = self.backend_info().await; - StorageInfo { backend: backend, disks } + StorageInfo { backend, disks } } async fn list_bucket(&self, opts: &BucketOptions) -> Result> { @@ -1323,7 +1323,7 @@ impl StorageAPI for ECStore { for obj in objects.iter() { futures.push(async move { self.internal_get_pool_info_existing_with_opts( - &bucket, + bucket, &obj.object_name, &ObjectOptions { no_lock: true, @@ -1381,18 +1381,14 @@ impl StorageAPI for ECStore { if let Some(obj) = objects.get(i) { 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()); - } + } 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); - } + } else if let Some(val) = orig_index_map.get_mut(&pinfo.index) { + val.push(i); } } } @@ -1934,8 +1930,8 @@ impl StorageAPI for ECStore { } r.disk_count += result.disk_count; r.set_count += result.set_count; - r.before.append(&mut result.before); - r.after.append(&mut result.after); + r.before.drives.append(&mut result.before.drives); + r.after.drives.append(&mut result.after.drives); } if count_no_heal == self.pools.len() { return Ok((r, Some(Error::new(DiskError::NoHealRequired)))); @@ -1953,6 +1949,7 @@ impl StorageAPI for ECStore { version_id: &str, opts: &HealOpts, ) -> Result<(HealResultItem, Option)> { + info!("ECStore heal_object"); let object = utils::path::encode_dir_object(object); let errs = Arc::new(RwLock::new(vec![None; self.pools.len()])); let results = Arc::new(RwLock::new(vec![HealResultItem::default(); self.pools.len()])); @@ -1967,10 +1964,10 @@ impl StorageAPI for ECStore { Ok((mut result, err)) => { result.object = utils::path::decode_dir_object(&result.object); results.write().await.insert(idx, result); - errs.write().await.insert(idx, err); + errs.write().await[idx] = err; } Err(err) => { - errs.write().await.insert(idx, Some(err)); + errs.write().await[idx] = Some(err); } } }); @@ -2009,9 +2006,10 @@ impl StorageAPI for ECStore { bucket: &str, prefix: &str, opts: &HealOpts, - hs: Arc>, + hs: Arc, is_meta: bool, ) -> Result<()> { + info!("heal objects"); let opts_clone = *opts; let heal_entry: HealEntryFn = Arc::new(move |bucket: String, entry: MetaCacheEntry, scan_mode: HealScanMode| { let opts_clone = opts_clone; @@ -2173,7 +2171,7 @@ async fn init_local_peer(endpoint_pools: &EndpointServerPools, host: &String, po *GLOBAL_Local_Node_Name.write().await = peer_set[0].clone(); } -fn is_valid_object_prefix(object: &str) -> bool { +pub fn is_valid_object_prefix(object: &str) -> bool { // Implement object prefix validation !object.is_empty() // Placeholder } diff --git a/ecstore/src/store_api.rs b/ecstore/src/store_api.rs index 75cb4133b..fc414684a 100644 --- a/ecstore/src/store_api.rs +++ b/ecstore/src/store_api.rs @@ -14,7 +14,6 @@ use serde::{Deserialize, Serialize}; use std::collections::HashMap; use std::sync::Arc; use time::OffsetDateTime; -use tokio::sync::RwLock; use uuid::Uuid; pub const ERASURE_ALGORITHM: &str = "rs-vandermonde"; @@ -292,7 +291,7 @@ pub struct ObjectPartInfo { // } // } -#[derive(Serialize, Deserialize)] +#[derive(Default, Serialize, Deserialize)] pub struct RawFileInfo { pub buf: Vec, } @@ -1023,14 +1022,8 @@ pub trait StorageAPI: ObjectIO { version_id: &str, opts: &HealOpts, ) -> Result<(HealResultItem, Option)>; - async fn heal_objects( - &self, - bucket: &str, - prefix: &str, - opts: &HealOpts, - hs: Arc>, - is_meta: bool, - ) -> Result<()>; + async fn heal_objects(&self, bucket: &str, prefix: &str, opts: &HealOpts, hs: Arc, is_meta: bool) + -> Result<()>; async fn get_pool_and_set(&self, id: &str) -> Result<(Option, Option, Option)>; async fn check_abandoned_parts(&self, bucket: &str, object: &str, opts: &HealOpts) -> Result<()>; } diff --git a/ecstore/src/store_err.rs b/ecstore/src/store_err.rs index ca7f58778..5ee591dd5 100644 --- a/ecstore/src/store_err.rs +++ b/ecstore/src/store_err.rs @@ -84,14 +84,14 @@ pub fn to_object_err(err: Error, params: Vec<&str>) -> Error { } DiskError::FileNotFound => { - let bucket = params.get(0).cloned().unwrap_or_default().to_owned(); - let object = params.get(1).cloned().map(|v| decode_dir_object(v)).unwrap_or_default(); + let bucket = params.first().cloned().unwrap_or_default().to_owned(); + let object = params.get(1).cloned().map(decode_dir_object).unwrap_or_default(); return Error::new(StorageError::ObjectNotFound(bucket, object)); } DiskError::FileVersionNotFound => { - let bucket = params.get(0).cloned().unwrap_or_default().to_owned(); - let object = params.get(1).cloned().map(|v| decode_dir_object(v)).unwrap_or_default(); + let bucket = params.first().cloned().unwrap_or_default().to_owned(); + let object = params.get(1).cloned().map(decode_dir_object).unwrap_or_default(); let version = params.get(2).cloned().unwrap_or_default().to_owned(); return Error::new(StorageError::VersionNotFound(bucket, object, version)); @@ -100,34 +100,34 @@ pub fn to_object_err(err: Error, params: Vec<&str>) -> Error { return Error::new(StorageError::SlowDown); } DiskError::FileNameTooLong => { - let bucket = params.get(0).cloned().unwrap_or_default().to_owned(); - let object = params.get(1).cloned().map(|v| decode_dir_object(v)).unwrap_or_default(); + let bucket = params.first().cloned().unwrap_or_default().to_owned(); + let object = params.get(1).cloned().map(decode_dir_object).unwrap_or_default(); return Error::new(StorageError::ObjectNameInvalid(bucket, object)); } DiskError::VolumeExists => { - let bucket = params.get(0).cloned().unwrap_or_default().to_owned(); + let bucket = params.first().cloned().unwrap_or_default().to_owned(); return Error::new(StorageError::BucketExists(bucket)); } DiskError::IsNotRegular => { - let bucket = params.get(0).cloned().unwrap_or_default().to_owned(); - let object = params.get(1).cloned().map(|v| decode_dir_object(v)).unwrap_or_default(); + let bucket = params.first().cloned().unwrap_or_default().to_owned(); + let object = params.get(1).cloned().map(decode_dir_object).unwrap_or_default(); return Error::new(StorageError::ObjectExistsAsDirectory(bucket, object)); } DiskError::VolumeNotFound => { - let bucket = params.get(0).cloned().unwrap_or_default().to_owned(); + let bucket = params.first().cloned().unwrap_or_default().to_owned(); return Error::new(StorageError::BucketNotFound(bucket)); } DiskError::VolumeNotEmpty => { - let bucket = params.get(0).cloned().unwrap_or_default().to_owned(); + let bucket = params.first().cloned().unwrap_or_default().to_owned(); return Error::new(StorageError::BucketNotEmpty(bucket)); } DiskError::FileAccessDenied => { - let bucket = params.get(0).cloned().unwrap_or_default().to_owned(); - let object = params.get(1).cloned().map(|v| decode_dir_object(v)).unwrap_or_default(); + let bucket = params.first().cloned().unwrap_or_default().to_owned(); + let object = params.get(1).cloned().map(decode_dir_object).unwrap_or_default(); return Error::new(StorageError::PrefixAccessDenied(bucket, object)); } diff --git a/ecstore/src/utils/fs.rs b/ecstore/src/utils/fs.rs index a3012ea28..463989c0f 100644 --- a/ecstore/src/utils/fs.rs +++ b/ecstore/src/utils/fs.rs @@ -123,6 +123,15 @@ pub async fn remove(path: impl AsRef) -> io::Result<()> { } } +pub async fn remove_all(path: impl AsRef) -> io::Result<()> { + let meta = fs::metadata(path.as_ref()).await?; + if meta.is_dir() { + fs::remove_dir_all(path.as_ref()).await + } else { + fs::remove_file(path.as_ref()).await + } +} + pub async fn mkdir(path: impl AsRef) -> io::Result<()> { fs::create_dir(path.as_ref()).await } diff --git a/madmin/src/heal_command.rs b/madmin/src/heal_command.rs index 0e26f100d..0d8471018 100644 --- a/madmin/src/heal_command.rs +++ b/madmin/src/heal_command.rs @@ -37,17 +37,13 @@ pub struct BgHealState { } pub async fn get_local_background_heal_status() -> (BgHealState, bool) { - let (bg_seq, ok) = GLOBAL_BackgroundHealState - .read() - .await - .get_heal_sequence_by_token(BG_HEALING_UUID) - .await; + let (bg_seq, ok) = GLOBAL_BackgroundHealState.get_heal_sequence_by_token(BG_HEALING_UUID).await; if !ok { return (BgHealState::default(), false); } let bg_seq = bg_seq.unwrap(); let mut status = BgHealState { - scanned_items_count: bg_seq.read().await.get_scanned_items_count() as u64, + scanned_items_count: bg_seq.get_scanned_items_count().await as u64, ..Default::default() }; let mut heal_disks_map = HashSet::new(); @@ -56,7 +52,7 @@ pub async fn get_local_background_heal_status() -> (BgHealState, bool) { } let Some(store) = new_object_layer_fn() else { - let healing = GLOBAL_BackgroundHealState.read().await.get_local_healing_disks().await; + let healing = GLOBAL_BackgroundHealState.get_local_healing_disks().await; for disk in healing.values() { status.heal_disks.push(disk.endpoint.clone()); } diff --git a/rustfs/src/admin/handlers.rs b/rustfs/src/admin/handlers.rs index 988a7dc31..bd9f48f29 100644 --- a/rustfs/src/admin/handlers.rs +++ b/rustfs/src/admin/handlers.rs @@ -1,13 +1,22 @@ use super::router::Operation; use crate::storage::error::to_s3_error; +use bytes::Bytes; use ecstore::bucket::policy::action::{Action, ActionSet}; use ecstore::bucket::policy::bucket_policy::{BPStatement, BucketPolicy}; use ecstore::bucket::policy::effect::Effect; use ecstore::bucket::policy::resource::{Resource, ResourceSet}; +use ecstore::error::Error as ec_Error; +use ecstore::global::GLOBAL_ALlHealState; +use ecstore::heal::heal_commands::HealOpts; +use ecstore::heal::heal_ops::new_heal_sequence; +use ecstore::peer::is_reserved_or_invalid_bucket; +use ecstore::store::is_valid_object_prefix; use ecstore::store_api::StorageAPI; +use ecstore::utils::path::path_join; use ecstore::utils::xml; use ecstore::GLOBAL_Endpoints; use ecstore::{new_object_layer_fn, store_api::BackendInfo}; +use http::Uri; use hyper::StatusCode; use matchit::Params; use s3s::S3ErrorCode; @@ -18,8 +27,12 @@ use s3s::{ use serde::{Deserialize, Serialize}; use serde_urlencoded::from_bytes; use std::collections::HashSet; +use std::path::PathBuf; +use std::sync::Arc; use time::{Duration, OffsetDateTime}; -use tracing::warn; +use tokio::spawn; +use tokio::sync::mpsc; +use tracing::{info, warn}; #[derive(Deserialize, Debug, Default)] #[serde(rename_all = "PascalCase", default)] @@ -224,14 +237,168 @@ impl Operation for MetricsHandler { } } +#[derive(Debug, Default, Serialize, Deserialize)] +struct HealInitParams { + bucket: String, + obj_prefix: String, + hs: HealOpts, + client_token: String, + force_start: bool, + force_stop: bool, +} + +fn extract_heal_init_params(body: &Bytes, uri: &Uri, params: Params<'_, '_>) -> S3Result { + let mut hip = HealInitParams { + bucket: params.get("bucket").map(|s| s.to_string()).unwrap_or_default(), + obj_prefix: params.get("prefix").map(|s| s.to_string()).unwrap_or_default(), + ..Default::default() + }; + if hip.bucket.is_empty() && !hip.obj_prefix.is_empty() { + return Err(s3_error!(InvalidRequest, "invalid bucket name")); + } + if is_reserved_or_invalid_bucket(&hip.bucket, false) { + return Err(s3_error!(InvalidRequest, "invalid bucket name")); + } + if !is_valid_object_prefix(&hip.obj_prefix) { + return Err(s3_error!(InvalidRequest, "invalid object name")); + } + + if let Some(query) = uri.query() { + let params: Vec<&str> = query.split('&').collect(); + for param in params { + let mut parts = param.split('='); + if let Some(key) = parts.next() { + if key == "clientToken" { + if let Some(value) = parts.next() { + hip.client_token = value.to_string(); + } + } + if key == "forceStart" { + if parts.next().is_some() { + hip.force_start = true; + } + } + if key == "forceStop" { + if parts.next().is_some() { + hip.force_stop = true; + } + } + } + } + } + + if (hip.force_start && hip.force_stop) || (!hip.client_token.is_empty() && (hip.force_start || hip.force_stop)) { + return Err(s3_error!(InvalidRequest, "")); + } + + if hip.client_token.is_empty() { + hip.hs = serde_json::from_slice(body).map_err(|e| { + info!("err request body parse, err: {:?}", e); + s3_error!(InvalidRequest, "err request body parse") + })?; + } + + Ok(hip) +} + pub struct HealHandler {} #[async_trait::async_trait] impl Operation for HealHandler { - async fn call(&self, _req: S3Request, _params: Params<'_, '_>) -> S3Result> { - warn!("handle HealHandler"); + async fn call(&self, req: S3Request, params: Params<'_, '_>) -> S3Result> { + warn!("handle HealHandler, req: {:?}, params: {:?}", req, params); + let Some(cred) = req.credentials else { return Err(s3_error!(InvalidRequest, "get cred failed")) }; + info!("cred: {:?}", cred); + let mut input = req.input; + let bytes = match input.store_all_unlimited().await { + Ok(b) => b, + Err(e) => { + warn!("get body failed, e: {:?}", e); + return Err(s3_error!(InvalidRequest, "get body failed")); + } + }; + info!("bytes: {:?}", bytes); + let hip = extract_heal_init_params(&bytes, &req.uri, params)?; + info!("body: {:?}", hip); - return Err(s3_error!(NotImplemented)); + #[derive(Default)] + struct HealResp { + resp_bytes: Vec, + _api_err: Option, + _err_body: String, + } + + let heap_path = path_join(&[PathBuf::from(hip.bucket.clone()), PathBuf::from(hip.obj_prefix.clone())]); + if !hip.client_token.is_empty() && !hip.force_start && !hip.force_stop { + match GLOBAL_ALlHealState + .pop_heal_status_json(heap_path.to_str().unwrap_or_default(), &hip.client_token) + .await + { + Ok(b) => { + info!("pop_heal_status_json success"); + return Ok(S3Response::new((StatusCode::OK, Body::from(b)))); + } + Err(_e) => { + info!("pop_heal_status_json failed"); + return Ok(S3Response::new((StatusCode::INTERNAL_SERVER_ERROR, Body::from(vec![])))); + } + } + } + let (tx, mut rx) = mpsc::channel(1); + if hip.force_stop { + let tx_clone = tx.clone(); + spawn(async move { + match GLOBAL_ALlHealState + .stop_heal_sequence(heap_path.to_str().unwrap_or_default()) + .await + { + Ok(b) => { + let _ = tx_clone + .send(HealResp { + resp_bytes: b, + ..Default::default() + }) + .await; + } + Err(e) => { + let _ = tx_clone + .send(HealResp { + _api_err: Some(e), + ..Default::default() + }) + .await; + } + } + }); + } else if hip.client_token.is_empty() { + let nh = Arc::new(new_heal_sequence(&hip.bucket, &hip.obj_prefix, "", hip.hs, hip.force_start)); + let tx_clone = tx.clone(); + spawn(async move { + match GLOBAL_ALlHealState.launch_new_heal_sequence(nh).await { + Ok(b) => { + let _ = tx_clone + .send(HealResp { + resp_bytes: b, + ..Default::default() + }) + .await; + } + Err(e) => { + let _ = tx_clone + .send(HealResp { + _api_err: Some(e), + ..Default::default() + }) + .await; + } + } + }); + } + + match rx.recv().await { + Some(result) => Ok(S3Response::new((StatusCode::OK, Body::from(result.resp_bytes)))), + None => Ok(S3Response::new((StatusCode::INTERNAL_SERVER_ERROR, Body::from(vec![])))), + } } } @@ -516,3 +683,15 @@ impl Operation for RebalanceStop { return Err(s3_error!(NotImplemented)); } } + +#[cfg(test)] +mod test { + use ecstore::heal::heal_commands::HealOpts; + + #[test] + fn test_decode() { + let b = b"{\"recursive\":false,\"dryRun\":false,\"remove\":false,\"recreate\":false,\"scanMode\":1,\"updateParity\":false,\"nolock\":false}"; + let s: HealOpts = serde_urlencoded::from_bytes(b).unwrap(); + println!("{:?}", s); + } +} diff --git a/rustfs/src/admin/mod.rs b/rustfs/src/admin/mod.rs index 673bc1d4c..c7223f840 100644 --- a/rustfs/src/admin/mod.rs +++ b/rustfs/src/admin/mod.rs @@ -2,6 +2,7 @@ pub mod handlers; pub mod router; use common::error::Result; +// use ecstore::global::{is_dist_erasure, is_erasure}; use hyper::Method; use router::{AdminOperation, S3Router}; use s3s::route::S3Route; @@ -53,17 +54,6 @@ pub fn make_admin_route() -> Result { AdminOperation(&handlers::MetricsHandler {}), )?; - r.insert( - Method::POST, - format!("{}{}", ADMIN_PREFIX, "/v3/heal/{bucket}/{prefix}").as_str(), - AdminOperation(&handlers::HealHandler {}), - )?; - r.insert( - Method::POST, - format!("{}{}", ADMIN_PREFIX, "/v3/background-heal/status").as_str(), - AdminOperation(&handlers::BackgroundHealStatusHandler {}), - )?; - r.insert( Method::GET, format!("{}{}", ADMIN_PREFIX, "/v3/pools/list").as_str(), @@ -101,5 +91,19 @@ pub fn make_admin_route() -> Result { AdminOperation(&handlers::RebalanceStop {}), )?; + // Some APIs are only available in EC mode + // if is_dist_erasure().await || is_erasure().await { + r.insert( + Method::POST, + format!("{}{}", ADMIN_PREFIX, "/v3/heal/{bucket}/{prefix}").as_str(), + AdminOperation(&handlers::HealHandler {}), + )?; + r.insert( + Method::POST, + format!("{}{}", ADMIN_PREFIX, "/v3/background-heal/status").as_str(), + AdminOperation(&handlers::BackgroundHealStatusHandler {}), + )?; + // } + Ok(r) } diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index 9ddde2bfc..92460bd24 100644 --- a/rustfs/src/main.rs +++ b/rustfs/src/main.rs @@ -10,7 +10,7 @@ use clap::Parser; use common::error::{Error, Result}; use ecstore::{ endpoints::EndpointServerPools, - heal::data_scanner::init_data_scanner, + heal::{background_heal_ops::init_auto_heal, data_scanner::init_data_scanner}, set_global_endpoints, store::{init_local_disks, ECStore}, update_erasure_type, @@ -193,6 +193,10 @@ async fn run(opt: config::Opt) -> Result<()> { warn!(" init store success!"); // init scanner init_data_scanner().await; + // init auto heal + init_auto_heal().await; + + info!("server was started"); tokio::select! { _ = tokio::signal::ctrl_c() => {