gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-27 00:38:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

Merge pull request #135 from rustfs/heal

Browse Source 
add mc admin heal command support
This commit is contained in:
junxiangMu
2024-12-01 14:07:03 +08:00
committed by GitHub
parent 98ed93133b 1a418c74b4
commit 48d1e6ccb2
28 changed files with 649 additions and 324 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Cargo.lock
Generated
+13
View File
@@ -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",
common/protos/src/generated/proto_gen/node_service.rs
-1
View File
@@ -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")]
ecstore/Cargo.toml
+1
View File
@@ -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
ecstore/src/admin_server_info.rs
+3 -3
View File
@@ -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());
}
}
}
ecstore/src/bitrot.rs
+11 -3
View File
@@ -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));
}
}
ecstore/src/bucket/versioning_sys.rs
+6
View File
@@ -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 {}
ecstore/src/cache_value/metacache_set.rs
+1 -1
View File
@@ -139,7 +139,7 @@ pub async fn list_path_raw(mut rx: B_Receiver<bool>, opts: ListPathRawOptions) -
});
}
let errs: Vec<Option<Error>> = Vec::with_capacity(readers.len());
let errs: Vec<Option<Error>> = vec![None; readers.len()];
loop {
let mut current = MetaCacheEntry::default();
let (mut at_eof, mut has_err, mut agree) = (0, 0, 0);
ecstore/src/disk/error.rs
+9 -6
View File
@@ -419,15 +419,18 @@ pub fn convert_access_error(e: io::Error, per_err: DiskError) -> Error {
}
pub fn is_all_not_found(errs: &[Option<Error>]) -> bool {
for err in errs.iter().flatten() {
if let Some(err) = err.downcast_ref::<DiskError>() {
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::<DiskError>() {
match err {
DiskError::FileNotFound | DiskError::VolumeNotFound | &DiskError::FileVersionNotFound => {
continue;
}
_ => return false,
}
_ => return false,
}
}
return false;
}
!errs.is_empty()
ecstore/src/disk/local.rs
+12 -3
View File
@@ -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
ecstore/src/disk/mod.rs
+1
View File
@@ -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,
ecstore/src/disk/os.rs
+18 -2
View File
@@ -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);
}
ecstore/src/erasure.rs
+13 -3
View File
@@ -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::<Vec<_>>();
let shards: Vec<Vec<u8>> = bufs.into_iter().flatten().collect::<Vec<_>>();
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);
}
}
}
}
ecstore/src/global.rs
+4 -7
View File
@@ -28,8 +28,9 @@ lazy_static! {
pub static ref GLOBAL_LOCAL_DISK_SET_DRIVES: Arc<RwLock<TypeLocalDiskSetDrives>> = Arc::new(RwLock::new(Vec::new()));
pub static ref GLOBAL_Endpoints: OnceLock<EndpointServerPools> = OnceLock::new();
pub static ref GLOBAL_RootDiskThreshold: RwLock<u64> = RwLock::new(0);
pub static ref GLOBAL_BackgroundHealRoutine: Arc<RwLock<HealRoutine>> = HealRoutine::new();
pub static ref GLOBAL_BackgroundHealState: Arc<RwLock<AllHealState>> = AllHealState::new(false);
pub static ref GLOBAL_BackgroundHealRoutine: Arc<HealRoutine> = HealRoutine::new();
pub static ref GLOBAL_BackgroundHealState: Arc<AllHealState> = AllHealState::new(false);
pub static ref GLOBAL_ALlHealState: Arc<AllHealState> = AllHealState::new(false);
static ref globalDeploymentIDPtr: RwLock<Uuid> = RwLock::new(Uuid::nil());
}
@@ -50,11 +51,7 @@ pub fn set_global_endpoints(eps: Vec<PoolEndpoints>) {
}
pub fn new_object_layer_fn() -> Option<Arc<ECStore>> {
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<ECStore>) {
ecstore/src/heal/background_heal_ops.rs
+23 -30
View File
@@ -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<Endpoint> {
@@ -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<Error>,
@@ -324,12 +312,12 @@ pub struct HealResult {
pub struct HealRoutine {
pub tasks_tx: Sender<HealTask>,
tasks_rx: Receiver<HealTask>,
tasks_rx: RwLock<Receiver<HealTask>>,
workers: usize,
}
impl HealRoutine {
pub fn new() -> Arc<RwLock<Self>> {
pub fn new() -> Arc<Self> {
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::<usize>() {
@@ -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<RwLock<HealSequence>>) {
pub async fn add_worker(&self, bgseq: Arc<HealSequence>) {
loop {
let mut d_res = HealResultItem::default();
let d_err: Option<Error>;
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;
}
}
}
}
ecstore/src/heal/data_scanner.rs
+6 -19
View File
@@ -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);
ecstore/src/heal/heal_commands.rs
+34 -7
View File
@@ -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<usize>,
pub set: Option<usize>,
}
#[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<HealDriveInfo>,
}
#[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<HealDriveInfo>,
pub after: Vec<HealDriveInfo>,
#[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<Utc>,
}
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);
ecstore/src/heal/heal_ops.rs
+216 -145
View File
@@ -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<HealOpts>,
}
#[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<RwLock<HealSequenceStatus>>,
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<usize>,
pub scanned_items_map: RwLock<ItemsMap>,
pub healed_items_map: RwLock<ItemsMap>,
pub heal_failed_items_map: RwLock<ItemsMap>,
pub last_heal_activity: RwLock<SystemTime>,
traverse_and_heal_done_tx: Arc<RwLock<M_Sender<Option<Error>>>>,
traverse_and_heal_done_rx: Arc<RwLock<M_Receiver<Option<Error>>>>,
tx: Arc<RwLock<Sender<bool>>>,
rx: Arc<RwLock<Receiver<bool>>>,
tx: W_Sender<bool>,
rx: W_Receiver<bool>,
}
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<RwLock<HealSequence>>) -> Result<()> {
async fn heal_disk_meta(h: Arc<HealSequence>) -> Result<()> {
HealSequence::heal_rustfs_sys_meta(h, CONFIG_PREFIX).await
}
async fn heal_items(h: Arc<RwLock<HealSequence>>, buckets_only: bool) -> Result<()> {
if h.read().await.client_token == *BG_HEALING_UUID {
async fn heal_items(h: Arc<HealSequence>, 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<RwLock<HealSequence>>) {
async fn traverse_and_heal(h: Arc<HealSequence>) {
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<RwLock<HealSequence>>, meta_prefix: &str) -> Result<()> {
async fn heal_rustfs_sys_meta(h: Arc<HealSequence>, 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<RwLock<HealSequence>>, bucket: &str, bucket_only: bool) -> Result<()> {
pub async fn heal_bucket(hs: Arc<HealSequence>, 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<RwLock<HealSequence>>,
hs: Arc<HealSequence>,
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<RwLock<HealSequence>>,
hs: Arc<HealSequence>,
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<RwLock<HealSequence>>) {
let r = h.read().await;
pub async fn heal_sequence_start(h: Arc<HealSequence>) {
{
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<RwLock<HealSequence>>) {
});
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<RwLock<HealSequence>>) {
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<RwLock<HealSequence>>) {
pub struct AllHealState {
mu: RwLock<bool>,
heal_seq_map: HashMap<String, Arc<RwLock<HealSequence>>>,
heal_local_disks: HashMap<Endpoint, bool>,
heal_status: HashMap<String, HealingTracker>,
heal_seq_map: RwLock<HashMap<String, Arc<HealSequence>>>,
heal_local_disks: RwLock<HashMap<Endpoint, bool>>,
heal_status: RwLock<HashMap<String, HealingTracker>>,
}
impl AllHealState {
pub fn new(cleanup: bool) -> Arc<RwLock<Self>> {
let hstate = Arc::new(RwLock::new(AllHealState::default()));
pub fn new(cleanup: bool) -> Arc<Self> {
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::<Vec<_>>();
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<Vec<u8>> {
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<String, HealingDisk> {
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<Arc<RwLock<HealSequence>>>, bool) {
pub async fn get_heal_sequence_by_token(&self, token: &str) -> (Option<Arc<HealSequence>>, 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<Arc<RwLock<HealSequence>>> {
pub async fn get_heal_sequence(&self, path: &str) -> Option<Arc<HealSequence>> {
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<Vec<u8>> {
pub async fn stop_heal_sequence(&self, path: &str) -> Result<Vec<u8>> {
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<RwLock<HealSequence>>) -> Result<Vec<u8>> {
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<HealSequence>) -> Result<Vec<u8>> {
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)
}
ecstore/src/peer.rs
+11 -8
View File
@@ -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<HealResu
}
for (disk, state) in disks.iter().zip(before_state.read().await.iter()) {
res.before.push(HealDriveInfo {
res.before.drives.push(HealDriveInfo {
uuid: "".to_string(),
endpoint: disk.clone().map(|s| s.to_string()).unwrap_or_default(),
state: state.to_string(),
});
}
if !is_rustfs_meta_bucket_name(bucket) && !is_all_buckets_not_found(&errs) && opts.remove {
if opts.remove && !is_rustfs_meta_bucket_name(bucket) && !is_all_buckets_not_found(&errs) {
let mut futures = Vec::new();
for disk in disks.iter() {
let disk = disk.clone();
let bucket = bucket.to_string();
info!("heal_bucket_local, errs: {:?}, opts: {:?}", errs, opts);
futures.push(async move {
match disk {
Some(disk) => {
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<HealResu
let errs_clone = errs.clone();
futures.push(async move {
if bs_clone.read().await[idx] == DRIVE_STATE_MISSING {
info!("bucket not find, will recreate");
match disk.as_ref().unwrap().make_volume(&bucket).await {
Ok(_) => {
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<HealResu
}
for (disk, state) in disks.iter().zip(after_state.read().await.iter()) {
res.before.push(HealDriveInfo {
res.before.drives.push(HealDriveInfo {
uuid: "".to_string(),
endpoint: disk.clone().map(|s| s.to_string()).unwrap_or_default(),
state: state.to_string(),
ecstore/src/pools.rs
+14 -16
View File
@@ -114,7 +114,7 @@ impl PoolMeta {
data.write_all(&buf)?;
for pool in pools {
save_config(pool, &POOL_META_NAME, &data).await?;
save_config(pool, POOL_META_NAME, &data).await?;
}
Ok(())
@@ -201,7 +201,7 @@ impl PoolMeta {
}
let now = OffsetDateTime::now_utc();
pool.last_update = now.clone();
pool.last_update = now;
pool.decommission = Some(PoolDecommissionInfo {
start_time: Some(now),
start_size: pi.free,
@@ -271,7 +271,7 @@ impl PoolMeta {
}
fn path2_bucket_object(name: &String) -> (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<S: StorageAPI>(&self, _idx: usize, _pool: Arc<S>, 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<StorageDisk>, 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<StorageDisk>, info: &StorageInfo) -> us
capacity
}
fn get_total_usable_capacity_free(disks: &Vec<StorageDisk>, 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 {
ecstore/src/sets.rs
+7 -7
View File
@@ -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<RwLock<HealSequence>>,
_hs: Arc<HealSequence>,
_is_meta: bool,
) -> Result<()> {
unimplemented!()
ecstore/src/store.rs
+15 -17
View File
@@ -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<PoolObjInfo>, opts: &ObjectOptions) -> Vec<PoolErr> {
async fn pools_with_object(&self, pools: &[PoolObjInfo], opts: &ObjectOptions) -> Vec<PoolErr> {
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<Vec<BucketInfo>> {
@@ -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<Error>)> {
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<RwLock<HealSequence>>,
hs: Arc<HealSequence>,
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
}
ecstore/src/store_api.rs
+3 -10
View File
@@ -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<u8>,
}
@@ -1023,14 +1022,8 @@ pub trait StorageAPI: ObjectIO {
version_id: &str,
opts: &HealOpts,
) -> Result<(HealResultItem, Option<Error>)>;
async fn heal_objects(
&self,
bucket: &str,
prefix: &str,
opts: &HealOpts,
hs: Arc<RwLock<HealSequence>>,
is_meta: bool,
) -> Result<()>;
async fn heal_objects(&self, bucket: &str, prefix: &str, opts: &HealOpts, hs: Arc<HealSequence>, is_meta: bool)
-> Result<()>;
async fn get_pool_and_set(&self, id: &str) -> Result<(Option<usize>, Option<usize>, Option<usize>)>;
async fn check_abandoned_parts(&self, bucket: &str, object: &str, opts: &HealOpts) -> Result<()>;
}
ecstore/src/store_err.rs
+13 -13
View File
@@ -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));
}
ecstore/src/utils/fs.rs
+9
View File
@@ -123,6 +123,15 @@ pub async fn remove(path: impl AsRef<Path>) -> io::Result<()> {
}
}
pub async fn remove_all(path: impl AsRef<Path>) -> 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<Path>) -> io::Result<()> {
fs::create_dir(path.as_ref()).await
}
madmin/src/heal_command.rs
+3 -7
View File
@@ -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());
}
rustfs/src/admin/handlers.rs
+183 -4
View File
@@ -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<HealInitParams> {
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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
warn!("handle HealHandler");
async fn call(&self, req: S3Request<Body>, params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
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<u8>,
_api_err: Option<ec_Error>,
_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);
}
}
rustfs/src/admin/mod.rs
+15 -11
View File
@@ -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<impl S3Route> {
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<impl S3Route> {
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)
}
rustfs/src/main.rs
+5 -1
View File
@@ -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() => {