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

refactor(heal): unify heal request interface, add disk field, update ahm/ecstore/common for erasure set healing

Browse Source 
Signed-off-by: junxiang Mu <1948535941@qq.com>
This commit is contained in:
junxiang Mu
2025-07-21 18:16:42 +08:00
parent 3d3c6e4e06
commit ea210d52dc
29 changed files with 2119 additions and 639 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ahm/src/heal/channel.rs
+22 -19
View File
@@ -19,7 +19,7 @@ use crate::heal::{
};
use rustfs_common::heal_channel::{
HealChannelCommand, HealChannelPriority, HealChannelReceiver, HealChannelRequest, HealChannelResponse,
HealChannelCommand, HealChannelPriority, HealChannelReceiver, HealChannelRequest, HealChannelResponse, HealScanMode,
};
use std::sync::Arc;
use tokio::sync::mpsc;
@@ -173,15 +173,27 @@ impl HealChannelProcessor {
/// Convert channel request to heal request
fn convert_to_heal_request(&self, request: HealChannelRequest) -> Result<HealRequest> {
let heal_type = match &request.object_prefix {
Some(prefix) if !prefix.is_empty() => HealType::Object {
let heal_type = if let Some(disk_id) = &request.disk {
HealType::ErasureSet {
buckets: vec![],
set_disk_id: disk_id.clone(),
}
} else if let Some(prefix) = &request.object_prefix {
if !prefix.is_empty() {
HealType::Object {
bucket: request.bucket.clone(),
object: prefix.clone(),
version_id: None,
}
} else {
HealType::Bucket {
bucket: request.bucket.clone(),
}
}
} else {
HealType::Bucket {
bucket: request.bucket.clone(),
object: prefix.clone(),
version_id: None,
},
_ => HealType::Bucket {
bucket: request.bucket.clone(),
},
}
};
let priority = match request.priority {
@@ -191,18 +203,9 @@ impl HealChannelProcessor {
HealChannelPriority::Critical => HealPriority::Urgent,
};
// Convert scan mode
let scan_mode = match request.scan_mode {
Some(rustfs_common::heal_channel::HealChannelScanMode::Normal) => {
rustfs_ecstore::heal::heal_commands::HEAL_NORMAL_SCAN
}
Some(rustfs_common::heal_channel::HealChannelScanMode::Deep) => rustfs_ecstore::heal::heal_commands::HEAL_DEEP_SCAN,
None => rustfs_ecstore::heal::heal_commands::HEAL_NORMAL_SCAN,
};
// Build HealOptions with all available fields
let mut options = HealOptions {
scan_mode,
scan_mode: request.scan_mode.unwrap_or(HealScanMode::Normal),
remove_corrupted: request.remove_corrupted.unwrap_or(false),
recreate_missing: request.recreate_missing.unwrap_or(true),
update_parity: request.update_parity.unwrap_or(true),
crates/ahm/src/heal/erasure_healer.rs
+4 -6
View File
@@ -19,10 +19,8 @@ use crate::heal::{
storage::HealStorageAPI,
};
use futures::future::join_all;
use rustfs_ecstore::{
disk::DiskStore,
heal::heal_commands::{HealOpts, HEAL_NORMAL_SCAN},
};
use rustfs_common::heal_channel::{HealOpts, HealScanMode};
use rustfs_ecstore::disk::DiskStore;
use std::sync::Arc;
use tokio::sync::RwLock;
use tracing::{error, info, warn};
@@ -252,7 +250,7 @@ impl ErasureSetHealer {
// heal object
let heal_opts = HealOpts {
scan_mode: HEAL_NORMAL_SCAN,
scan_mode: HealScanMode::Normal,
remove: true,
recreate: true,
..Default::default()
@@ -361,7 +359,7 @@ impl ErasureSetHealer {
// 4. heal objects concurrently
let heal_opts = HealOpts {
scan_mode: HEAL_NORMAL_SCAN,
scan_mode: HealScanMode::Normal,
remove: true, // remove corrupted data
recreate: true, // recreate missing data
..Default::default()
crates/ahm/src/heal/manager.rs
-6
View File
@@ -288,12 +288,6 @@ impl HealManager {
continue;
}
}
// disk currently healing and not finished
if let Some(h) = disk.healing().await {
if !h.finished {
endpoints.push(disk.endpoint());
}
}
}
}
crates/ahm/src/heal/storage.rs
+3 -3
View File
@@ -14,9 +14,9 @@
use crate::error::{Error, Result};
use async_trait::async_trait;
use rustfs_common::heal_channel::{HealOpts, HealScanMode};
use rustfs_ecstore::{
disk::{endpoint::Endpoint, DiskStore},
heal::heal_commands::{HealOpts, HEAL_DEEP_SCAN, HEAL_NORMAL_SCAN},
store::ECStore,
store_api::{BucketInfo, ObjectIO, StorageAPI},
};
@@ -238,7 +238,7 @@ impl HealStorageAPI for ECStoreHealStorage {
dry_run: false,
remove: false,
recreate: true,
scan_mode: HEAL_DEEP_SCAN,
scan_mode: HealScanMode::Deep,
update_parity: true,
no_lock: false,
pool: None,
@@ -322,7 +322,7 @@ impl HealStorageAPI for ECStoreHealStorage {
dry_run: false,
remove: false,
recreate: false,
scan_mode: HEAL_NORMAL_SCAN,
scan_mode: HealScanMode::Normal,
update_parity: false,
no_lock: false,
pool: None,
crates/ahm/src/heal/task.rs
+22 -14
View File
@@ -13,9 +13,9 @@
// limitations under the License.
use crate::error::{Error, Result};
use crate::heal::{erasure_healer::ErasureSetHealer, progress::HealProgress, storage::HealStorageAPI};
use rustfs_ecstore::heal::heal_commands::HealScanMode;
use rustfs_ecstore::heal::heal_commands::HEAL_NORMAL_SCAN;
use crate::heal::ErasureSetHealer;
use crate::heal::{progress::HealProgress, storage::HealStorageAPI};
use rustfs_common::heal_channel::{HealOpts, HealScanMode};
use serde::{Deserialize, Serialize};
use std::sync::Arc;
use std::time::{Duration, SystemTime};
@@ -93,7 +93,7 @@ pub struct HealOptions {
impl Default for HealOptions {
fn default() -> Self {
Self {
scan_mode: HEAL_NORMAL_SCAN,
scan_mode: HealScanMode::Normal,
remove_corrupted: false,
recreate_missing: true,
update_parity: true,
@@ -324,7 +324,7 @@ impl HealTask {
// Step 2: directly call ecstore to perform heal
info!("Step 2: Performing heal using ecstore");
let heal_opts = rustfs_ecstore::heal::heal_commands::HealOpts {
let heal_opts = HealOpts {
recursive: self.options.recursive,
dry_run: self.options.dry_run,
remove: self.options.remove_corrupted,
@@ -410,12 +410,12 @@ impl HealTask {
info!("Attempting to recreate missing object: {}/{}", bucket, object);
// Use ecstore's heal_object with recreate option
let heal_opts = rustfs_ecstore::heal::heal_commands::HealOpts {
let heal_opts = HealOpts {
recursive: false,
dry_run: self.options.dry_run,
remove: false,
recreate: true,
scan_mode: rustfs_ecstore::heal::heal_commands::HEAL_DEEP_SCAN,
scan_mode: HealScanMode::Deep,
update_parity: true,
no_lock: false,
pool: None,
@@ -476,7 +476,7 @@ impl HealTask {
// Step 2: Perform bucket heal using ecstore
info!("Step 2: Performing bucket heal using ecstore");
let heal_opts = rustfs_ecstore::heal::heal_commands::HealOpts {
let heal_opts = HealOpts {
recursive: self.options.recursive,
dry_run: self.options.dry_run,
remove: self.options.remove_corrupted,
@@ -538,12 +538,12 @@ impl HealTask {
// Step 2: Perform metadata heal using ecstore
info!("Step 2: Performing metadata heal using ecstore");
let heal_opts = rustfs_ecstore::heal::heal_commands::HealOpts {
let heal_opts = HealOpts {
recursive: false,
dry_run: self.options.dry_run,
remove: false,
recreate: false,
scan_mode: rustfs_ecstore::heal::heal_commands::HEAL_DEEP_SCAN,
scan_mode: HealScanMode::Deep,
update_parity: false,
no_lock: false,
pool: self.options.pool_index,
@@ -612,12 +612,12 @@ impl HealTask {
// Step 1: Perform MRF heal using ecstore
info!("Step 1: Performing MRF heal using ecstore");
let heal_opts = rustfs_ecstore::heal::heal_commands::HealOpts {
let heal_opts = HealOpts {
recursive: true,
dry_run: self.options.dry_run,
remove: self.options.remove_corrupted,
recreate: self.options.recreate_missing,
scan_mode: rustfs_ecstore::heal::heal_commands::HEAL_DEEP_SCAN,
scan_mode: HealScanMode::Deep,
update_parity: true,
no_lock: false,
pool: None,
@@ -685,12 +685,12 @@ impl HealTask {
// Step 2: Perform EC decode heal using ecstore
info!("Step 2: Performing EC decode heal using ecstore");
let heal_opts = rustfs_ecstore::heal::heal_commands::HealOpts {
let heal_opts = HealOpts {
recursive: false,
dry_run: self.options.dry_run,
remove: false,
recreate: true,
scan_mode: rustfs_ecstore::heal::heal_commands::HEAL_DEEP_SCAN,
scan_mode: HealScanMode::Deep,
update_parity: true,
no_lock: false,
pool: None,
@@ -748,6 +748,14 @@ impl HealTask {
progress.update_progress(0, 4, 0, 0);
}
let buckets = if buckets.is_empty() {
info!("No buckets specified, listing all buckets");
let bucket_infos = self.storage.list_buckets().await?;
bucket_infos.into_iter().map(|info| info.name).collect()
} else {
buckets
};
// Step 1: Perform disk format heal using ecstore
info!("Step 1: Performing disk format heal using ecstore");
match self.storage.heal_format(self.options.dry_run).await {
crates/ahm/src/scanner/data_scanner.rs
+9 -8
View File
@@ -22,22 +22,22 @@ use ecstore::{
disk::{DiskAPI, DiskStore, WalkDirOptions},
set_disk::SetDisks,
};
use rustfs_ecstore::{self as ecstore, StorageAPI};
use rustfs_ecstore::{self as ecstore, data_usage::store_data_usage_in_backend, StorageAPI};
use rustfs_filemeta::MetacacheReader;
use tokio::sync::{Mutex, RwLock};
use tokio_util::sync::CancellationToken;
use tracing::{debug, error, info, warn};
use super::{
data_usage::DataUsageInfo,
metrics::{BucketMetrics, DiskMetrics, MetricsCollector, ScannerMetrics},
};
use super::metrics::{BucketMetrics, DiskMetrics, MetricsCollector, ScannerMetrics};
use crate::heal::HealManager;
use crate::{
error::{Error, Result},
get_ahm_services_cancel_token, HealRequest,
};
use rustfs_common::metrics::{globalMetrics, Metric, Metrics};
use rustfs_common::{
data_usage::DataUsageInfo,
metrics::{globalMetrics, Metric, Metrics},
};
use rustfs_ecstore::disk::RUSTFS_META_BUCKET;
@@ -1182,7 +1182,7 @@ impl Scanner {
// Offload persistence to background task
let data_clone = data_usage.clone();
tokio::spawn(async move {
if let Err(e) = super::data_usage::store_data_usage_in_backend(data_clone, store).await {
if let Err(e) = store_data_usage_in_backend(data_clone, store).await {
error!("Failed to store data usage statistics to backend: {}", e);
} else {
info!("Successfully stored data usage statistics to backend");
@@ -1214,6 +1214,7 @@ impl Scanner {
#[cfg(test)]
mod tests {
use super::*;
use rustfs_ecstore::data_usage::load_data_usage_from_backend;
use rustfs_ecstore::disk::endpoint::Endpoint;
use rustfs_ecstore::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints};
use rustfs_ecstore::store::ECStore;
@@ -1441,7 +1442,7 @@ mod tests {
// verify correctness of persisted data
tokio::time::sleep(std::time::Duration::from_millis(100)).await;
let persisted = crate::scanner::data_usage::load_data_usage_from_backend(ecstore.clone())
let persisted = load_data_usage_from_backend(ecstore.clone())
.await
.expect("load persisted usage");
assert_eq!(persisted.objects_total_count, du_after.objects_total_count);