gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-29 00:17:11 +00:00
Code Issues Packages Projects Releases Wiki Activity

auto heal(2)

Browse Source 
Signed-off-by: junxiang Mu <1948535941@qq.com>
This commit is contained in:
junxiang Mu
2024-11-17 18:34:29 +08:00
parent 01e0e4b673
commit 6291087ecb
12 changed files with 1855 additions and 532 deletions
Download Patch File Download Diff File Expand all files Collapse all files
ecstore/src/heal/background_heal_ops.rs
+153 -14
View File
@@ -1,18 +1,17 @@
use std::{env, sync::Arc};
use std::{cmp::Ordering, env, path::PathBuf, sync::Arc, time::Duration};
use tokio::sync::{
mpsc::{self, Receiver, Sender},
RwLock,
use tokio::{
sync::{
mpsc::{self, Receiver, Sender},
RwLock,
},
time::interval,
};
use tracing::info;
use uuid::Uuid;
use crate::{
endpoints::Endpoints,
error::{Error, Result},
global::{GLOBAL_BackgroundHealRoutine, GLOBAL_BackgroundHealState, GLOBAL_LOCAL_DISK_MAP},
heal::heal_ops::NOP_HEAL,
new_object_layer_fn,
store_api::StorageAPI,
utils::path::SLASH_SEPARATOR,
config::RUSTFS_CONFIG_PREFIX, disk::{endpoint::Endpoint, error::DiskError, DiskAPI, DiskInfoOptions, BUCKET_META_PREFIX, RUSTFS_META_BUCKET}, error::{Error, Result}, global::{GLOBAL_BackgroundHealRoutine, GLOBAL_BackgroundHealState, GLOBAL_LOCAL_DISK_MAP}, heal::{data_usage::{DATA_USAGE_CACHE_NAME, DATA_USAGE_ROOT}, data_usage_cache::DataUsageCache, heal_commands::{init_healing_tracker, load_healing_tracker}, heal_ops::NOP_HEAL}, new_object_layer_fn, store::get_disk_via_endpoint, store_api::{BucketInfo, BucketOptions, StorageAPI}, utils::path::{path_join, SLASH_SEPARATOR}
};
use super::{
@@ -20,11 +19,17 @@ use super::{
heal_ops::{new_bg_heal_sequence, HealSequence},
};
pub static DEFAULT_MONITOR_NEW_DISK_INTERVAL: Duration = Duration::from_secs(10);
pub async fn init_auto_heal() {
init_background_healing().await;
if let Ok(v) = env::var("_RUSTFS_AUTO_DRIVE_HEALING") {
if v == "on" {
// GLOBAL_BackgroundHealState.write().await.push_heal_local_disks(heal_local_disks).await;
GLOBAL_BackgroundHealState
.write()
.await
.push_heal_local_disks(&get_local_disks_to_heal().await)
.await;
}
}
}
@@ -44,8 +49,142 @@ async fn init_background_healing() {
.await;
}
async fn get_local_disks_to_heal() -> Endpoints {
for (_, disk) in GLOBAL_LOCAL_DISK_MAP.read().await.iter() {}
async fn get_local_disks_to_heal() -> Vec<Endpoint> {
let mut disks_to_heal = Vec::new();
for (_, disk) in GLOBAL_LOCAL_DISK_MAP.read().await.iter() {
if let Some(disk) = disk {
if let Err(err) = disk.disk_info(&DiskInfoOptions::default()).await {
if let Some(DiskError::UnformattedDisk) = err.downcast_ref() {
disks_to_heal.push(disk.endpoint());
}
}
let h = disk.healing().await;
if let Some(h) = h {
if !h.finished {
disks_to_heal.push(disk.endpoint());
}
}
}
}
// todo
// if disks_to_heal.len() == GLOBAL_Endpoints.read().await.n {
// }
disks_to_heal
}
async fn monitor_local_disks_and_heal() {
let mut interval = interval(DEFAULT_MONITOR_NEW_DISK_INTERVAL);
loop {
interval.tick().await;
let heal_disks = GLOBAL_BackgroundHealState.read().await.get_heal_local_disk_endpoints().await;
if heal_disks.is_empty() {
interval.reset();
continue;
}
let layer = new_object_layer_fn();
let lock = layer.read().await;
let store = lock.as_ref().expect("errServerNotInitialized");
if let (_, Some(err)) = store.heal_format(false).await.expect("heal format failed") {
if let Some(DiskError::NoHealRequired) = err.downcast_ref() {
} else {
info!("heal format err: {}", err.to_string());
interval.reset();
continue;
}
}
for disk in heal_disks.into_ref().iter() {
let disk_clone = disk.clone();
tokio::spawn(async move {
GLOBAL_BackgroundHealState.write().await.set_disk_healing_status(disk_clone.clone(), true).await;
});
}
}
}
async fn heal_fresh_disk(endpoint: &Endpoint) -> Result<()> {
let (pool_idx, set_idx) = (endpoint.pool_idx.unwrap(), endpoint.disk_idx.unwrap());
let disk = match get_disk_via_endpoint(endpoint).await {
Some(disk) => disk,
None => return Err(Error::from_string(format!("Unexpected error disk must be initialized by now after formatting: {}", endpoint.to_string()))),
};
if let Err(err) = disk.disk_info(&DiskInfoOptions::default()).await {
match err.downcast_ref() {
Some(DiskError::DriveIsRoot) => {
return Ok(());
},
Some(DiskError::UnformattedDisk) => {
},
_ => {
return Err(err);
}
}
}
let mut tracker = match load_healing_tracker(&Some(disk.clone())).await {
Ok(tracker) => tracker,
Err(err) => {
match err.downcast_ref() {
Some(DiskError::FileNotFound) => {
return Ok(());
}
_ => {
info!("Unable to load healing tracker on '{}': {}, re-initializing..", disk.to_string(), err.to_string());
},
}
init_healing_tracker(disk.clone(), &Uuid::new_v4().to_string()).await?
}
};
info!("Healing drive '{}' - 'mc admin heal alias/ --verbose' to check the current status.", endpoint.to_string());
let layer = new_object_layer_fn();
let lock = layer.read().await;
let store = match lock.as_ref() {
Some(s) => s,
None => return Err(Error::msg("errServerNotInitialized")),
};
let mut buckets = store.list_bucket(&BucketOptions::default()).await?;
buckets.push(BucketInfo {
name: path_join(&[PathBuf::from(RUSTFS_META_BUCKET), PathBuf::from(RUSTFS_CONFIG_PREFIX)]).to_string_lossy().to_string(),
..Default::default()
});
buckets.push(BucketInfo {
name: path_join(&[PathBuf::from(RUSTFS_META_BUCKET), PathBuf::from(BUCKET_META_PREFIX)]).to_string_lossy().to_string(),
..Default::default()
});
buckets.sort_by(|a, b| {
let a_has_prefix = a.name.starts_with(RUSTFS_META_BUCKET);
let b_has_prefix = b.name.starts_with(RUSTFS_META_BUCKET);
match (a_has_prefix, b_has_prefix) {
(true, false) => Ordering::Less,
(false, true) => Ordering::Greater,
_ => b.created.cmp(&a.created),
}
});
let mut cache = match DataUsageCache::load(&store.pools[pool_idx].disk_set[set_idx], DATA_USAGE_CACHE_NAME).await {
Ok(cache) => {
let data_usage_info = cache.dui(DATA_USAGE_ROOT, &Vec::new());
tracker.objects_total_count = data_usage_info.objects_total_count;
tracker.objects_total_size = data_usage_info.objects_total_size;
cache
},
Err(_) => DataUsageCache::default()
};
let location = disk.get_disk_location();
tracker.set_queue_buckets(&buckets).await;
tracker.save().await?;
// store.pools[pool_idx].disk_set[set_idx].
todo!()
}
ecstore/src/heal/heal_commands.rs
+3 -4
View File
@@ -161,8 +161,7 @@ pub struct HealingTracker {
impl HealingTracker {
pub fn marshal_msg(&self) -> Result<Vec<u8>> {
serde_json::to_string(self)
.map(|s| s.as_bytes().to_vec())
serde_json::to_vec(self)
.map_err(|err| Error::from_string(err.to_string()))
}
@@ -336,7 +335,7 @@ impl HealingTracker {
self.queue_buckets.retain(|x| x != bucket);
}
async fn set_queue_buckets(&mut self, buckets: &[BucketInfo]) {
pub async fn set_queue_buckets(&mut self, buckets: &[BucketInfo]) {
let _ = self.mu.write().await;
buckets.iter().for_each(|bucket| {
@@ -417,7 +416,7 @@ impl Clone for HealingTracker {
}
}
async fn load_healing_tracker(disk: &Option<DiskStore>) -> Result<HealingTracker> {
pub async fn load_healing_tracker(disk: &Option<DiskStore>) -> Result<HealingTracker> {
if let Some(disk) = disk {
let disk_id = disk.get_disk_id().await?;
if let Some(disk_id) = disk_id {
ecstore/src/heal/heal_ops.rs
+6 -6
View File
@@ -406,7 +406,7 @@ impl AllHealState {
hstate
}
async fn pop_heal_local_disks(&mut self, heal_local_disks: &[Endpoint]) {
pub async fn pop_heal_local_disks(&mut self, heal_local_disks: &[Endpoint]) {
let _ = self.mu.write().await;
self.heal_local_disks.retain(|k, _| {
@@ -433,7 +433,7 @@ impl AllHealState {
self.heal_status.insert(tracker.id.clone(), tracker.clone());
}
async fn get_local_healing_disks(&self) -> HashMap<String, HealingDisk> {
pub async fn get_local_healing_disks(&self) -> HashMap<String, HealingDisk> {
let _ = self.mu.read().await;
let mut dst = HashMap::new();
@@ -444,7 +444,7 @@ impl AllHealState {
dst
}
async fn get_heal_local_disk_endpoints(&self) -> Endpoints {
pub async fn get_heal_local_disk_endpoints(&self) -> Endpoints {
let _ = self.mu.read().await;
let mut endpoints = Vec::new();
@@ -471,7 +471,7 @@ impl AllHealState {
});
}
async fn periodic_heal_seqs_clean(&mut self) {
pub async fn periodic_heal_seqs_clean(&mut self) {
let _ = self.mu.write().await;
let now = SystemTime::now();
@@ -500,13 +500,13 @@ impl AllHealState {
(None, false)
}
async fn get_heal_sequence(&self, path: &str) -> Option<Arc<RwLock<HealSequence>>> {
pub async fn get_heal_sequence(&self, path: &str) -> Option<Arc<RwLock<HealSequence>>> {
let _ = self.mu.read().await;
self.heal_seq_map.get(path).cloned()
}
async fn stop_heal_sequence(&mut self, path: &str) -> Result<Vec<u8>> {
pub async fn stop_heal_sequence(&mut 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;