diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index e768908cc..f35e9c65b 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -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, 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 { 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 diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index 3c742dcd5..9c71acf24 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/crates/ecstore/src/disk/mod.rs @@ -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 { if ep.is_local { let s = LocalDisk::new(ep, opt.cleanup).await?; diff --git a/crates/ecstore/src/rpc/remote_disk.rs b/crates/ecstore/src/rpc/remote_disk.rs index 993d4fe47..eb17ea583 100644 --- a/crates/ecstore/src/rpc/remote_disk.rs +++ b/crates/ecstore/src/rpc/remote_disk.rs @@ -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; diff --git a/crates/ecstore/src/store.rs b/crates/ecstore/src/store.rs index 4f82909c7..9a419c62c 100644 --- a/crates/ecstore/src/store.rs +++ b/crates/ecstore/src/store.rs @@ -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); }