From 1a418c74b4730d2d2452543df0d8ad2a892c2dff Mon Sep 17 00:00:00 2001 From: mujunxiang <1948535941@qq.com> Date: Sat, 30 Nov 2024 21:29:37 +0800 Subject: [PATCH] support mc admin heal command Signed-off-by: mujunxiang <1948535941@qq.com> --- Cargo.lock | 13 ++ ecstore/Cargo.toml | 1 + ecstore/src/admin_server_info.rs | 6 +- ecstore/src/bucket/versioning_sys.rs | 6 + ecstore/src/disk/local.rs | 2 +- ecstore/src/erasure.rs | 2 +- ecstore/src/global.rs | 9 +- ecstore/src/heal/background_heal_ops.rs | 20 +-- ecstore/src/heal/data_scanner.rs | 6 +- ecstore/src/heal/heal_commands.rs | 41 +++++- ecstore/src/heal/heal_ops.rs | 129 +++++++++++----- ecstore/src/peer.rs | 8 +- ecstore/src/pools.rs | 30 ++-- ecstore/src/sets.rs | 12 +- ecstore/src/store.rs | 22 ++- ecstore/src/store_err.rs | 26 ++-- madmin/src/heal_command.rs | 8 +- rustfs/src/admin/handlers.rs | 187 +++++++++++++++++++++++- rustfs/src/admin/mod.rs | 2 +- 19 files changed, 388 insertions(+), 142 deletions(-) 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/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/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/disk/local.rs b/ecstore/src/disk/local.rs index 29967506d..40e05fb84 100644 --- a/ecstore/src/disk/local.rs +++ b/ecstore/src/disk/local.rs @@ -1842,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; diff --git a/ecstore/src/erasure.rs b/ecstore/src/erasure.rs index e38df8821..9691a763a 100644 --- a/ecstore/src/erasure.rs +++ b/ecstore/src/erasure.rs @@ -10,8 +10,8 @@ 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, info}; // use tracing::debug; use uuid::Uuid; diff --git a/ecstore/src/global.rs b/ecstore/src/global.rs index 5bb5f0153..69e3bc218 100644 --- a/ecstore/src/global.rs +++ b/ecstore/src/global.rs @@ -29,7 +29,8 @@ lazy_static! { 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_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 c3f821bba..fa9a1f63d 100644 --- a/ecstore/src/heal/background_heal_ops.rs +++ b/ecstore/src/heal/background_heal_ops.rs @@ -39,8 +39,6 @@ pub async fn init_auto_heal() { 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 { @@ -58,11 +56,7 @@ async fn init_background_healing() { 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 { @@ -95,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; @@ -115,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(); diff --git a/ecstore/src/heal/data_scanner.rs b/ecstore/src/heal/data_scanner.rs index 3f1656969..61efee453 100644 --- a/ecstore/src/heal/data_scanner.rs +++ b/ecstore/src/heal/data_scanner.rs @@ -719,11 +719,7 @@ impl FolderScanner { 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 'outer; } 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 9a450fe0d..73d7f82f7 100644 --- a/ecstore/src/heal/heal_ops.rs +++ b/ecstore/src/heal/heal_ops.rs @@ -7,7 +7,6 @@ use super::{ HEAL_ITEM_BUCKET_METADATA, }, }; -use crate::heal::heal_commands::{HEAL_ITEM_BUCKET, HEAL_ITEM_OBJECT}; use crate::store_api::StorageAPI; use crate::{ config::common::CONFIG_PREFIX, @@ -27,12 +26,18 @@ use crate::{ new_object_layer_fn, utils::path::has_profix, }; +use crate::{ + heal::heal_commands::{HEAL_ITEM_BUCKET, HEAL_ITEM_OBJECT}, + 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}, @@ -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, @@ -103,7 +108,7 @@ pub struct HealSequence { pub force_started: bool, pub setting: HealOpts, pub current_status: Arc>, - pub last_sent_result_index: usize, + pub last_sent_result_index: RwLock, pub scanned_items_map: RwLock, pub healed_items_map: RwLock, pub heal_failed_items_map: RwLock, @@ -151,7 +156,7 @@ pub fn new_heal_sequence(bucket: &str, obj_prefix: &str, client_addr: &str, hs: client_token, client_address: client_addr.to_string(), force_started: force_start, - setting: hs.clone(), + setting: hs, current_status: Arc::new(RwLock::new(HealSequenceStatus { summary: HEAL_NOT_STARTED_STATUS.to_string(), heal_setting: hs, @@ -300,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); @@ -369,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 {}", @@ -563,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(); @@ -583,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; } } } @@ -593,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; } @@ -604,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; } @@ -613,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); } @@ -635,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()); } @@ -644,39 +688,39 @@ 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() { + 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) { let _ = self.mu.read().await; - for v in self.heal_seq_map.values() { + for v in self.heal_seq_map.read().await.values() { if v.client_token == token { return (Some(v.clone()), true); } @@ -688,10 +732,10 @@ impl AllHealState { 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 client_token = he.client_token.clone(); @@ -701,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; @@ -714,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(); } @@ -733,8 +777,11 @@ 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 path = Path::new(&heal_sequence.bucket).join(heal_sequence.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 heal_sequence.force_started { self.stop_heal_sequence(path_s).await?; @@ -746,7 +793,7 @@ impl AllHealState { let _ = self.mu.write().await; - for (k, v) in self.heal_seq_map.iter() { + 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: {}", @@ -755,7 +802,10 @@ 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 = heal_sequence.client_token.clone(); if *GLOBAL_IsDistErasure.read().await { @@ -774,7 +824,8 @@ impl AllHealState { let b = serde_json::to_vec(&HealStartSuccess { client_token, client_address: heal_sequence.client_address.clone(), - start_time: heal_sequence.start_time, + // 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 53ac12d73..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::{error, info, 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::{ @@ -716,7 +716,7 @@ pub async fn heal_bucket_local(bucket: &str, opts: &HealOpts) -> Result 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 b68e3283d..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()); } } diff --git a/ecstore/src/store.rs b/ecstore/src/store.rs index 14c9ca03a..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)))); 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/madmin/src/heal_command.rs b/madmin/src/heal_command.rs index b68607c7a..0d8471018 100644 --- a/madmin/src/heal_command.rs +++ b/madmin/src/heal_command.rs @@ -37,11 +37,7 @@ 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); } @@ -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 ec2982fa5..c7223f840 100644 --- a/rustfs/src/admin/mod.rs +++ b/rustfs/src/admin/mod.rs @@ -2,7 +2,7 @@ pub mod handlers; pub mod router; use common::error::Result; -use ecstore::global::{is_dist_erasure, is_erasure}; +// use ecstore::global::{is_dist_erasure, is_erasure}; use hyper::Method; use router::{AdminOperation, S3Router}; use s3s::route::S3Route;