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

Chore: remove dirty file(heal)

Browse Source 
Signed-off-by: junxiang Mu <1948535941@qq.com>
This commit is contained in:
junxiang Mu
2025-07-24 11:15:00 +08:00
parent 4fefd63a5b
commit 2ce7e01f55
11 changed files with 0 additions and 5585 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/heal/background_heal_ops.rs
-512
View File
@@ -1,512 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use futures::future::join_all;
use rustfs_madmin::heal_commands::HealResultItem;
use rustfs_utils::path::{SLASH_SEPARATOR, path_join};
use std::{cmp::Ordering, env, path::PathBuf, sync::Arc, time::Duration};
use tokio::{
spawn,
sync::{
RwLock,
mpsc::{self, Receiver, Sender},
},
time::interval,
};
use tokio_util::sync::CancellationToken;
use tracing::{error, info};
use uuid::Uuid;
use super::{
heal_commands::HealOpts,
heal_ops::{HealSequence, new_bg_heal_sequence},
};
use crate::error::{Error, Result};
use crate::global::get_background_services_cancel_token;
use crate::heal::error::ERR_RETRY_HEALING;
use crate::heal::heal_commands::{HEAL_ITEM_BUCKET, HealScanMode};
use crate::heal::heal_ops::{BG_HEALING_UUID, HealSource};
use crate::{
config::RUSTFS_CONFIG_PREFIX,
disk::{BUCKET_META_PREFIX, DiskAPI, DiskInfoOptions, RUSTFS_META_BUCKET, endpoint::Endpoint, error::DiskError},
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},
};
pub static DEFAULT_MONITOR_NEW_DISK_INTERVAL: Duration = Duration::from_secs(10);
pub async fn init_auto_heal() {
info!("Initializing auto heal background task");
let Some(cancel_token) = get_background_services_cancel_token() else {
error!("Background services cancel token not initialized");
return;
};
init_background_healing().await;
let v = env::var("_RUSTFS_AUTO_DRIVE_HEALING").unwrap_or("on".to_string());
if v == "on" {
info!("start monitor local disks and heal");
GLOBAL_BackgroundHealState
.push_heal_local_disks(&get_local_disks_to_heal().await)
.await;
let cancel_clone = cancel_token.clone();
spawn(async move {
monitor_local_disks_and_heal(cancel_clone).await;
});
}
// let cancel_clone = cancel_token.clone();
// spawn(async move {
// GLOBAL_MRFState.heal_routine_with_cancel(cancel_clone).await;
// });
}
async fn init_background_healing() {
let bg_seq = Arc::new(new_bg_heal_sequence());
for _ in 0..GLOBAL_BackgroundHealRoutine.workers {
let bg_seq_clone = bg_seq.clone();
spawn(async {
GLOBAL_BackgroundHealRoutine.add_worker(bg_seq_clone).await;
});
}
let _ = GLOBAL_BackgroundHealState.launch_new_heal_sequence(bg_seq).await;
}
pub 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 err == DiskError::UnformattedDisk {
info!("get_local_disks_to_heal, disk is unformatted: {}", err);
disks_to_heal.push(disk.endpoint());
}
}
let h = disk.healing().await;
if let Some(h) = h {
if !h.finished {
info!("get_local_disks_to_heal, disk healing not 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(cancel_token: CancellationToken) {
info!("Auto heal monitor started");
let mut interval = interval(DEFAULT_MONITOR_NEW_DISK_INTERVAL);
loop {
tokio::select! {
_ = cancel_token.cancelled() => {
info!("Auto heal monitor received shutdown signal, exiting gracefully");
break;
}
_ = interval.tick() => {
let heal_disks = GLOBAL_BackgroundHealState.get_heal_local_disk_endpoints().await;
if heal_disks.is_empty() {
info!("heal local disks is empty");
interval.reset();
continue;
}
info!("heal local disks: {:?}", heal_disks);
let store = new_object_layer_fn().expect("errServerNotInitialized");
if let (_result, Some(err)) = store.heal_format(false).await.expect("heal format failed") {
error!("heal local disk format error: {}", err);
if err == Error::NoHealRequired {
} else {
info!("heal format err: {}", err.to_string());
interval.reset();
continue;
}
}
let mut futures = Vec::new();
for disk in heal_disks.into_ref().iter() {
let disk_clone = disk.clone();
let cancel_clone = cancel_token.clone();
futures.push(async move {
let disk_for_cancel = disk_clone.clone();
tokio::select! {
_ = cancel_clone.cancelled() => {
info!("Disk healing task cancelled for disk: {}", disk_for_cancel);
}
_ = async {
GLOBAL_BackgroundHealState
.set_disk_healing_status(disk_clone.clone(), true)
.await;
if heal_fresh_disk(&disk_clone).await.is_err() {
info!("heal_fresh_disk is err");
GLOBAL_BackgroundHealState
.set_disk_healing_status(disk_clone.clone(), false)
.await;
}
GLOBAL_BackgroundHealState.pop_heal_local_disks(&[disk_clone]).await;
} => {}
}
});
}
let _ = join_all(futures).await;
interval.reset();
}
}
}
}
async fn heal_fresh_disk(endpoint: &Endpoint) -> Result<()> {
let (pool_idx, set_idx) = (endpoint.pool_idx as usize, endpoint.set_idx as usize);
let disk = match get_disk_via_endpoint(endpoint).await {
Some(disk) => disk,
None => {
return Err(Error::other(format!(
"Unexpected error disk must be initialized by now after formatting: {endpoint}"
)));
}
};
if let Err(err) = disk.disk_info(&DiskInfoOptions::default()).await {
match err {
DiskError::DriveIsRoot => {
return Ok(());
}
DiskError::UnformattedDisk => {}
_ => {
return Err(err.into());
}
}
}
let mut tracker = match load_healing_tracker(&Some(disk.clone())).await {
Ok(tracker) => tracker,
Err(err) => {
match err {
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 Some(store) = new_object_layer_fn() else {
return Err(Error::other("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),
}
});
if let Ok(cache) = DataUsageCache::load(&store.pools[pool_idx].disk_set[set_idx], DATA_USAGE_CACHE_NAME).await {
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;
};
tracker.set_queue_buckets(&buckets).await;
tracker.save().await?;
let tracker = Arc::new(RwLock::new(tracker));
let qb = tracker.read().await.queue_buckets.clone();
store.pools[pool_idx].disk_set[set_idx]
.clone()
.heal_erasure_set(&qb, tracker.clone())
.await?;
let mut tracker_w = tracker.write().await;
if tracker_w.items_failed > 0 && tracker_w.retry_attempts < 4 {
tracker_w.retry_attempts += 1;
tracker_w.reset_healing().await;
if let Err(err) = tracker_w.update().await {
info!("update tracker failed: {}", err.to_string());
}
return Err(Error::other(ERR_RETRY_HEALING));
}
if tracker_w.items_failed > 0 {
info!(
"Healing of drive '{}' is incomplete, retried {} times (healed: {}, skipped: {}, failed: {}).",
disk.to_string(),
tracker_w.retry_attempts,
tracker_w.items_healed,
tracker_w.item_skipped,
tracker_w.items_failed
);
} else if tracker_w.retry_attempts > 0 {
info!(
"Healing of drive '{}' is incomplete, retried {} times (healed: {}, skipped: {}).",
disk.to_string(),
tracker_w.retry_attempts,
tracker_w.items_healed,
tracker_w.item_skipped
);
} else {
info!(
"Healing of drive '{}' is finished (healed: {}, skipped: {}).",
disk.to_string(),
tracker_w.items_healed,
tracker_w.item_skipped
);
}
if tracker_w.heal_id.is_empty() {
if let Err(err) = tracker_w.delete().await {
error!("delete tracker failed: {}", err.to_string());
}
}
let Some(store) = new_object_layer_fn() else {
return Err(Error::other("errServerNotInitialized"));
};
let disks = store.get_disks(pool_idx, set_idx).await?;
for disk in disks.into_iter() {
if disk.is_none() {
continue;
}
let mut tracker = match load_healing_tracker(&disk).await {
Ok(tracker) => tracker,
Err(err) => {
match err {
DiskError::FileNotFound => {}
_ => {
info!("Unable to load healing tracker on '{:?}': {}, re-initializing..", disk, err.to_string());
}
}
continue;
}
};
if tracker.heal_id == tracker_w.heal_id {
tracker.finished = true;
tracker.update().await?;
}
}
Ok(())
}
#[derive(Debug)]
pub struct HealTask {
pub bucket: String,
pub object: String,
pub version_id: String,
pub opts: HealOpts,
pub resp_tx: Option<Sender<HealResult>>,
pub resp_rx: Option<Receiver<HealResult>>,
}
impl HealTask {
pub fn new(bucket: &str, object: &str, version_id: &str, opts: &HealOpts) -> Self {
Self {
bucket: bucket.to_string(),
object: object.to_string(),
version_id: version_id.to_string(),
opts: *opts,
resp_tx: None,
resp_rx: None,
}
}
}
#[derive(Debug)]
pub struct HealResult {
pub result: HealResultItem,
pub err: Option<Error>,
}
pub struct HealRoutine {
pub tasks_tx: Sender<HealTask>,
tasks_rx: RwLock<Receiver<HealTask>>,
workers: usize,
}
impl HealRoutine {
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>() {
workers = num_healers;
}
}
if workers == 0 {
workers = 4;
}
let (tx, rx) = mpsc::channel(100);
Arc::new(Self {
tasks_tx: tx,
tasks_rx: RwLock::new(rx),
workers,
})
}
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.write().await.recv().await {
Some(task) => {
info!("got task: {:?}", task);
if task.bucket == NOP_HEAL {
d_err = Some(Error::other("skip file"));
} else if task.bucket == SLASH_SEPARATOR {
match heal_disk_format(task.opts).await {
Ok((res, err)) => {
d_res = res;
d_err = err;
}
Err(err) => d_err = Some(err),
}
} else {
let store = new_object_layer_fn().expect("errServerNotInitialized");
if task.object.is_empty() {
match store.heal_bucket(&task.bucket, &task.opts).await {
Ok(res) => {
d_res = res;
d_err = None;
}
Err(err) => d_err = Some(err),
}
} else {
match store
.heal_object(&task.bucket, &task.object, &task.version_id, &task.opts)
.await
{
Ok((res, err)) => {
d_res = res;
d_err = err;
}
Err(err) => d_err = Some(err),
}
}
}
info!("task finished, task: {:?}", task);
if let Some(resp_tx) = task.resp_tx {
let _ = resp_tx
.send(HealResult {
result: d_res,
err: d_err,
})
.await;
} else {
// when respCh is not set caller is not waiting but we
// update the relevant metrics for them
if d_err.is_none() {
bgseq.count_healed(d_res.heal_item_type).await;
} else {
bgseq.count_failed(d_res.heal_item_type).await;
}
}
}
None => {
info!("add_worker, tasks_rx was closed, return");
return;
}
}
}
}
}
// pub fn active_listeners() -> Result<usize> {
// }
async fn heal_disk_format(opts: HealOpts) -> Result<(HealResultItem, Option<Error>)> {
let Some(store) = new_object_layer_fn() else {
return Err(Error::other("errServerNotInitialized"));
};
let (res, err) = store.heal_format(opts.dry_run).await?;
// return any error, ignore error returned when disks have
// already healed.
if err.is_some() {
return Ok((HealResultItem::default(), err));
}
Ok((res, err))
}
pub(crate) async fn heal_bucket(bucket: &str) -> Result<()> {
let (bg_seq, ok) = GLOBAL_BackgroundHealState.get_heal_sequence_by_token(BG_HEALING_UUID).await;
if ok {
// bg_seq must be Some when ok is true
return bg_seq
.unwrap()
.queue_heal_task(
HealSource {
bucket: bucket.to_string(),
..Default::default()
},
HEAL_ITEM_BUCKET.to_string(),
)
.await;
}
Ok(())
}
pub(crate) async fn heal_object(bucket: &str, object: &str, version_id: &str, scan_mode: HealScanMode) -> Result<()> {
let (bg_seq, ok) = GLOBAL_BackgroundHealState.get_heal_sequence_by_token(BG_HEALING_UUID).await;
if ok {
// bg_seq must be Some when ok is true
return HealSequence::heal_object(bg_seq.unwrap(), bucket, object, version_id, scan_mode).await;
}
Ok(())
}
crates/ecstore/src/heal/data_scanner.rs
-1826
View File
File diff suppressed because it is too large Load Diff
crates/ecstore/src/heal/data_scanner_metric.rs
-486
View File
@@ -1,486 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use super::data_scanner::CurrentScannerCycle;
use crate::bucket::lifecycle::lifecycle;
use chrono::Utc;
use lazy_static::lazy_static;
use rustfs_common::last_minute::{AccElem, LastMinuteLatency};
use rustfs_madmin::metrics::ScannerMetrics as M_ScannerMetrics;
use std::{
collections::HashMap,
pin::Pin,
sync::{
Arc,
atomic::{AtomicU64, Ordering},
},
time::{Duration, SystemTime},
};
use tokio::sync::{Mutex, RwLock};
lazy_static! {
pub static ref globalScannerMetrics: Arc<ScannerMetrics> = Arc::new(ScannerMetrics::new());
}
#[derive(Clone, Debug, PartialEq, PartialOrd)]
pub enum ScannerMetric {
// START Realtime metrics, that only records
// last minute latencies and total operation count.
ReadMetadata = 0,
CheckMissing,
SaveUsage,
ApplyAll,
ApplyVersion,
TierObjSweep,
HealCheck,
Ilm,
CheckReplication,
Yield,
CleanAbandoned,
ApplyNonCurrent,
HealAbandonedVersion,
// START Trace metrics:
StartTrace,
ScanObject, // Scan object. All operations included.
HealAbandonedObject,
// END realtime metrics:
LastRealtime,
// Trace only metrics:
ScanFolder, // Scan a folder on disk, recursively.
ScanCycle, // Full cycle, cluster global.
ScanBucketDrive, // Single bucket on one drive.
CompactFolder, // Folder compacted.
// Must be last:
Last,
}
impl ScannerMetric {
/// Convert to string representation for metrics
pub fn as_str(self) -> &'static str {
match self {
Self::ReadMetadata => "read_metadata",
Self::CheckMissing => "check_missing",
Self::SaveUsage => "save_usage",
Self::ApplyAll => "apply_all",
Self::ApplyVersion => "apply_version",
Self::TierObjSweep => "tier_obj_sweep",
Self::HealCheck => "heal_check",
Self::Ilm => "ilm",
Self::CheckReplication => "check_replication",
Self::Yield => "yield",
Self::CleanAbandoned => "clean_abandoned",
Self::ApplyNonCurrent => "apply_non_current",
Self::HealAbandonedVersion => "heal_abandoned_version",
Self::StartTrace => "start_trace",
Self::ScanObject => "scan_object",
Self::HealAbandonedObject => "heal_abandoned_object",
Self::LastRealtime => "last_realtime",
Self::ScanFolder => "scan_folder",
Self::ScanCycle => "scan_cycle",
Self::ScanBucketDrive => "scan_bucket_drive",
Self::CompactFolder => "compact_folder",
Self::Last => "last",
}
}
/// Convert from index back to enum (safe version)
pub fn from_index(index: usize) -> Option<Self> {
if index >= Self::Last as usize {
return None;
}
// Safe conversion using match instead of unsafe transmute
match index {
0 => Some(Self::ReadMetadata),
1 => Some(Self::CheckMissing),
2 => Some(Self::SaveUsage),
3 => Some(Self::ApplyAll),
4 => Some(Self::ApplyVersion),
5 => Some(Self::TierObjSweep),
6 => Some(Self::HealCheck),
7 => Some(Self::Ilm),
8 => Some(Self::CheckReplication),
9 => Some(Self::Yield),
10 => Some(Self::CleanAbandoned),
11 => Some(Self::ApplyNonCurrent),
12 => Some(Self::HealAbandonedVersion),
13 => Some(Self::StartTrace),
14 => Some(Self::ScanObject),
15 => Some(Self::HealAbandonedObject),
16 => Some(Self::LastRealtime),
17 => Some(Self::ScanFolder),
18 => Some(Self::ScanCycle),
19 => Some(Self::ScanBucketDrive),
20 => Some(Self::CompactFolder),
21 => Some(Self::Last),
_ => None,
}
}
}
/// Thread-safe wrapper for LastMinuteLatency with atomic operations
#[derive(Default)]
pub struct LockedLastMinuteLatency {
latency: Arc<Mutex<LastMinuteLatency>>,
}
impl Clone for LockedLastMinuteLatency {
fn clone(&self) -> Self {
Self {
latency: Arc::clone(&self.latency),
}
}
}
impl LockedLastMinuteLatency {
pub fn new() -> Self {
Self {
latency: Arc::new(Mutex::new(LastMinuteLatency::default())),
}
}
/// Add a duration measurement
pub async fn add(&self, duration: Duration) {
self.add_size(duration, 0).await;
}
/// Add a duration measurement with size
pub async fn add_size(&self, duration: Duration, size: u64) {
let mut latency = self.latency.lock().await;
let now = SystemTime::now()
.duration_since(SystemTime::UNIX_EPOCH)
.unwrap_or_default()
.as_secs();
let elem = AccElem {
n: 1,
total: duration.as_secs(),
size,
};
latency.add_all(now, &elem);
}
/// Get total accumulated metrics for the last minute
pub async fn total(&self) -> AccElem {
let mut latency = self.latency.lock().await;
latency.get_total()
}
}
/// Current path tracker for monitoring active scan paths
struct CurrentPathTracker {
current_path: Arc<RwLock<String>>,
}
impl CurrentPathTracker {
fn new(initial_path: String) -> Self {
Self {
current_path: Arc::new(RwLock::new(initial_path)),
}
}
async fn update_path(&self, path: String) {
*self.current_path.write().await = path;
}
async fn get_path(&self) -> String {
self.current_path.read().await.clone()
}
}
/// Main scanner metrics structure
pub struct ScannerMetrics {
// All fields must be accessed atomically and aligned.
operations: Vec<AtomicU64>,
latency: Vec<LockedLastMinuteLatency>,
actions: Vec<AtomicU64>,
actions_latency: Vec<LockedLastMinuteLatency>,
// Current paths contains disk -> tracker mappings
current_paths: Arc<RwLock<HashMap<String, Arc<CurrentPathTracker>>>>,
// Cycle information
cycle_info: Arc<RwLock<Option<CurrentScannerCycle>>>,
}
impl ScannerMetrics {
pub fn new() -> Self {
let operations = (0..ScannerMetric::Last as usize).map(|_| AtomicU64::new(0)).collect();
let latency = (0..ScannerMetric::LastRealtime as usize)
.map(|_| LockedLastMinuteLatency::new())
.collect();
Self {
operations,
latency,
actions: (0..ScannerMetric::Last as usize).map(|_| AtomicU64::new(0)).collect(),
actions_latency: vec![LockedLastMinuteLatency::default(); ScannerMetric::LastRealtime as usize],
current_paths: Arc::new(RwLock::new(HashMap::new())),
cycle_info: Arc::new(RwLock::new(None)),
}
}
/// Log scanner action with custom metadata - compatible with existing usage
pub fn log(metric: ScannerMetric) -> impl Fn(&HashMap<String, String>) {
let metric = metric as usize;
let start_time = SystemTime::now();
move |_custom: &HashMap<String, String>| {
let duration = SystemTime::now().duration_since(start_time).unwrap_or_default();
// Update operation count
globalScannerMetrics.operations[metric].fetch_add(1, Ordering::Relaxed);
// Update latency for realtime metrics (spawn async task for this)
if (metric) < ScannerMetric::LastRealtime as usize {
let metric_index = metric;
tokio::spawn(async move {
globalScannerMetrics.latency[metric_index].add(duration).await;
});
}
// Log trace metrics
if metric as u8 > ScannerMetric::StartTrace as u8 {
//debug!(metric = metric.as_str(), duration_ms = duration.as_millis(), "Scanner trace metric");
}
}
}
/// Time scanner action with size - returns function that takes size
pub fn time_size(metric: ScannerMetric) -> impl Fn(u64) {
let metric = metric as usize;
let start_time = SystemTime::now();
move |size: u64| {
let duration = SystemTime::now().duration_since(start_time).unwrap_or_default();
// Update operation count
globalScannerMetrics.operations[metric].fetch_add(1, Ordering::Relaxed);
// Update latency for realtime metrics with size (spawn async task)
if (metric) < ScannerMetric::LastRealtime as usize {
let metric_index = metric;
tokio::spawn(async move {
globalScannerMetrics.latency[metric_index].add_size(duration, size).await;
});
}
}
}
/// Time a scanner action - returns a closure to call when done
pub fn time(metric: ScannerMetric) -> impl Fn() {
let metric = metric as usize;
let start_time = SystemTime::now();
move || {
let duration = SystemTime::now().duration_since(start_time).unwrap_or_default();
// Update operation count
globalScannerMetrics.operations[metric].fetch_add(1, Ordering::Relaxed);
// Update latency for realtime metrics (spawn async task)
if (metric) < ScannerMetric::LastRealtime as usize {
let metric_index = metric;
tokio::spawn(async move {
globalScannerMetrics.latency[metric_index].add(duration).await;
});
}
}
}
/// Time N scanner actions - returns function that takes count, then returns completion function
pub fn time_n(metric: ScannerMetric) -> Box<dyn Fn(usize) -> Box<dyn Fn() + Send + Sync> + Send + Sync> {
let metric = metric as usize;
let start_time = SystemTime::now();
Box::new(move |count: usize| {
Box::new(move || {
let duration = SystemTime::now().duration_since(start_time).unwrap_or_default();
// Update operation count
globalScannerMetrics.operations[metric].fetch_add(count as u64, Ordering::Relaxed);
// Update latency for realtime metrics (spawn async task)
if (metric) < ScannerMetric::LastRealtime as usize {
let metric_index = metric;
tokio::spawn(async move {
globalScannerMetrics.latency[metric_index].add(duration).await;
});
}
})
})
}
pub fn time_ilm(a: lifecycle::IlmAction) -> Box<dyn Fn(u64) -> Box<dyn Fn() + Send + Sync> + Send + Sync> {
let a_clone = a as usize;
if a_clone == lifecycle::IlmAction::NoneAction as usize || a_clone >= lifecycle::IlmAction::ActionCount as usize {
return Box::new(move |_: u64| Box::new(move || {}));
}
let start = SystemTime::now();
Box::new(move |versions: u64| {
Box::new(move || {
let duration = SystemTime::now().duration_since(start).unwrap_or(Duration::from_secs(0));
tokio::spawn(async move {
globalScannerMetrics.actions[a_clone].fetch_add(versions, Ordering::Relaxed);
globalScannerMetrics.actions_latency[a_clone].add(duration).await;
});
})
})
}
/// Increment time with specific duration
pub async fn inc_time(metric: ScannerMetric, duration: Duration) {
let metric = metric as usize;
// Update operation count
globalScannerMetrics.operations[metric].fetch_add(1, Ordering::Relaxed);
// Update latency for realtime metrics
if (metric) < ScannerMetric::LastRealtime as usize {
globalScannerMetrics.latency[metric].add(duration).await;
}
}
/// Get lifetime operation count for a metric
pub fn lifetime(&self, metric: ScannerMetric) -> u64 {
let metric = metric as usize;
if (metric) >= ScannerMetric::Last as usize {
return 0;
}
self.operations[metric].load(Ordering::Relaxed)
}
/// Get last minute statistics for a metric
pub async fn last_minute(&self, metric: ScannerMetric) -> AccElem {
let metric = metric as usize;
if (metric) >= ScannerMetric::LastRealtime as usize {
return AccElem::default();
}
self.latency[metric].total().await
}
/// Set current cycle information
pub async fn set_cycle(&self, cycle: Option<CurrentScannerCycle>) {
*self.cycle_info.write().await = cycle;
}
/// Get current cycle information
pub async fn get_cycle(&self) -> Option<CurrentScannerCycle> {
self.cycle_info.read().await.clone()
}
/// Get current active paths
pub async fn get_current_paths(&self) -> Vec<String> {
let mut result = Vec::new();
let paths = self.current_paths.read().await;
for (disk, tracker) in paths.iter() {
let path = tracker.get_path().await;
result.push(format!("{disk}/{path}"));
}
result
}
/// Get number of active drives
pub async fn active_drives(&self) -> usize {
self.current_paths.read().await.len()
}
/// Generate metrics report
pub async fn report(&self) -> M_ScannerMetrics {
let mut metrics = M_ScannerMetrics::default();
// Set cycle information
if let Some(cycle) = self.get_cycle().await {
metrics.current_cycle = cycle.current;
metrics.cycles_completed_at = cycle.cycle_completed;
metrics.current_started = cycle.started;
}
metrics.collected_at = Utc::now();
metrics.active_paths = self.get_current_paths().await;
// Lifetime operations
for i in 0..ScannerMetric::Last as usize {
let count = self.operations[i].load(Ordering::Relaxed);
if count > 0 {
if let Some(metric) = ScannerMetric::from_index(i) {
metrics.life_time_ops.insert(metric.as_str().to_string(), count);
}
}
}
// Last minute statistics for realtime metrics
for i in 0..ScannerMetric::LastRealtime as usize {
let last_min = self.latency[i].total().await;
if last_min.n > 0 {
if let Some(_metric) = ScannerMetric::from_index(i) {
// Convert to madmin TimedAction format if needed
// This would require implementing the conversion
}
}
}
metrics
}
}
// Type aliases for compatibility with existing code
pub type UpdateCurrentPathFn = Arc<dyn Fn(&str) -> Pin<Box<dyn std::future::Future<Output = ()> + Send>> + Send + Sync>;
pub type CloseDiskFn = Arc<dyn Fn() -> Pin<Box<dyn std::future::Future<Output = ()> + Send>> + Send + Sync>;
/// Create a current path updater for tracking scan progress
pub fn current_path_updater(disk: &str, initial: &str) -> (UpdateCurrentPathFn, CloseDiskFn) {
let tracker = Arc::new(CurrentPathTracker::new(initial.to_string()));
let disk_name = disk.to_string();
// Store the tracker in global metrics
let tracker_clone = Arc::clone(&tracker);
let disk_clone = disk_name.clone();
tokio::spawn(async move {
globalScannerMetrics
.current_paths
.write()
.await
.insert(disk_clone, tracker_clone);
});
let update_fn = {
let tracker = Arc::clone(&tracker);
Arc::new(move |path: &str| -> Pin<Box<dyn std::future::Future<Output = ()> + Send>> {
let tracker = Arc::clone(&tracker);
let path = path.to_string();
Box::pin(async move {
tracker.update_path(path).await;
})
})
};
let done_fn = {
let disk_name = disk_name.clone();
Arc::new(move || -> Pin<Box<dyn std::future::Future<Output = ()> + Send>> {
let disk_name = disk_name.clone();
Box::pin(async move {
globalScannerMetrics.current_paths.write().await.remove(&disk_name);
})
})
};
(update_fn, done_fn)
}
impl Default for ScannerMetrics {
fn default() -> Self {
Self::new()
}
}
crates/ecstore/src/heal/data_usage.rs
-221
View File
@@ -1,221 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::error::{Error, Result};
use crate::{
bucket::metadata_sys::get_replication_config,
config::com::{read_config, save_config},
disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET},
error::to_object_err,
new_object_layer_fn,
store::ECStore,
};
use lazy_static::lazy_static;
use rustfs_utils::path::SLASH_SEPARATOR;
use serde::{Deserialize, Serialize};
use std::{collections::HashMap, sync::Arc, time::SystemTime};
use tokio::sync::mpsc::Receiver;
use tracing::{error, warn};
pub const DATA_USAGE_ROOT: &str = SLASH_SEPARATOR;
const DATA_USAGE_OBJ_NAME: &str = ".usage.json";
const DATA_USAGE_BLOOM_NAME: &str = ".bloomcycle.bin";
pub const DATA_USAGE_CACHE_NAME: &str = ".usage-cache.bin";
lazy_static! {
pub static ref DATA_USAGE_BUCKET: String = format!("{}{}{}", RUSTFS_META_BUCKET, SLASH_SEPARATOR, BUCKET_META_PREFIX);
pub static ref DATA_USAGE_OBJ_NAME_PATH: String = format!("{}{}{}", BUCKET_META_PREFIX, SLASH_SEPARATOR, DATA_USAGE_OBJ_NAME);
pub static ref DATA_USAGE_BLOOM_NAME_PATH: String =
format!("{}{}{}", BUCKET_META_PREFIX, SLASH_SEPARATOR, DATA_USAGE_BLOOM_NAME);
pub static ref BACKGROUND_HEAL_INFO_PATH: String =
format!("{}{}{}", BUCKET_META_PREFIX, SLASH_SEPARATOR, ".background-heal.json");
}
// BucketTargetUsageInfo - bucket target usage info provides
// - replicated size for all objects sent to this target
// - replica size for all objects received from this target
// - replication pending size for all objects pending replication to this target
// - replication failed size for all objects failed replication to this target
// - replica pending count
// - replica failed count
#[derive(Debug, Default, Serialize, Deserialize)]
pub struct BucketTargetUsageInfo {
pub replication_pending_size: u64,
pub replication_failed_size: u64,
pub replicated_size: u64,
pub replica_size: u64,
pub replication_pending_count: u64,
pub replication_failed_count: u64,
pub replicated_count: u64,
}
// BucketUsageInfo - bucket usage info provides
// - total size of the bucket
// - total objects in a bucket
// - object size histogram per bucket
#[derive(Debug, Default, Serialize, Deserialize)]
pub struct BucketUsageInfo {
pub size: u64,
// Following five fields suffixed with V1 are here for backward compatibility
// Total Size for objects that have not yet been replicated
pub replication_pending_size_v1: u64,
// Total size for objects that have witness one or more failures and will be retried
pub replication_failed_size_v1: u64,
// Total size for objects that have been replicated to destination
pub replicated_size_v1: u64,
// Total number of objects pending replication
pub replication_pending_count_v1: u64,
// Total number of objects that failed replication
pub replication_failed_count_v1: u64,
pub objects_count: u64,
pub object_size_histogram: HashMap<String, u64>,
pub object_versions_histogram: HashMap<String, u64>,
pub versions_count: u64,
pub delete_markers_count: u64,
pub replica_size: u64,
pub replica_count: u64,
pub replication_info: HashMap<String, BucketTargetUsageInfo>,
}
// DataUsageInfo represents data usage stats of the underlying Object API
#[derive(Debug, Default, Serialize, Deserialize)]
pub struct DataUsageInfo {
pub total_capacity: u64,
pub total_used_capacity: u64,
pub total_free_capacity: u64,
// LastUpdate is the timestamp of when the data usage info was last updated.
// This does not indicate a full scan.
pub last_update: Option<SystemTime>,
// Objects total count across all buckets
pub objects_total_count: u64,
// Versions total count across all buckets
pub versions_total_count: u64,
// Delete markers total count across all buckets
pub delete_markers_total_count: u64,
// Objects total size across all buckets
pub objects_total_size: u64,
pub replication_info: HashMap<String, BucketTargetUsageInfo>,
// Total number of buckets in this cluster
pub buckets_count: u64,
// Buckets usage info provides following information across all buckets
// - total size of the bucket
// - total objects in a bucket
// - object size histogram per bucket
pub buckets_usage: HashMap<String, BucketUsageInfo>,
// Deprecated kept here for backward compatibility reasons.
pub bucket_sizes: HashMap<String, u64>,
// Todo: TierStats
// TierStats contains per-tier stats of all configured remote tiers
}
pub async fn store_data_usage_in_backend(mut rx: Receiver<DataUsageInfo>) {
let Some(store) = new_object_layer_fn() else {
error!("errServerNotInitialized");
return;
};
let mut attempts = 1;
loop {
match rx.recv().await {
Some(data_usage_info) => {
if let Ok(data) = serde_json::to_vec(&data_usage_info) {
if attempts > 10 {
let _ =
save_config(store.clone(), &format!("{}{}", *DATA_USAGE_OBJ_NAME_PATH, ".bkp"), data.clone()).await;
attempts += 1;
}
let _ = save_config(store.clone(), &DATA_USAGE_OBJ_NAME_PATH, data).await;
attempts += 1;
} else {
continue;
}
}
None => {
return;
}
}
}
}
// TODO: cancel ctx
pub async fn load_data_usage_from_backend(store: Arc<ECStore>) -> Result<DataUsageInfo> {
let buf = match read_config(store, &DATA_USAGE_OBJ_NAME_PATH).await {
Ok(data) => data,
Err(e) => {
error!("Failed to read data usage info from backend: {}", e);
if e == Error::ConfigNotFound {
return Ok(DataUsageInfo::default());
}
return Err(to_object_err(e, vec![RUSTFS_META_BUCKET, &DATA_USAGE_OBJ_NAME_PATH]));
}
};
let mut data_usage_info: DataUsageInfo = serde_json::from_slice(&buf)?;
warn!("Loaded data usage info from backend {:?}", &data_usage_info);
if data_usage_info.buckets_usage.is_empty() {
data_usage_info.buckets_usage = data_usage_info
.bucket_sizes
.iter()
.map(|(bucket, &size)| {
(
bucket.clone(),
BucketUsageInfo {
size,
..Default::default()
},
)
})
.collect();
}
if data_usage_info.bucket_sizes.is_empty() {
data_usage_info.bucket_sizes = data_usage_info
.buckets_usage
.iter()
.map(|(bucket, bui)| (bucket.clone(), bui.size))
.collect();
}
for (bucket, bui) in &data_usage_info.buckets_usage {
if bui.replicated_size_v1 > 0
|| bui.replication_failed_count_v1 > 0
|| bui.replication_failed_size_v1 > 0
|| bui.replication_pending_count_v1 > 0
{
if let Ok((cfg, _)) = get_replication_config(bucket).await {
if !cfg.role.is_empty() {
data_usage_info.replication_info.insert(
cfg.role.clone(),
BucketTargetUsageInfo {
replication_failed_size: bui.replication_failed_size_v1,
replication_failed_count: bui.replication_failed_count_v1,
replicated_size: bui.replicated_size_v1,
replication_pending_count: bui.replication_pending_count_v1,
replication_pending_size: bui.replication_pending_size_v1,
..Default::default()
},
);
}
}
}
}
Ok(data_usage_info)
}
crates/ecstore/src/heal/data_usage_cache.rs
-928
View File
@@ -1,928 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::config::com::save_config;
use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET};
use crate::error::{Error, Result};
use crate::new_object_layer_fn;
use crate::set_disk::SetDisks;
use crate::store_api::{BucketInfo, ObjectIO, ObjectOptions};
use bytesize::ByteSize;
use http::HeaderMap;
use path_clean::PathClean;
use rand::Rng;
use rmp_serde::Serializer;
use s3s::dto::{BucketLifecycleConfiguration, ReplicationConfiguration};
use serde::{Deserialize, Serialize};
use std::collections::{HashMap, HashSet};
use std::hash::{DefaultHasher, Hash, Hasher};
use std::path::Path;
use std::time::{Duration, SystemTime};
use tokio::sync::mpsc::Sender;
use tokio::time::sleep;
use super::data_scanner::{DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS, SizeSummary};
use super::data_usage::{BucketTargetUsageInfo, BucketUsageInfo, DataUsageInfo};
// DATA_USAGE_BUCKET_LEN must be length of ObjectsHistogramIntervals
pub const DATA_USAGE_BUCKET_LEN: usize = 11;
pub const DATA_USAGE_VERSION_LEN: usize = 7;
pub type DataUsageHashMap = HashSet<String>;
struct ObjectHistogramInterval {
name: &'static str,
start: u64,
end: u64,
}
const OBJECTS_HISTOGRAM_INTERVALS: [ObjectHistogramInterval; DATA_USAGE_BUCKET_LEN] = [
ObjectHistogramInterval {
name: "LESS_THAN_1024_B",
start: 0,
end: ByteSize::kib(1).as_u64() - 1,
},
ObjectHistogramInterval {
name: "BETWEEN_1024_B_AND_64_KB",
start: ByteSize::kib(1).as_u64(),
end: ByteSize::kib(64).as_u64() - 1,
},
ObjectHistogramInterval {
name: "BETWEEN_64_KB_AND_256_KB",
start: ByteSize::kib(64).as_u64(),
end: ByteSize::kib(256).as_u64() - 1,
},
ObjectHistogramInterval {
name: "BETWEEN_256_KB_AND_512_KB",
start: ByteSize::kib(256).as_u64(),
end: ByteSize::kib(512).as_u64() - 1,
},
ObjectHistogramInterval {
name: "BETWEEN_512_KB_AND_1_MB",
start: ByteSize::kib(512).as_u64(),
end: ByteSize::mib(1).as_u64() - 1,
},
ObjectHistogramInterval {
name: "BETWEEN_1024B_AND_1_MB",
start: ByteSize::kib(1).as_u64(),
end: ByteSize::mib(1).as_u64() - 1,
},
ObjectHistogramInterval {
name: "BETWEEN_1_MB_AND_10_MB",
start: ByteSize::mib(1).as_u64(),
end: ByteSize::mib(10).as_u64() - 1,
},
ObjectHistogramInterval {
name: "BETWEEN_10_MB_AND_64_MB",
start: ByteSize::mib(10).as_u64(),
end: ByteSize::mib(64).as_u64() - 1,
},
ObjectHistogramInterval {
name: "BETWEEN_64_MB_AND_128_MB",
start: ByteSize::mib(64).as_u64(),
end: ByteSize::mib(128).as_u64() - 1,
},
ObjectHistogramInterval {
name: "BETWEEN_128_MB_AND_512_MB",
start: ByteSize::mib(128).as_u64(),
end: ByteSize::mib(512).as_u64() - 1,
},
ObjectHistogramInterval {
name: "GREATER_THAN_512_MB",
start: ByteSize::mib(512).as_u64(),
end: u64::MAX,
},
];
const OBJECTS_VERSION_COUNT_INTERVALS: [ObjectHistogramInterval; DATA_USAGE_VERSION_LEN] = [
ObjectHistogramInterval {
name: "UNVERSIONED",
start: 0,
end: 0,
},
ObjectHistogramInterval {
name: "SINGLE_VERSION",
start: 1,
end: 1,
},
ObjectHistogramInterval {
name: "BETWEEN_2_AND_10",
start: 2,
end: 9,
},
ObjectHistogramInterval {
name: "BETWEEN_10_AND_100",
start: 10,
end: 99,
},
ObjectHistogramInterval {
name: "BETWEEN_100_AND_1000",
start: 100,
end: 999,
},
ObjectHistogramInterval {
name: "BETWEEN_1000_AND_10000",
start: 1000,
end: 9999,
},
ObjectHistogramInterval {
name: "GREATER_THAN_10000",
start: 10000,
end: u64::MAX,
},
];
#[derive(Clone, Copy, Default)]
pub struct TierStats {
pub total_size: u64,
pub num_versions: i32,
pub num_objects: i32,
}
impl TierStats {
pub fn add(&self, u: &TierStats) -> TierStats {
TierStats {
total_size: self.total_size + u.total_size,
num_versions: self.num_versions + u.num_versions,
num_objects: self.num_objects + u.num_objects,
}
}
}
struct AllTierStats {
tiers: HashMap<String, TierStats>,
}
impl AllTierStats {
pub fn new() -> Self {
Self { tiers: HashMap::new() }
}
fn add_sizes(&mut self, tiers: HashMap<String, TierStats>) {
for (tier, st) in tiers {
self.tiers.insert(tier.clone(), self.tiers[&tier].add(&st));
}
}
fn merge(&mut self, other: AllTierStats) {
for (tier, st) in other.tiers {
self.tiers.insert(tier.clone(), self.tiers[&tier].add(&st));
}
}
fn populate_stats(&self, stats: &mut HashMap<String, TierStats>) {
for (tier, st) in &self.tiers {
stats.insert(
tier.clone(),
TierStats {
total_size: st.total_size,
num_versions: st.num_versions,
num_objects: st.num_objects,
},
);
}
}
}
// sizeHistogram is a size histogram.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct SizeHistogram(Vec<u64>);
impl Default for SizeHistogram {
fn default() -> Self {
Self(vec![0; DATA_USAGE_BUCKET_LEN])
}
}
impl SizeHistogram {
fn add(&mut self, size: u64) {
for (idx, interval) in OBJECTS_HISTOGRAM_INTERVALS.iter().enumerate() {
if size >= interval.start && size <= interval.end {
self.0[idx] += 1;
break;
}
}
}
pub fn to_map(&self) -> HashMap<String, u64> {
let mut res = HashMap::new();
let mut spl_count = 0;
for (count, oh) in self.0.iter().zip(OBJECTS_HISTOGRAM_INTERVALS.iter()) {
if ByteSize::kib(1).as_u64() == oh.start && oh.end == ByteSize::mib(1).as_u64() - 1 {
res.insert(oh.name.to_string(), spl_count);
} else if ByteSize::kib(1).as_u64() <= oh.start && oh.end < ByteSize::mib(1).as_u64() {
spl_count += count;
res.insert(oh.name.to_string(), *count);
} else {
res.insert(oh.name.to_string(), *count);
}
}
res
}
}
// versionsHistogram is a histogram of number of versions in an object.
#[derive(Clone, Debug, Serialize, Deserialize)]
pub struct VersionsHistogram(Vec<u64>);
impl Default for VersionsHistogram {
fn default() -> Self {
Self(vec![0; DATA_USAGE_VERSION_LEN])
}
}
impl VersionsHistogram {
fn add(&mut self, size: u64) {
for (idx, interval) in OBJECTS_VERSION_COUNT_INTERVALS.iter().enumerate() {
if size >= interval.start && size <= interval.end {
self.0[idx] += 1;
break;
}
}
}
pub fn to_map(&self) -> HashMap<String, u64> {
let mut res = HashMap::new();
for (count, ov) in self.0.iter().zip(OBJECTS_VERSION_COUNT_INTERVALS.iter()) {
res.insert(ov.name.to_string(), *count);
}
res
}
}
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct ReplicationStats {
pub pending_size: u64,
pub replicated_size: u64,
pub failed_size: u64,
pub failed_count: u64,
pub pending_count: u64,
pub missed_threshold_size: u64,
pub after_threshold_size: u64,
pub missed_threshold_count: u64,
pub after_threshold_count: u64,
pub replicated_count: u64,
}
impl ReplicationStats {
pub fn empty(&self) -> bool {
self.replicated_size == 0 && self.failed_size == 0 && self.failed_count == 0
}
}
#[derive(Debug, Default, Clone, Serialize, Deserialize)]
pub struct ReplicationAllStats {
pub targets: HashMap<String, ReplicationStats>,
pub replica_size: u64,
pub replica_count: u64,
}
impl ReplicationAllStats {
pub fn empty(&self) -> bool {
if self.replica_size != 0 && self.replica_count != 0 {
return false;
}
for (_, v) in self.targets.iter() {
if !v.empty() {
return false;
}
}
true
}
}
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
pub struct DataUsageEntry {
pub children: DataUsageHashMap,
// These fields do not include any children.
pub size: usize,
pub objects: usize,
pub versions: usize,
pub delete_markers: usize,
pub obj_sizes: SizeHistogram,
pub obj_versions: VersionsHistogram,
pub replication_stats: Option<ReplicationAllStats>,
// Todo: tier
// pub all_tier_stats: ,
pub compacted: bool,
}
impl DataUsageEntry {
pub fn add_child(&mut self, hash: &DataUsageHash) {
if self.children.contains(&hash.key()) {
return;
}
self.children.insert(hash.key());
}
pub fn add_sizes(&mut self, summary: &SizeSummary) {
self.size += summary.total_size;
self.versions += summary.versions;
self.delete_markers += summary.delete_markers;
self.obj_sizes.add(summary.total_size as u64);
self.obj_versions.add(summary.versions as u64);
let replication_stats = if self.replication_stats.is_none() {
self.replication_stats = Some(ReplicationAllStats::default());
self.replication_stats.as_mut().unwrap()
} else {
self.replication_stats.as_mut().unwrap()
};
replication_stats.replica_size += summary.replica_size as u64;
replication_stats.replica_count += summary.replica_count as u64;
for (arn, st) in &summary.repl_target_stats {
let tgt_stat = replication_stats
.targets
.entry(arn.to_string())
.or_insert(ReplicationStats::default());
tgt_stat.pending_size += st.pending_size as u64;
tgt_stat.failed_size += st.failed_size as u64;
tgt_stat.replicated_size += st.replicated_size as u64;
tgt_stat.replicated_count += st.replicated_count as u64;
tgt_stat.failed_count += st.failed_count as u64;
tgt_stat.pending_count += st.pending_count as u64;
}
// Todo:: tiers
}
pub fn merge(&mut self, other: &DataUsageEntry) {
self.objects += other.objects;
self.versions += other.versions;
self.delete_markers += other.delete_markers;
self.size += other.size;
if let Some(o_rep) = &other.replication_stats {
if self.replication_stats.is_none() {
self.replication_stats = Some(ReplicationAllStats::default());
}
let s_rep = self.replication_stats.as_mut().unwrap();
s_rep.targets.clear();
s_rep.replica_size += o_rep.replica_size;
s_rep.replica_count += o_rep.replica_count;
for (arn, stat) in o_rep.targets.iter() {
let st = s_rep.targets.entry(arn.clone()).or_default();
*st = ReplicationStats {
pending_size: stat.pending_size + st.pending_size,
failed_size: stat.failed_size + st.failed_size,
replicated_size: stat.replicated_size + st.replicated_size,
pending_count: stat.pending_count + st.pending_count,
failed_count: stat.failed_count + st.failed_count,
replicated_count: stat.replicated_count + st.replicated_count,
..Default::default()
};
}
}
for (i, v) in other.obj_sizes.0.iter().enumerate() {
self.obj_sizes.0[i] += v;
}
for (i, v) in other.obj_versions.0.iter().enumerate() {
self.obj_versions.0[i] += v;
}
// todo: tiers
}
}
#[derive(Clone)]
pub struct DataUsageEntryInfo {
pub name: String,
pub parent: String,
pub entry: DataUsageEntry,
}
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
pub struct DataUsageCacheInfo {
pub name: String,
pub next_cycle: u32,
pub last_update: Option<SystemTime>,
pub skip_healing: bool,
#[serde(skip)]
pub lifecycle: Option<BucketLifecycleConfiguration>,
#[serde(skip)]
pub updates: Option<Sender<DataUsageEntry>>,
#[serde(skip)]
pub replication: Option<ReplicationConfiguration>,
}
// impl Default for DataUsageCacheInfo {
// fn default() -> Self {
// Self {
// name: Default::default(),
// next_cycle: Default::default(),
// last_update: SystemTime::now(),
// skip_healing: Default::default(),
// updates: Default::default(),
// replication: Default::default(),
// }
// }
// }
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
pub struct DataUsageCache {
pub info: DataUsageCacheInfo,
pub cache: HashMap<String, DataUsageEntry>,
}
impl DataUsageCache {
pub async fn load(store: &SetDisks, name: &str) -> Result<Self> {
let mut d = DataUsageCache::default();
let mut retries = 0;
while retries < 5 {
let path = Path::new(BUCKET_META_PREFIX).join(name);
// warn!("Loading data usage cache from backend: {}", path.display());
match store
.get_object_reader(
RUSTFS_META_BUCKET,
path.to_str().unwrap(),
None,
HeaderMap::new(),
&ObjectOptions {
no_lock: true,
..Default::default()
},
)
.await
{
Ok(mut reader) => {
if let Ok(info) = Self::unmarshal(&reader.read_all().await?) {
d = info
}
break;
}
Err(err) => {
// warn!("Failed to load data usage cache from backend: {}", &err);
match err {
Error::FileNotFound | Error::VolumeNotFound => {
match store
.get_object_reader(
RUSTFS_META_BUCKET,
name,
None,
HeaderMap::new(),
&ObjectOptions {
no_lock: true,
..Default::default()
},
)
.await
{
Ok(mut reader) => {
if let Ok(info) = Self::unmarshal(&reader.read_all().await?) {
d = info
}
break;
}
Err(_) => match err {
Error::FileNotFound | Error::VolumeNotFound => {
break;
}
_ => {}
},
}
}
_ => {
break;
}
}
}
}
retries += 1;
let dur = {
let mut rng = rand::rng();
rng.random_range(0..1_000)
};
sleep(Duration::from_millis(dur)).await;
}
Ok(d)
}
pub async fn save(&self, name: &str) -> Result<()> {
let Some(store) = new_object_layer_fn() else {
return Err(Error::other("errServerNotInitialized"));
};
let buf = self.marshal_msg()?;
let buf_clone = buf.clone();
let store_clone = store.clone();
let name = Path::new(BUCKET_META_PREFIX).join(name).to_string_lossy().to_string();
let name_clone = name.clone();
tokio::spawn(async move {
let _ = save_config(store_clone, &format!("{}{}", &name_clone, ".bkp"), buf_clone).await;
});
save_config(store, &name, buf).await?;
Ok(())
}
pub fn replace(&mut self, path: &str, parent: &str, e: DataUsageEntry) {
let hash = hash_path(path);
self.cache.insert(hash.key(), e);
if !parent.is_empty() {
let phash = hash_path(parent);
let p = {
let p = self.cache.entry(phash.key()).or_default();
p.add_child(&hash);
p.clone()
};
self.cache.insert(phash.key(), p);
}
}
pub fn replace_hashed(&mut self, hash: &DataUsageHash, parent: &Option<DataUsageHash>, e: &DataUsageEntry) {
self.cache.insert(hash.key(), e.clone());
if let Some(parent) = parent {
self.cache.entry(parent.key()).or_default().add_child(hash);
}
}
pub fn find(&self, path: &str) -> Option<DataUsageEntry> {
self.cache.get(&hash_path(path).key()).cloned()
}
pub fn find_children_copy(&mut self, h: DataUsageHash) -> DataUsageHashMap {
self.cache.entry(h.string()).or_default().children.clone()
}
pub fn flatten(&self, root: &DataUsageEntry) -> DataUsageEntry {
let mut root = root.clone();
for id in root.children.clone().iter() {
if let Some(e) = self.cache.get(id) {
let mut e = e.clone();
if !e.children.is_empty() {
e = self.flatten(&e);
}
root.merge(&e);
}
}
root.children.clear();
root
}
pub fn copy_with_children(&mut self, src: &DataUsageCache, hash: &DataUsageHash, parent: &Option<DataUsageHash>) {
if let Some(e) = src.cache.get(&hash.string()) {
self.cache.insert(hash.key(), e.clone());
for ch in e.children.iter() {
if *ch == hash.key() {
return;
}
self.copy_with_children(src, &DataUsageHash(ch.to_string()), &Some(hash.clone()));
}
if let Some(parent) = parent {
let p = self.cache.entry(parent.key()).or_default();
p.add_child(hash);
}
}
}
pub fn delete_recursive(&mut self, hash: &DataUsageHash) {
let mut need_remove = Vec::new();
if let Some(v) = self.cache.get(&hash.string()) {
for child in v.children.iter() {
need_remove.push(child.clone());
}
}
self.cache.remove(&hash.string());
need_remove.iter().for_each(|child| {
self.delete_recursive(&DataUsageHash(child.to_string()));
});
}
pub fn size_recursive(&self, path: &str) -> Option<DataUsageEntry> {
match self.find(path) {
Some(root) => {
if root.children.is_empty() {
return Some(root);
}
let mut flat = self.flatten(&root);
if flat.replication_stats.is_some() && flat.replication_stats.as_ref().unwrap().empty() {
flat.replication_stats = None;
}
Some(flat)
}
None => None,
}
}
pub fn search_parent(&self, hash: &DataUsageHash) -> Option<DataUsageHash> {
let want = hash.key();
if let Some(last_index) = want.rfind('/') {
if let Some(v) = self.find(&want[0..last_index]) {
if v.children.contains(&want) {
let found = hash_path(&want[0..last_index]);
return Some(found);
}
}
}
for (k, v) in self.cache.iter() {
if v.children.contains(&want) {
let found = DataUsageHash(k.clone());
return Some(found);
}
}
None
}
pub fn is_compacted(&self, hash: &DataUsageHash) -> bool {
match self.cache.get(&hash.key()) {
Some(due) => due.compacted,
None => false,
}
}
pub fn force_compact(&mut self, limit: usize) {
if self.cache.len() < limit {
return;
}
let top = hash_path(&self.info.name).key();
let top_e = match self.find(&top) {
Some(e) => e,
None => return,
};
if top_e.children.len() > <u64 as TryInto<usize>>::try_into(DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS).unwrap() {
self.reduce_children_of(&hash_path(&self.info.name), limit, true);
}
if self.cache.len() <= limit {
return;
}
let mut found = HashSet::new();
found.insert(top);
mark(self, &top_e, &mut found);
self.cache.retain(|k, _| {
if !found.contains(k) {
return false;
}
true
});
}
pub fn reduce_children_of(&mut self, path: &DataUsageHash, limit: usize, compact_self: bool) {
let e = match self.cache.get(&path.key()) {
Some(e) => e,
None => return,
};
if e.compacted {
return;
}
if e.children.len() > limit && compact_self {
let mut flat = self.size_recursive(&path.key()).unwrap_or_default();
flat.compacted = true;
self.delete_recursive(path);
self.replace_hashed(path, &None, &flat);
return;
}
let total = self.total_children_rec(&path.key());
if total < limit {
return;
}
let mut leaves = Vec::new();
let mut remove = total - limit;
add(self, path, &mut leaves);
leaves.sort_by(|a, b| a.objects.cmp(&b.objects));
while remove > 0 && !leaves.is_empty() {
let e = leaves.first().unwrap();
let candidate = e.path.clone();
if candidate == *path && !compact_self {
break;
}
let removing = self.total_children_rec(&candidate.key());
let mut flat = match self.size_recursive(&candidate.key()) {
Some(flat) => flat,
None => {
leaves.remove(0);
continue;
}
};
flat.compacted = true;
self.delete_recursive(&candidate);
self.replace_hashed(&candidate, &None, &flat);
remove -= removing;
leaves.remove(0);
}
}
pub fn total_children_rec(&self, path: &str) -> usize {
let root = self.find(path);
if root.is_none() {
return 0;
}
let root = root.unwrap();
if root.children.is_empty() {
return 0;
}
let mut n = root.children.len();
for ch in root.children.iter() {
n += self.total_children_rec(ch);
}
n
}
pub fn merge(&mut self, o: &DataUsageCache) {
let mut existing_root = self.root();
let other_root = o.root();
if existing_root.is_none() && other_root.is_none() {
return;
}
if other_root.is_none() {
return;
}
if existing_root.is_none() {
*self = o.clone();
return;
}
if o.info.last_update.gt(&self.info.last_update) {
self.info.last_update = o.info.last_update;
}
existing_root.as_mut().unwrap().merge(other_root.as_ref().unwrap());
self.cache.insert(hash_path(&self.info.name).key(), existing_root.unwrap());
let e_hash = self.root_hash();
for key in other_root.as_ref().unwrap().children.iter() {
let entry = &o.cache[key];
let flat = o.flatten(entry);
let mut existing = self.cache[key].clone();
existing.merge(&flat);
self.replace_hashed(&DataUsageHash(key.clone()), &Some(e_hash.clone()), &existing);
}
}
pub fn root_hash(&self) -> DataUsageHash {
hash_path(&self.info.name)
}
pub fn root(&self) -> Option<DataUsageEntry> {
self.find(&self.info.name)
}
pub fn dui(&self, path: &str, buckets: &[BucketInfo]) -> DataUsageInfo {
let e = match self.find(path) {
Some(e) => e,
None => return DataUsageInfo::default(),
};
let flat = self.flatten(&e);
DataUsageInfo {
last_update: self.info.last_update,
objects_total_count: flat.objects as u64,
versions_total_count: flat.versions as u64,
delete_markers_total_count: flat.delete_markers as u64,
objects_total_size: flat.size as u64,
buckets_count: e.children.len() as u64,
buckets_usage: self.buckets_usage_info(buckets),
..Default::default()
}
}
pub fn buckets_usage_info(&self, buckets: &[BucketInfo]) -> HashMap<String, BucketUsageInfo> {
let mut dst = HashMap::new();
for bucket in buckets.iter() {
let e = match self.find(&bucket.name) {
Some(e) => e,
None => continue,
};
let flat = self.flatten(&e);
let mut bui = BucketUsageInfo {
size: flat.size as u64,
versions_count: flat.versions as u64,
objects_count: flat.objects as u64,
delete_markers_count: flat.delete_markers as u64,
object_size_histogram: flat.obj_sizes.to_map(),
object_versions_histogram: flat.obj_versions.to_map(),
..Default::default()
};
if let Some(rs) = &flat.replication_stats {
bui.replica_size = rs.replica_size;
bui.replica_count = rs.replica_count;
for (arn, stat) in rs.targets.iter() {
bui.replication_info.insert(
arn.clone(),
BucketTargetUsageInfo {
replication_pending_size: stat.pending_size,
replicated_size: stat.replicated_size,
replication_failed_size: stat.failed_size,
replication_pending_count: stat.pending_count,
replication_failed_count: stat.failed_count,
replicated_count: stat.replicated_count,
..Default::default()
},
);
}
}
dst.insert(bucket.name.clone(), bui);
}
dst
}
pub fn marshal_msg(&self) -> Result<Vec<u8>> {
let mut buf = Vec::new();
self.serialize(&mut Serializer::new(&mut buf))?;
Ok(buf)
}
pub fn unmarshal(buf: &[u8]) -> Result<Self> {
let t: Self = rmp_serde::from_slice(buf)?;
Ok(t)
}
}
#[derive(Default, Clone)]
struct Inner {
objects: usize,
path: DataUsageHash,
}
fn add(data_usage_cache: &DataUsageCache, path: &DataUsageHash, leaves: &mut Vec<Inner>) {
let e = match data_usage_cache.cache.get(&path.key()) {
Some(e) => e,
None => return,
};
if !e.children.is_empty() {
return;
}
let sz = data_usage_cache.size_recursive(&path.key()).unwrap_or_default();
leaves.push(Inner {
objects: sz.objects,
path: path.clone(),
});
for ch in e.children.iter() {
add(data_usage_cache, &DataUsageHash(ch.clone()), leaves);
}
}
fn mark(duc: &DataUsageCache, entry: &DataUsageEntry, found: &mut HashSet<String>) {
for k in entry.children.iter() {
found.insert(k.to_string());
if let Some(ch) = duc.cache.get(k) {
mark(duc, ch, found);
}
}
}
#[derive(Clone, Debug, Default, Eq, PartialEq)]
pub struct DataUsageHash(pub String);
impl DataUsageHash {
pub fn string(&self) -> String {
self.0.clone()
}
pub fn key(&self) -> String {
self.0.clone()
}
pub fn mod_(&self, cycle: u32, cycles: u32) -> bool {
if cycles <= 1 {
return cycles == 1;
}
let hash = self.calculate_hash();
hash as u32 % cycles == cycle % cycles
}
pub fn mod_alt(&self, cycle: u32, cycles: u32) -> bool {
if cycles <= 1 {
return cycles == 1;
}
let hash = self.calculate_hash();
(hash >> 32) as u32 % cycles == cycle % cycles
}
fn calculate_hash(&self) -> u64 {
let mut hasher = DefaultHasher::new();
self.0.hash(&mut hasher);
hasher.finish()
}
}
pub fn hash_path(data: &str) -> DataUsageHash {
DataUsageHash(Path::new(&data).clean().to_string_lossy().to_string())
}
crates/ecstore/src/heal/error.rs
-19
View File
@@ -1,19 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
pub const ERR_IGNORE_FILE_CONTRIB: &str = "ignore this file's contribution toward data-usage";
pub const ERR_SKIP_FILE: &str = "skip this file";
pub const ERR_HEAL_STOP_SIGNALLED: &str = "heal stop signaled";
pub const ERR_HEAL_IDLE_TIMEOUT: &str = "healing results were not consumed for too long";
pub const ERR_RETRY_HEALING: &str = "some items failed to heal, we will retry healing this drive again";
crates/ecstore/src/heal/heal_commands.rs
-544
View File
@@ -1,544 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use std::{
collections::{HashMap, HashSet},
path::Path,
time::SystemTime,
};
use crate::{
config::storageclass::{RRS, STANDARD},
disk::{BUCKET_META_PREFIX, DeleteOptions, DiskAPI, DiskStore, RUSTFS_META_BUCKET, error::DiskError, fs::read_file},
global::GLOBAL_BackgroundHealState,
heal::heal_ops::HEALING_TRACKER_FILENAME,
new_object_layer_fn,
store_api::{BucketInfo, StorageAPI},
};
use crate::{disk, error::Result};
use chrono::{DateTime, Utc};
use lazy_static::lazy_static;
use serde::{Deserialize, Serialize};
use time::OffsetDateTime;
use tokio::sync::RwLock;
use super::{background_heal_ops::get_local_disks_to_heal, heal_ops::BG_HEALING_UUID};
pub type HealScanMode = usize;
pub const HEAL_UNKNOWN_SCAN: HealScanMode = 0;
pub const HEAL_NORMAL_SCAN: HealScanMode = 1;
pub const HEAL_DEEP_SCAN: HealScanMode = 2;
pub const HEAL_ITEM_METADATA: &str = "metadata";
pub const HEAL_ITEM_BUCKET: &str = "bucket";
pub const HEAL_ITEM_BUCKET_METADATA: &str = "bucket-metadata";
pub const HEAL_ITEM_OBJECT: &str = "object";
pub const DRIVE_STATE_OK: &str = "ok";
pub const DRIVE_STATE_OFFLINE: &str = "offline";
pub const DRIVE_STATE_CORRUPT: &str = "corrupt";
pub const DRIVE_STATE_MISSING: &str = "missing";
pub const DRIVE_STATE_PERMISSION: &str = "permission-denied";
pub const DRIVE_STATE_FAULTY: &str = "faulty";
pub const DRIVE_STATE_ROOT_MOUNT: &str = "root-mount";
pub const DRIVE_STATE_UNKNOWN: &str = "unknown";
pub const DRIVE_STATE_UNFORMATTED: &str = "unformatted"; // only returned by disk
lazy_static! {
pub static ref TIME_SENTINEL: OffsetDateTime = OffsetDateTime::from_unix_timestamp(0).unwrap();
}
#[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(Debug, Serialize, Deserialize)]
pub struct HealStartSuccess {
#[serde(rename = "clientToken")]
pub client_token: String,
#[serde(rename = "clientAddress")]
pub client_address: String,
#[serde(rename = "startTime")]
pub start_time: DateTime<Utc>,
}
impl Default for HealStartSuccess {
fn default() -> Self {
Self {
client_token: Default::default(),
client_address: Default::default(),
start_time: Utc::now(),
}
}
}
pub type HealStopSuccess = HealStartSuccess;
#[derive(Debug, Default, Deserialize, Serialize)]
pub struct HealingTracker {
#[serde(skip_serializing, skip_deserializing)]
pub disk: Option<DiskStore>,
pub id: String,
pub pool_index: Option<usize>,
pub set_index: Option<usize>,
pub disk_index: Option<usize>,
pub path: String,
pub endpoint: String,
pub started: Option<OffsetDateTime>,
pub last_update: Option<SystemTime>,
pub objects_total_count: u64,
pub objects_total_size: u64,
pub items_healed: u64,
pub items_failed: u64,
pub item_skipped: u64,
pub bytes_done: u64,
pub bytes_failed: u64,
pub bytes_skipped: u64,
pub bucket: String,
pub object: String,
pub resume_items_healed: u64,
pub resume_items_failed: u64,
pub resume_items_skipped: u64,
pub resume_bytes_done: u64,
pub resume_bytes_failed: u64,
pub resume_bytes_skipped: u64,
pub queue_buckets: Vec<String>,
pub healed_buckets: Vec<String>,
pub heal_id: String,
pub retry_attempts: u64,
pub finished: bool,
#[serde(skip_serializing, skip_deserializing)]
pub mu: RwLock<bool>,
}
impl HealingTracker {
pub fn marshal_msg(&self) -> disk::error::Result<Vec<u8>> {
Ok(serde_json::to_vec(self)?)
}
pub fn unmarshal_msg(data: &[u8]) -> disk::error::Result<Self> {
Ok(serde_json::from_slice::<HealingTracker>(data)?)
}
pub async fn reset_healing(&mut self) {
let _ = self.mu.write().await;
self.items_healed = 0;
self.items_failed = 0;
self.bytes_done = 0;
self.bytes_failed = 0;
self.resume_items_healed = 0;
self.resume_items_failed = 0;
self.resume_bytes_done = 0;
self.resume_bytes_failed = 0;
self.item_skipped = 0;
self.bytes_skipped = 0;
self.healed_buckets = Vec::new();
self.bucket = String::new();
self.object = String::new();
}
pub async fn get_last_update(&self) -> Option<SystemTime> {
let _ = self.mu.read().await;
self.last_update
}
pub async fn get_bucket(&self) -> String {
let _ = self.mu.read().await;
self.bucket.clone()
}
pub async fn set_bucket(&mut self, bucket: &str) {
let _ = self.mu.write().await;
self.bucket = bucket.to_string();
}
pub async fn get_object(&self) -> String {
let _ = self.mu.read().await;
self.object.clone()
}
pub async fn set_object(&mut self, object: &str) {
let _ = self.mu.write().await;
self.object = object.to_string();
}
pub async fn update_progress(&mut self, success: bool, skipped: bool, by: u64) {
let _ = self.mu.write().await;
if success {
self.items_healed += 1;
self.bytes_done += by;
} else if skipped {
self.item_skipped += 1;
self.bytes_skipped += by;
} else {
self.items_failed += 1;
self.bytes_failed += by;
}
}
pub async fn update(&mut self) -> disk::error::Result<()> {
if let Some(disk) = &self.disk {
if healing(disk.path().to_string_lossy().as_ref()).await?.is_none() {
return Err(DiskError::other(format!("healingTracker: drive {} is not marked as healing", self.id)));
}
let _ = self.mu.write().await;
if self.id.is_empty() || self.pool_index.is_none() || self.set_index.is_none() || self.disk_index.is_none() {
self.id = disk.get_disk_id().await?.map_or("".to_string(), |id| id.to_string());
let disk_location = disk.get_disk_location();
self.pool_index = disk_location.pool_idx;
self.set_index = disk_location.set_idx;
self.disk_index = disk_location.disk_idx;
}
}
self.save().await
}
pub async fn save(&mut self) -> disk::error::Result<()> {
let _ = self.mu.write().await;
if self.pool_index.is_none() || self.set_index.is_none() || self.disk_index.is_none() {
let Some(store) = new_object_layer_fn() else {
return Err(DiskError::other("errServerNotInitialized"));
};
// TODO: check error type
(self.pool_index, self.set_index, self.disk_index) =
store.get_pool_and_set(&self.id).await.map_err(|_| DiskError::DiskNotFound)?;
}
self.last_update = Some(SystemTime::now());
let htracker_bytes = self.marshal_msg()?;
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);
disk.write_all(RUSTFS_META_BUCKET, file_path.to_str().unwrap(), htracker_bytes.into())
.await?;
}
Ok(())
}
pub async fn delete(&self) -> Result<()> {
if let Some(disk) = &self.disk {
let file_path = Path::new(BUCKET_META_PREFIX).join(HEALING_TRACKER_FILENAME);
disk.delete(
RUSTFS_META_BUCKET,
file_path.to_str().unwrap(),
DeleteOptions {
recursive: false,
immediate: false,
..Default::default()
},
)
.await?;
}
Ok(())
}
pub async fn is_healed(&self, bucket: &str) -> bool {
let _ = self.mu.read().await;
for v in self.healed_buckets.iter() {
if v == bucket {
return true;
}
}
false
}
pub async fn resume(&mut self) {
let _ = self.mu.write().await;
self.items_healed = self.resume_items_healed;
self.items_failed = self.resume_items_failed;
self.item_skipped = self.resume_items_skipped;
self.bytes_done = self.resume_bytes_done;
self.bytes_failed = self.resume_bytes_failed;
self.bytes_skipped = self.resume_bytes_skipped;
}
pub async fn bucket_done(&mut self, bucket: &str) {
let _ = self.mu.write().await;
self.resume_items_healed = self.items_healed;
self.resume_items_failed = self.items_failed;
self.resume_items_skipped = self.item_skipped;
self.resume_bytes_done = self.bytes_done;
self.resume_bytes_failed = self.bytes_failed;
self.resume_bytes_skipped = self.bytes_skipped;
self.healed_buckets.push(bucket.to_string());
self.queue_buckets.retain(|x| x != bucket);
}
pub async fn set_queue_buckets(&mut self, buckets: &[BucketInfo]) {
let _ = self.mu.write().await;
buckets.iter().for_each(|bucket| {
if !self.healed_buckets.contains(&bucket.name) {
self.queue_buckets.push(bucket.name.clone());
}
});
}
pub async fn to_healing_disk(&self) -> rustfs_madmin::HealingDisk {
let _ = self.mu.read().await;
rustfs_madmin::HealingDisk {
id: self.id.clone(),
heal_id: self.heal_id.clone(),
pool_index: self.pool_index,
set_index: self.set_index,
disk_index: self.disk_index,
endpoint: self.endpoint.clone(),
path: self.path.clone(),
started: self.started,
last_update: self.last_update,
retry_attempts: self.retry_attempts,
objects_total_count: self.objects_total_count,
objects_total_size: self.objects_total_size,
items_healed: self.items_healed,
items_failed: self.items_failed,
item_skipped: self.item_skipped,
bytes_done: self.bytes_done,
bytes_failed: self.bytes_failed,
bytes_skipped: self.bytes_skipped,
objects_healed: self.items_healed,
objects_failed: self.items_failed,
bucket: self.bucket.clone(),
object: self.object.clone(),
queue_buckets: self.queue_buckets.clone(),
healed_buckets: self.healed_buckets.clone(),
finished: self.finished,
}
}
}
impl Clone for HealingTracker {
fn clone(&self) -> Self {
Self {
disk: self.disk.clone(),
id: self.id.clone(),
pool_index: self.pool_index,
set_index: self.set_index,
disk_index: self.disk_index,
path: self.path.clone(),
endpoint: self.endpoint.clone(),
started: self.started,
last_update: self.last_update,
objects_total_count: self.objects_total_count,
objects_total_size: self.objects_total_size,
items_healed: self.items_healed,
items_failed: self.items_failed,
item_skipped: self.item_skipped,
bytes_done: self.bytes_done,
bytes_failed: self.bytes_failed,
bytes_skipped: self.bytes_skipped,
bucket: self.bucket.clone(),
object: self.object.clone(),
resume_items_healed: self.resume_items_healed,
resume_items_failed: self.resume_items_failed,
resume_items_skipped: self.resume_items_skipped,
resume_bytes_done: self.resume_bytes_done,
resume_bytes_failed: self.resume_bytes_failed,
resume_bytes_skipped: self.resume_bytes_skipped,
queue_buckets: self.queue_buckets.clone(),
healed_buckets: self.healed_buckets.clone(),
heal_id: self.heal_id.clone(),
retry_attempts: self.retry_attempts,
finished: self.finished,
mu: RwLock::new(false),
}
}
}
pub async fn load_healing_tracker(disk: &Option<DiskStore>) -> disk::error::Result<HealingTracker> {
if let Some(disk) = disk {
let disk_id = disk.get_disk_id().await?;
if let Some(disk_id) = disk_id {
let disk_id = disk_id.to_string();
let file_path = Path::new(BUCKET_META_PREFIX).join(HEALING_TRACKER_FILENAME);
let data = disk.read_all(RUSTFS_META_BUCKET, file_path.to_str().unwrap()).await?;
let mut healing_tracker = HealingTracker::unmarshal_msg(&data)?;
if healing_tracker.id != disk_id && !healing_tracker.id.is_empty() {
return Err(DiskError::other(format!(
"loadHealingTracker: drive id mismatch expected {}, got {}",
healing_tracker.id, disk_id
)));
}
healing_tracker.id = disk_id;
healing_tracker.disk = Some(disk.clone());
Ok(healing_tracker)
} else {
Err(DiskError::other("loadHealingTracker: disk not have id"))
}
} else {
Err(DiskError::other("loadHealingTracker: nil drive given"))
}
}
pub async fn init_healing_tracker(disk: DiskStore, heal_id: &str) -> disk::error::Result<HealingTracker> {
let disk_location = disk.get_disk_location();
Ok(HealingTracker {
id: disk
.get_disk_id()
.await
.map_or("".to_string(), |id| id.map_or("".to_string(), |id| id.to_string())),
heal_id: heal_id.to_string(),
path: disk.to_string(),
endpoint: disk.endpoint().to_string(),
started: Some(OffsetDateTime::now_utc()),
pool_index: disk_location.pool_idx,
set_index: disk_location.set_idx,
disk_index: disk_location.disk_idx,
disk: Some(disk),
..Default::default()
})
}
pub async fn healing(derive_path: &str) -> disk::error::Result<Option<HealingTracker>> {
let healing_file = Path::new(derive_path)
.join(RUSTFS_META_BUCKET)
.join(BUCKET_META_PREFIX)
.join(HEALING_TRACKER_FILENAME);
let b = read_file(healing_file).await?;
if b.is_empty() {
return Ok(None);
}
let healing_tracker = HealingTracker::unmarshal_msg(&b)?;
Ok(Some(healing_tracker))
}
#[derive(Debug, Default, Serialize, Deserialize)]
pub struct MRFStatus {
bytes_healed: u64,
items_healed: u64,
}
#[derive(Debug, Default, Serialize, Deserialize)]
pub struct SetStatus {
pub id: String,
pub pool_index: i32,
pub set_index: i32,
pub heal_status: String,
pub heal_priority: String,
pub total_objects: usize,
pub disks: Vec<rustfs_madmin::Disk>,
}
#[derive(Debug, Default, Serialize, Deserialize)]
pub struct BgHealState {
offline_endpoints: Vec<String>,
scanned_items_count: u64,
heal_disks: Vec<String>,
sets: Vec<SetStatus>,
mrf: HashMap<String, MRFStatus>,
scparity: HashMap<String, usize>,
}
pub async fn get_local_background_heal_status() -> (BgHealState, bool) {
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.get_scanned_items_count().await as u64,
..Default::default()
};
let mut heal_disks_map = HashSet::new();
for ep in get_local_disks_to_heal().await.iter() {
heal_disks_map.insert(ep.to_string());
}
let Some(store) = new_object_layer_fn() else {
let healing = GLOBAL_BackgroundHealState.get_local_healing_disks().await;
for disk in healing.values() {
status.heal_disks.push(disk.endpoint.clone());
}
return (status, true);
};
let si = store.local_storage_info().await;
let mut indexed = HashMap::new();
for disk in si.disks.iter() {
let set_idx = format!("{}-{}", disk.pool_index, disk.set_index);
// indexed.insert(set_idx, disk);
indexed.entry(set_idx).or_insert(Vec::new()).push(disk);
}
for (id, disks) in indexed {
let mut ss = SetStatus {
id,
set_index: disks[0].set_index,
pool_index: disks[0].pool_index,
..Default::default()
};
for disk in disks {
ss.disks.push(disk.clone());
if disk.healing {
ss.heal_status = "healing".to_string();
ss.heal_priority = "high".to_string();
status.heal_disks.push(disk.endpoint.clone());
}
}
ss.disks.sort_by(|a, b| {
if a.pool_index != b.pool_index {
return a.pool_index.cmp(&b.pool_index);
}
if a.set_index != b.set_index {
return a.set_index.cmp(&b.set_index);
}
a.disk_index.cmp(&b.disk_index)
});
status.sets.push(ss);
}
status.sets.sort_by(|a, b| a.id.cmp(&b.id));
let backend_info = store.backend_info().await;
status
.scparity
.insert(STANDARD.to_string(), backend_info.standard_sc_parity.unwrap_or_default());
status
.scparity
.insert(RRS.to_string(), backend_info.rr_sc_parity.unwrap_or_default());
(status, true)
}
crates/ecstore/src/heal/heal_ops.rs
-842
View File
@@ -1,842 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use super::{
background_heal_ops::HealTask,
data_scanner::HEAL_DELETE_DANGLING,
error::ERR_SKIP_FILE,
heal_commands::{HEAL_ITEM_BUCKET_METADATA, HealOpts, HealScanMode, HealStopSuccess, HealingTracker},
};
use crate::error::{Error, Result};
use crate::heal::heal_commands::{HEAL_ITEM_BUCKET, HEAL_ITEM_OBJECT};
use crate::store_api::StorageAPI;
use crate::{
config::com::CONFIG_PREFIX,
disk::RUSTFS_META_BUCKET,
global::GLOBAL_BackgroundHealRoutine,
heal::{error::ERR_HEAL_STOP_SIGNALLED, heal_commands::DRIVE_STATE_OK},
};
use crate::{
disk::endpoint::Endpoint,
endpoints::Endpoints,
global::GLOBAL_IsDistErasure,
heal::heal_commands::{HEAL_UNKNOWN_SCAN, HealStartSuccess},
new_object_layer_fn,
};
use chrono::Utc;
use futures::join;
use lazy_static::lazy_static;
use rustfs_filemeta::MetaCacheEntry;
use rustfs_madmin::heal_commands::{HealDriveInfo, HealItemType, HealResultItem};
use rustfs_utils::path::has_prefix;
use rustfs_utils::path::path_join;
use serde::{Deserialize, Serialize};
use std::{
collections::HashMap,
future::Future,
path::PathBuf,
pin::Pin,
sync::Arc,
time::{Duration, SystemTime, UNIX_EPOCH},
};
use tokio::{
select, spawn,
sync::{
RwLock, broadcast,
mpsc::{self, Receiver as M_Receiver, Sender as M_Sender},
watch::{self, Receiver as W_Receiver, Sender as W_Sender},
},
time::{interval, sleep},
};
use tracing::{error, info};
use uuid::Uuid;
type HealStatusSummary = String;
type ItemsMap = HashMap<HealItemType, usize>;
pub type HealEntryFn =
Arc<dyn Fn(String, MetaCacheEntry, HealScanMode) -> Pin<Box<dyn Future<Output = Result<()>> + Send>> + Send + Sync + 'static>;
pub const BG_HEALING_UUID: &str = "0000-0000-0000-0000";
pub const HEALING_TRACKER_FILENAME: &str = ".healing.bin";
const KEEP_HEAL_SEQ_STATE_DURATION: Duration = Duration::from_secs(10 * 60);
const HEAL_NOT_STARTED_STATUS: &str = "not started";
const HEAL_RUNNING_STATUS: &str = "running";
const HEAL_STOPPED_STATUS: &str = "stopped";
const HEAL_FINISHED_STATUS: &str = "finished";
pub const RUSTFS_RESERVED_BUCKET: &str = "rustfs";
pub const RUSTFS_RESERVED_BUCKET_PATH: &str = "/rustfs";
pub const LOGIN_PATH_PREFIX: &str = "/login";
const MAX_UNCONSUMED_HEAL_RESULT_ITEMS: usize = 1000;
const HEAL_UNCONSUMED_TIMEOUT: Duration = Duration::from_secs(24 * 60 * 60);
pub const NOP_HEAL: &str = "";
lazy_static! {}
#[derive(Clone, Debug, Default, Serialize, Deserialize)]
pub struct HealSequenceStatus {
pub summary: HealStatusSummary,
pub failure_detail: String,
pub start_time: u64,
pub heal_setting: HealOpts,
pub items: Vec<HealResultItem>,
}
#[derive(Debug, Default)]
pub struct HealSource {
pub bucket: String,
pub object: String,
pub version_id: String,
pub no_wait: bool,
pub opts: Option<HealOpts>,
}
#[derive(Debug)]
pub struct HealSequence {
pub bucket: String,
pub object: String,
pub report_progress: bool,
pub start_time: SystemTime,
pub end_time: Arc<RwLock<SystemTime>>,
pub client_token: String,
pub client_address: String,
pub force_started: bool,
pub setting: HealOpts,
pub current_status: Arc<RwLock<HealSequenceStatus>>,
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: W_Sender<bool>,
rx: W_Receiver<bool>,
}
pub fn new_bg_heal_sequence() -> HealSequence {
let hs = HealOpts {
remove: HEAL_DELETE_DANGLING,
..Default::default()
};
HealSequence {
start_time: SystemTime::now(),
client_token: BG_HEALING_UUID.to_string(),
bucket: RUSTFS_RESERVED_BUCKET.to_string(),
setting: hs,
current_status: Arc::new(RwLock::new(HealSequenceStatus {
summary: HEAL_NOT_STARTED_STATUS.to_string(),
heal_setting: hs,
..Default::default()
})),
report_progress: false,
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()
}
}
impl Default for HealSequence {
fn default() -> Self {
let (h_tx, h_rx) = mpsc::channel(1);
let (tx, rx) = watch::channel(false);
Self {
bucket: Default::default(),
object: Default::default(),
report_progress: Default::default(),
start_time: SystemTime::now(),
end_time: Arc::new(RwLock::new(SystemTime::now())),
client_token: Default::default(),
client_address: Default::default(),
force_started: Default::default(),
setting: Default::default(),
current_status: Default::default(),
last_sent_result_index: Default::default(),
scanned_items_map: Default::default(),
healed_items_map: Default::default(),
heal_failed_items_map: 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,
rx,
}
}
}
impl HealSequence {
pub fn new(bucket: &str, obj_prefix: &str, client_addr: &str, hs: HealOpts, force_start: bool) -> Self {
let client_token = Uuid::new_v4().to_string();
Self {
bucket: bucket.to_string(),
object: obj_prefix.to_string(),
report_progress: true,
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()
})),
..Default::default()
}
}
}
impl HealSequence {
pub async fn get_scanned_items_count(&self) -> usize {
self.scanned_items_map.read().await.values().sum()
}
async fn _get_scanned_items_map(&self) -> ItemsMap {
self.scanned_items_map.read().await.clone()
}
async fn _get_healed_items_map(&self) -> ItemsMap {
self.healed_items_map.read().await.clone()
}
async fn _get_heal_failed_items_map(&self) -> ItemsMap {
self.heal_failed_items_map.read().await.clone()
}
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 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 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 {
if let Ok(true) = self.rx.has_changed() {
info!("quited");
return true;
}
false
}
async fn has_ended(&self) -> bool {
if self.client_token == *BG_HEALING_UUID {
return false;
}
*(self.end_time.read().await) != self.start_time
}
async fn stop(&self) {
let _ = self.tx.send(true);
}
async fn push_heal_result_item(&self, r: &HealResultItem) -> Result<()> {
let mut r = r.clone();
let mut interval_timer = interval(HEAL_UNCONSUMED_TIMEOUT);
#[allow(unused_assignments)]
let mut items_len = 0;
loop {
{
let current_status_r = self.current_status.read().await;
items_len = current_status_r.items.len();
}
if items_len == MAX_UNCONSUMED_HEAL_RESULT_ITEMS {
select! {
_ = sleep(Duration::from_secs(1)) => {
}
_ = self.is_done() => {
return Err(Error::other("stopped"));
}
_ = interval_timer.tick() => {
return Err(Error::other("timeout"));
}
}
} else {
break;
}
}
let mut current_status_w = self.current_status.write().await;
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.read().await;
}
current_status_w.items.push(r);
Ok(())
}
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()).await;
if source.no_wait {
let task_str = format!("{task:?}");
if GLOBAL_BackgroundHealRoutine.tasks_tx.try_send(task).is_ok() {
info!("Task in the queue: {:?}", task_str);
}
return Ok(());
}
let (resp_tx, mut resp_rx) = mpsc::channel(1);
task.resp_tx = Some(resp_tx);
let task_str = format!("{task:?}");
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;
for drive in drivers.iter() {
if drive.state == DRIVE_STATE_OK {
count += 1;
}
}
count
};
match resp_rx.recv().await {
Some(mut res) => {
if res.err.is_none() {
self.count_healed(heal_type.clone()).await;
} else {
self.count_failed(heal_type.clone()).await;
}
if !self.report_progress {
return if let Some(err) = res.err {
if err.to_string() == ERR_SKIP_FILE {
return Ok(());
}
Err(err)
} else {
Ok(())
};
}
res.result.heal_item_type = heal_type.clone();
if let Some(err) = res.err.as_ref() {
res.result.detail = err.to_string();
}
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.drives);
if got < res.result.parity_blocks {
res.result.detail = format!(
"quorum loss - expected {} minimum, got drive states in OK {}",
res.result.parity_blocks, got
);
}
}
info!("queue_heal_task, HealResult: {:?}", res);
self.push_heal_result_item(&res.result).await
}
None => Ok(()),
}
}
async fn heal_disk_meta(h: Arc<HealSequence>) -> Result<()> {
HealSequence::heal_rustfs_sys_meta(h, CONFIG_PREFIX).await
}
async fn heal_items(h: Arc<HealSequence>, buckets_only: bool) -> Result<()> {
if h.client_token == *BG_HEALING_UUID {
return Ok(());
}
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<HealSequence>) {
let buckets_only = false;
let result = Self::heal_items(h.clone(), buckets_only).await.err();
let _ = h.traverse_and_heal_done_tx.read().await.send(result).await;
}
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::other("errServerNotInitialized"));
};
let setting = h.setting;
store
.heal_objects(RUSTFS_META_BUCKET, meta_prefix, &setting, h.clone(), true)
.await
}
async fn is_done(&self) -> bool {
if let Ok(true) = self.rx.has_changed() {
return true;
}
false
}
pub async fn heal_bucket(hs: Arc<HealSequence>, bucket: &str, bucket_only: bool) -> Result<()> {
info!("heal_bucket, hs: {:?}", hs);
let (object, setting) = {
hs.queue_heal_task(
HealSource {
bucket: bucket.to_string(),
..Default::default()
},
HEAL_ITEM_BUCKET.to_string(),
)
.await?;
if bucket_only {
return Ok(());
}
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.object.clone(), hs.setting)
};
let Some(store) = new_object_layer_fn() else {
return Err(Error::other("errServerNotInitialized"));
};
store.heal_objects(bucket, &object, &setting, hs.clone(), false).await
}
pub async fn heal_object(
hs: Arc<HealSequence>,
bucket: &str,
object: &str,
version_id: &str,
_scan_mode: HealScanMode,
) -> Result<()> {
info!("heal_object");
if hs.is_quitting().await {
info!("heal_object hs is quitting");
return Err(Error::other(ERR_HEAL_STOP_SIGNALLED));
}
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(hs.setting),
..Default::default()
},
HEAL_ITEM_OBJECT.to_string(),
)
.await?;
Ok(())
}
pub async fn heal_meta_object(
hs: Arc<HealSequence>,
bucket: &str,
object: &str,
version_id: &str,
_scan_mode: HealScanMode,
) -> Result<()> {
if hs.is_quitting().await {
return Err(Error::other(ERR_HEAL_STOP_SIGNALLED));
}
hs.queue_heal_task(
HealSource {
bucket: bucket.to_string(),
object: object.to_string(),
version_id: version_id.to_string(),
..Default::default()
},
HEAL_ITEM_BUCKET_METADATA.to_string(),
)
.await?;
Ok(())
}
}
pub async fn heal_sequence_start(h: Arc<HealSequence>) {
{
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)
.expect("Time went backwards")
.as_secs();
}
let h_clone = h.clone();
spawn(async move {
HealSequence::traverse_and_heal(h_clone).await;
});
let h_clone_1 = h.clone();
let mut x = h.traverse_and_heal_done_rx.write().await;
select! {
_ = 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 mut rx_w = h_clone_1.traverse_and_heal_done_rx.write().await;
rx_w.recv().await;
});
}
result = x.recv() => {
if let Some(err) = result {
match err {
Some(err) => {
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 = h.current_status.write().await;
current_status_w.summary = HEAL_FINISHED_STATUS.to_string();
}
}
}
}
}
}
#[derive(Debug, Default)]
pub struct AllHealState {
mu: RwLock<bool>,
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<Self> {
let state = Arc::new(AllHealState::default());
let (_, mut rx) = broadcast::channel(1);
if cleanup {
let state_clone = state.clone();
spawn(async move {
loop {
select! {
result = rx.recv() =>{
if let Ok(true) = result {
return;
}
}
_ = sleep(Duration::from_secs(5 * 60)) => {
state_clone.periodic_heal_seqs_clean().await;
}
}
}
});
}
state
}
pub async fn pop_heal_local_disks(&self, heal_local_disks: &[Endpoint]) {
let _ = self.mu.write().await;
self.heal_local_disks.write().await.retain(|k, _| {
if heal_local_disks.contains(k) {
return false;
}
true
});
let heal_local_disks = heal_local_disks.iter().map(|s| s.to_string()).collect::<Vec<_>>();
self.heal_status.write().await.retain(|_, v| {
if heal_local_disks.contains(&v.endpoint) {
return false;
}
true
});
}
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::other("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::other(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::other(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.write().await.insert(tracker.id.clone(), tracker.clone());
}
pub async fn get_local_healing_disks(&self) -> HashMap<String, rustfs_madmin::HealingDisk> {
let _ = self.mu.read().await;
let mut dst = HashMap::new();
for v in self.heal_status.read().await.values() {
dst.insert(v.endpoint.clone(), v.to_healing_disk().await);
}
dst
}
pub async fn get_heal_local_disk_endpoints(&self) -> Endpoints {
let _ = self.mu.read().await;
let mut endpoints = Vec::new();
self.heal_local_disks.read().await.iter().for_each(|(k, v)| {
if !v {
endpoints.push(k.clone());
}
});
Endpoints::from(endpoints)
}
pub async fn set_disk_healing_status(&self, ep: Endpoint, healing: bool) {
let _ = self.mu.write().await;
self.heal_local_disks.write().await.insert(ep, healing);
}
pub async fn push_heal_local_disks(&self, heal_local_disks: &[Endpoint]) {
let _ = self.mu.write().await;
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(&self) {
let _ = self.mu.write().await;
let now = SystemTime::now();
let mut keys_to_remove = Vec::new();
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_remove.push(k.clone())
}
}
for key in keys_to_remove.iter() {
self.heal_seq_map.write().await.remove(key);
}
}
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.read().await.values() {
if v.client_token == token {
return (Some(v.clone()), true);
}
}
(None, false)
}
pub async fn get_heal_sequence(&self, path: &str) -> Option<Arc<HealSequence>> {
let _ = self.mu.read().await;
self.heal_seq_map.read().await.get(path).cloned()
}
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 client_token = he.client_token.clone();
if *GLOBAL_IsDistErasure.read().await {
// TODO: proxy
}
hsp.client_token = client_token;
hsp.client_address = he.client_address.clone();
hsp.start_time = Utc::now();
he.stop().await;
loop {
if he.has_ended().await {
break;
}
sleep(Duration::from_secs(1)).await;
}
let _ = self.mu.write().await;
self.heal_seq_map.write().await.remove(path);
} else {
hsp.client_token = "unknown".to_string();
}
let b = serde_json::to_string(&hsp)?;
Ok(b.as_bytes().to_vec())
}
// LaunchNewHealSequence - launches a background routine that performs
// healing according to the healSequence argument. For each heal
// sequence, state is stored in the `globalAllHealState`, which is a
// map of the heal path to `healSequence` which holds state about the
// heal sequence.
//
// Heal results are persisted in server memory for
// `keepHealSeqStateDuration`. This function also launches a
// background routine to clean up heal results after the
// aforementioned duration.
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 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.has_ended().await {
return Err(Error::other(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.read().await.iter() {
if (has_prefix(k, path_s) || has_prefix(path_s, k)) && !v.has_ended().await {
return Err(Error::other(format!(
"The provided heal sequence path overlaps with an existing heal path: {k}"
)));
}
}
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 {
// TODO: proxy
}
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();
spawn(async {
heal_sequence_start(heal_sequence_clone).await;
});
}
let b = serde_json::to_vec(&HealStartSuccess {
client_token,
client_address: heal_sequence.client_address.clone(),
// start_time: Utc::now(),
start_time: heal_sequence.start_time.into(),
})?;
Ok(b)
}
}
crates/ecstore/src/heal/mod.rs
-23
View File
@@ -1,23 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
// pub mod background_heal_ops;
// pub mod data_scanner;
// pub mod data_scanner_metric;
// pub mod data_usage;
// pub mod data_usage_cache;
// pub mod error;
// pub mod heal_commands;
// pub mod heal_ops;
// pub mod mrf;
crates/ecstore/src/heal/mrf.rs
-183
View File
@@ -1,183 +0,0 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET};
use crate::heal::background_heal_ops::{heal_bucket, heal_object};
use crate::heal::heal_commands::{HEAL_DEEP_SCAN, HEAL_NORMAL_SCAN};
use chrono::{DateTime, Utc};
use lazy_static::lazy_static;
use regex::Regex;
use rustfs_utils::path::SLASH_SEPARATOR;
use std::ops::Sub;
use std::sync::atomic::{AtomicBool, Ordering};
use std::time::Duration;
use tokio::sync::RwLock;
use tokio::sync::mpsc::{Receiver, Sender};
use tokio::time::sleep;
use tokio_util::sync::CancellationToken;
use tracing::{error, info};
use uuid::Uuid;
pub const MRF_OPS_QUEUE_SIZE: u64 = 100000;
pub const HEAL_DIR: &str = ".heal";
pub const HEAL_MRFMETA_FORMAT: u64 = 1;
pub const HEAL_MRFMETA_VERSION_V1: u64 = 1;
lazy_static! {
pub static ref HEAL_MRF_DIR: String =
format!("{}{}{}{}{}", BUCKET_META_PREFIX, SLASH_SEPARATOR, HEAL_DIR, SLASH_SEPARATOR, "mrf");
static ref PATTERNS: Vec<Regex> = vec![
Regex::new(r"^buckets/.*/.metacache/.*").unwrap(),
Regex::new(r"^tmp/.*").unwrap(),
Regex::new(r"^multipart/.*").unwrap(),
Regex::new(r"^tmp-old/.*").unwrap(),
];
}
#[derive(Default)]
pub struct PartialOperation {
pub bucket: String,
pub object: String,
pub version_id: Option<String>,
pub versions: Vec<u8>,
pub set_index: usize,
pub pool_index: usize,
pub queued: DateTime<Utc>,
pub bitrot_scan: bool,
}
pub struct MRFState {
tx: Sender<PartialOperation>,
rx: RwLock<Receiver<PartialOperation>>,
closed: AtomicBool,
closing: AtomicBool,
}
impl Default for MRFState {
fn default() -> Self {
Self::new()
}
}
impl MRFState {
pub fn new() -> MRFState {
let (tx, rx) = tokio::sync::mpsc::channel(MRF_OPS_QUEUE_SIZE as usize);
MRFState {
tx,
rx: RwLock::new(rx),
closed: Default::default(),
closing: Default::default(),
}
}
pub async fn add_partial(&self, op: PartialOperation) {
if self.closed.load(Ordering::SeqCst) || self.closing.load(Ordering::SeqCst) {
return;
}
let _ = self.tx.send(op).await;
}
/// Enhanced heal routine with cancellation support
///
/// This method implements the same healing logic as the original heal_routine,
/// but adds proper cancellation support via CancellationToken.
/// The core logic remains identical to maintain compatibility.
pub async fn heal_routine_with_cancel(&self, cancel_token: CancellationToken) {
info!("MRF heal routine started with cancellation support");
loop {
tokio::select! {
_ = cancel_token.cancelled() => {
info!("MRF heal routine received shutdown signal, exiting gracefully");
break;
}
op_result = async {
let mut rx_guard = self.rx.write().await;
rx_guard.recv().await
} => {
if let Some(op) = op_result {
// Special path filtering (original logic)
if op.bucket == RUSTFS_META_BUCKET {
for pattern in &*PATTERNS {
if pattern.is_match(&op.object) {
continue; // Skip this operation, continue with next
}
}
}
// Network reconnection delay (original logic)
let now = Utc::now();
if now.sub(op.queued).num_seconds() < 1 {
tokio::select! {
_ = cancel_token.cancelled() => {
info!("MRF heal routine cancelled during reconnection delay");
break;
}
_ = sleep(Duration::from_secs(1)) => {}
}
}
// Core healing logic (original logic preserved)
let scan_mode = if op.bitrot_scan { HEAL_DEEP_SCAN } else { HEAL_NORMAL_SCAN };
if op.object.is_empty() {
// Heal bucket (original logic)
if let Err(err) = heal_bucket(&op.bucket).await {
error!("heal bucket failed, bucket: {}, err: {:?}", op.bucket, err);
}
} else if op.versions.is_empty() {
// Heal single object (original logic)
if let Err(err) = heal_object(
&op.bucket,
&op.object,
&op.version_id.clone().unwrap_or_default(),
scan_mode
).await {
error!("heal object failed, bucket: {}, object: {}, err: {:?}", op.bucket, op.object, err);
}
} else {
// Heal multiple versions (original logic)
let vers = op.versions.len() / 16;
if vers > 0 {
for i in 0..vers {
// Check for cancellation before each version
if cancel_token.is_cancelled() {
info!("MRF heal routine cancelled during version processing");
return;
}
let start = i * 16;
let end = start + 16;
if let Err(err) = heal_object(
&op.bucket,
&op.object,
&Uuid::from_slice(&op.versions[start..end]).expect("").to_string(),
scan_mode,
).await {
error!("heal object failed, bucket: {}, object: {}, err: {:?}", op.bucket, op.object, err);
}
}
}
}
} else {
info!("MRF heal routine channel closed, exiting");
break;
}
}
}
}
info!("MRF heal routine stopped gracefully");
}
}
crates/ecstore/src/lib.rs
-1
View File
@@ -30,7 +30,6 @@ pub mod endpoints;
pub mod erasure_coding;
pub mod error;
pub mod global;
pub mod heal;
pub mod lock_utils;
pub mod metrics_realtime;
pub mod notification_sys;