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

fix(startup): Only monitor disk health after format loading (#1854)

Browse Source 
Co-authored-by: houseme <housemecn@gmail.com>
Co-authored-by: loverustfs <hello@rustfs.com>
This commit is contained in:
evan slack
2026-02-21 07:46:14 -05:00
committed by GitHub
parent f31cd4b716
commit 23f7ffe36b
4 changed files with 58 additions and 41 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/disk/disk_store.rs
+19 -22
View File
@@ -20,7 +20,6 @@ use crate::disk::{
};
use bytes::Bytes;
use rustfs_filemeta::{FileInfo, ObjectPartInfo, RawFileInfo};
use rustfs_utils::string::parse_bool_with_default;
use std::{
path::PathBuf,
sync::{
@@ -39,6 +38,7 @@ const DISK_HEALTH_OK: u32 = 0;
const DISK_HEALTH_FAULTY: u32 = 1;
pub const ENV_RUSTFS_DRIVE_ACTIVE_MONITORING: &str = "RUSTFS_DRIVE_ACTIVE_MONITORING";
pub const DEFAULT_RUSTFS_DRIVE_ACTIVE_MONITORING: bool = true;
pub const ENV_RUSTFS_DRIVE_MAX_TIMEOUT_DURATION: &str = "RUSTFS_DRIVE_MAX_TIMEOUT_DURATION";
pub const CHECK_EVERY: Duration = Duration::from_secs(15);
pub const SKIP_IF_SUCCESS_BEFORE: Duration = Duration::from_secs(5);
@@ -180,40 +180,37 @@ pub struct LocalDiskWrapper {
impl LocalDiskWrapper {
/// Create a new LocalDiskWrapper
pub fn new(disk: Arc<LocalDisk>, health_check: bool) -> Self {
// Check environment variable for health check override
// Default to true if not set, but only enable if both param and env are true
let env_health_check = std::env::var(ENV_RUSTFS_DRIVE_ACTIVE_MONITORING)
.map(|v| parse_bool_with_default(&v, true))
.unwrap_or(true);
// Check environment variable for health check override.
// Only enable if both param and env are true.
let env_health_check =
rustfs_utils::get_env_bool(ENV_RUSTFS_DRIVE_ACTIVE_MONITORING, DEFAULT_RUSTFS_DRIVE_ACTIVE_MONITORING);
let ret = Self {
Self {
disk,
health: Arc::new(DiskHealthTracker::new()),
health_check: health_check && env_health_check,
cancel_token: CancellationToken::new(),
disk_id: Arc::new(RwLock::new(None)),
};
ret.start_monitoring();
ret
}
}
pub fn get_disk(&self) -> Arc<LocalDisk> {
self.disk.clone()
}
/// Start the disk monitoring if health_check is enabled
pub fn start_monitoring(&self) {
if self.health_check {
let health = Arc::clone(&self.health);
let cancel_token = self.cancel_token.clone();
let disk = Arc::clone(&self.disk);
tokio::spawn(async move {
Self::monitor_disk_writable(disk, health, cancel_token).await;
});
/// Enable health monitoring after disk creation.
/// Used to defer health checks until after startup format loading completes.
pub fn enable_health_check(&self) {
if !self.health_check {
return;
}
let health = Arc::clone(&self.health);
let cancel_token = self.cancel_token.clone();
let disk = Arc::clone(&self.disk);
tokio::spawn(async move {
Self::monitor_disk_writable(disk, health, cancel_token).await;
});
}
/// Stop the disk monitoring
crates/ecstore/src/disk/mod.rs
+12
View File
@@ -412,6 +412,18 @@ impl DiskAPI for Disk {
}
}
impl Disk {
/// Enable health monitoring on this disk.
/// Called after startup format loading completes so that remote peers
/// have time to come online before being marked as faulty.
pub fn enable_health_check(&self) {
match self {
Disk::Local(local_disk) => local_disk.enable_health_check(),
Disk::Remote(remote_disk) => remote_disk.enable_health_check(),
}
}
}
pub async fn new_disk(ep: &Endpoint, opt: &DiskOption) -> Result<DiskStore> {
if ep.is_local {
let s = LocalDisk::new(ep, opt.cleanup).await?;
crates/ecstore/src/rpc/remote_disk.rs
+18 -18
View File
@@ -16,7 +16,8 @@ use crate::disk::{
CheckPartsResp, DeleteOptions, DiskAPI, DiskInfo, DiskInfoOptions, DiskLocation, DiskOption, FileInfoVersions, FileReader,
FileWriter, ReadMultipleReq, ReadMultipleResp, ReadOptions, RenameDataResp, UpdateMetadataOpts, VolumeInfo, WalkDirOptions,
disk_store::{
CHECK_EVERY, CHECK_TIMEOUT_DURATION, ENV_RUSTFS_DRIVE_ACTIVE_MONITORING, SKIP_IF_SUCCESS_BEFORE, get_max_timeout_duration,
CHECK_EVERY, CHECK_TIMEOUT_DURATION, DEFAULT_RUSTFS_DRIVE_ACTIVE_MONITORING, ENV_RUSTFS_DRIVE_ACTIVE_MONITORING,
SKIP_IF_SUCCESS_BEFORE, get_max_timeout_duration,
},
endpoint::Endpoint,
};
@@ -39,7 +40,6 @@ use rustfs_protos::proto_gen::node_service::{
node_service_client::NodeServiceClient,
};
use rustfs_rio::{HttpReader, HttpWriter};
use rustfs_utils::string::parse_bool_with_default;
use std::{
path::PathBuf,
sync::{
@@ -77,9 +77,8 @@ impl RemoteDisk {
format!("{}://{}", ep.url.scheme(), ep.url.host_str().unwrap())
};
let env_health_check = std::env::var(ENV_RUSTFS_DRIVE_ACTIVE_MONITORING)
.map(|v| parse_bool_with_default(&v, true))
.unwrap_or(true);
let env_health_check =
rustfs_utils::get_env_bool(ENV_RUSTFS_DRIVE_ACTIVE_MONITORING, DEFAULT_RUSTFS_DRIVE_ACTIVE_MONITORING);
let disk = Self {
id: Mutex::new(None),
@@ -91,23 +90,23 @@ impl RemoteDisk {
cancel_token: CancellationToken::new(),
};
// Start health monitoring
disk.start_health_monitoring();
Ok(disk)
}
/// Start health monitoring for the remote disk
fn start_health_monitoring(&self) {
if self.health_check {
let health = Arc::clone(&self.health);
let cancel_token = self.cancel_token.clone();
let addr = self.addr.clone();
tokio::spawn(async move {
Self::monitor_remote_disk_health(addr, health, cancel_token).await;
});
/// Enable health monitoring after disk creation.
/// Used to defer health checks until after startup format loading completes,
/// so that remote peers have time to come online.
pub fn enable_health_check(&self) {
if !self.health_check {
return;
}
let health = Arc::clone(&self.health);
let cancel_token = self.cancel_token.clone();
let addr = self.addr.clone();
tokio::spawn(async move {
Self::monitor_remote_disk_health(addr, health, cancel_token).await;
});
}
/// Monitor remote disk health periodically
@@ -1536,6 +1535,7 @@ mod tests {
};
let remote_disk = RemoteDisk::new(&endpoint, &disk_option).await.unwrap();
remote_disk.enable_health_check();
// wait for health check connect timeout
tokio::time::sleep(Duration::from_secs(6)).await;
crates/ecstore/src/store.rs
+9 -1
View File
@@ -202,11 +202,14 @@ impl ECStore {
// validate_parity(parity_count, pool_eps.drives_per_set)?;
// Initialize disks without health monitoring so that remote peers
// are not immediately marked as faulty before they have a chance to
// start up. Health monitoring is enabled after format loading succeeds.
let (disks, errs) = store_init::init_disks(
&pool_eps.endpoints,
&DiskOption {
cleanup: true,
health_check: true,
health_check: false,
},
)
.await;
@@ -250,6 +253,11 @@ impl ECStore {
}
}?;
// Format loading succeeded, enable health monitoring on all disks
for disk in disks.iter().flatten() {
disk.enable_health_check();
}
if deployment_id.is_none() {
deployment_id = Some(fm.id);
}