From 149929539339ed1a447b6bee2e1c7c65c885237c Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Mon, 7 Sep 2026 12:35:01 +0800 Subject: [PATCH] fix(admin): probe set drives concurrently for storage info (#7338) The admin storage walk probed a set's drives one after another, each bounded by the disk_info timeout, so a few drives still recovering after a power cut pushed the local snapshot past the peer probe budget and healthy peers rendered as unknown. Probe all drives at once so the walk costs one timeout at most, and add a test-only probe delay hook to pin that bound. --- crates/ecstore/src/disk/disk_store.rs | 11 ++ crates/ecstore/src/set_disk/mod.rs | 240 +++++++++++++++----------- 2 files changed, 149 insertions(+), 102 deletions(-) diff --git a/crates/ecstore/src/disk/disk_store.rs b/crates/ecstore/src/disk/disk_store.rs index 2035b9ee3..0d79ca559 100644 --- a/crates/ecstore/src/disk/disk_store.rs +++ b/crates/ecstore/src/disk/disk_store.rs @@ -195,6 +195,13 @@ fn resolve_drive_timeout_profile_from_env() -> DriveTimeoutProfile { DriveTimeoutProfile::parse(rustfs_config::DEFAULT_DRIVE_TIMEOUT_PROFILE).unwrap_or(DriveTimeoutProfile::Default) } +#[cfg(test)] +tokio::task_local! { + /// Artificial `disk_info` latency for tests that pin how the admin storage + /// walk composes per-drive probe time. + pub(crate) static DISK_INFO_PROBE_DELAY_FOR_TEST: Duration; +} + fn get_drive_timeout_profile() -> DriveTimeoutProfile { #[cfg(test)] { @@ -2036,6 +2043,10 @@ impl DiskAPI for LocalDiskWrapper { .track_disk_health_with_op_and_timeout_action( "disk_info", || async { + #[cfg(test)] + if let Ok(delay) = DISK_INFO_PROBE_DELAY_FOR_TEST.try_with(|delay| *delay) { + tokio::time::sleep(delay).await; + } let result = self.disk.disk_info(opts).await?; if let Some(current_disk_id) = *self.disk_id.read().await diff --git a/crates/ecstore/src/set_disk/mod.rs b/crates/ecstore/src/set_disk/mod.rs index 71d885e97..e79599936 100644 --- a/crates/ecstore/src/set_disk/mod.rs +++ b/crates/ecstore/src/set_disk/mod.rs @@ -6463,113 +6463,114 @@ pub fn should_heal_object_on_disk( (false, false, None) } +/// Probe every drive of the set at once. Each live probe is bounded by the +/// drive `disk_info` timeout, and the admin peer probe budget only covers one +/// such timeout; a sequential walk over several stalled drives after a power +/// cut would exceed it and make healthy peers render as unknown (#6488). async fn get_disks_info(disks: &[Option], eps: &[Endpoint]) -> Vec { - let mut ret = Vec::new(); + join_all(disks.iter().zip(eps).map(|(disk, ep)| disk_admin_info(disk.as_ref(), ep))).await +} - for (i, pool) in disks.iter().enumerate() { - if let Some(disk) = pool { - let runtime_state = disk.runtime_state(); - let offline_duration_seconds = disk.offline_duration_secs(); - let capacity_snapshot = disk.last_capacity_snapshot(); - let cached_disk_id = disk.cached_disk_id().await; - if runtime_state.should_probe_for_admin() || runtime_state == disk::health_state::RuntimeDriveHealthState::Suspect { - match disk - .disk_info(&DiskInfoOptions { - metrics: true, - ..Default::default() - }) - .await - { - Ok(res) => { - disk.record_capacity_probe(res.total, res.used, res.free); - ret.push(rustfs_madmin::Disk { - endpoint: eps[i].to_string(), - local: eps[i].is_local, - pool_index: eps[i].pool_idx, - set_index: eps[i].set_idx, - disk_index: eps[i].disk_idx, - state: "ok".to_owned(), +async fn disk_admin_info(disk: Option<&DiskStore>, ep: &Endpoint) -> rustfs_madmin::Disk { + let Some(disk) = disk else { + return rustfs_madmin::Disk { + endpoint: ep.to_string(), + drive_path: ep.get_file_path(), + local: ep.is_local, + pool_index: ep.pool_idx, + set_index: ep.set_idx, + disk_index: ep.disk_idx, + runtime_state: None, + offline_duration_seconds: None, + state: DiskError::DiskNotFound.to_string(), + capacity_observation_source: Some("missing".to_owned()), + capacity_observation_age_seconds: Some(0), + ..Default::default() + }; + }; - root_disk: res.root_disk, - drive_path: res.mount_path.clone(), - healing: res.healing, - scanning: res.scanning, - runtime_state: Some(runtime_state.as_str().to_string()), - offline_duration_seconds, - capacity_observation_source: Some("live_probe".to_owned()), - capacity_observation_age_seconds: Some(0), - - uuid: res.id.map_or_else(|| "".to_string(), |id| id.to_string()), - major: res.major as u32, - minor: res.minor as u32, - model: None, - total_space: res.total, - used_space: res.used, - available_space: res.free, - physical_device_ids: (!res.physical_device_ids.is_empty()).then_some(res.physical_device_ids.clone()), - utilization: utilization_percent(res.total, res.used), - used_inodes: res.used_inodes, - free_inodes: res.free_inodes, - metrics: Some(res.metrics), - ..Default::default() - }); - } - Err(err) => { - let mut disk_info = rustfs_madmin::Disk { - state: err.to_string(), - endpoint: eps[i].to_string(), - drive_path: eps[i].get_file_path(), - local: eps[i].is_local, - pool_index: eps[i].pool_idx, - set_index: eps[i].set_idx, - disk_index: eps[i].disk_idx, - runtime_state: Some(runtime_state.as_str().to_string()), - offline_duration_seconds, - metrics: disk.metrics_snapshot(), - uuid: cached_disk_id.map_or_else(String::new, |id| id.to_string()), - ..Default::default() - }; - if let Some((total, used, free, _)) = capacity_snapshot { - disk_info.total_space = total; - disk_info.used_space = used; - disk_info.available_space = free; - disk_info.utilization = utilization_percent(total, used); - disk_info.capacity_observation_source = Some("snapshot".to_owned()); - disk_info.capacity_observation_age_seconds = capacity_snapshot - .map(|(_, _, _, probe_unix_secs)| capacity_snapshot_age_seconds(probe_unix_secs)); - } else { - disk_info.capacity_observation_source = Some("missing".to_owned()); - disk_info.capacity_observation_age_seconds = Some(0); - } - ret.push(disk_info); - } - } - } else { - let mut disk_info = - build_runtime_snapshot_disk(&eps[i], runtime_state, offline_duration_seconds, capacity_snapshot); - disk_info.metrics = disk.metrics_snapshot(); - disk_info.uuid = cached_disk_id.map_or_else(String::new, |id| id.to_string()); - ret.push(disk_info); - } - } else { - ret.push(rustfs_madmin::Disk { - endpoint: eps[i].to_string(), - drive_path: eps[i].get_file_path(), - local: eps[i].is_local, - pool_index: eps[i].pool_idx, - set_index: eps[i].set_idx, - disk_index: eps[i].disk_idx, - runtime_state: None, - offline_duration_seconds: None, - state: DiskError::DiskNotFound.to_string(), - capacity_observation_source: Some("missing".to_owned()), - capacity_observation_age_seconds: Some(0), - ..Default::default() - }) - } + let runtime_state = disk.runtime_state(); + let offline_duration_seconds = disk.offline_duration_secs(); + let capacity_snapshot = disk.last_capacity_snapshot(); + let cached_disk_id = disk.cached_disk_id().await; + if !(runtime_state.should_probe_for_admin() || runtime_state == disk::health_state::RuntimeDriveHealthState::Suspect) { + let mut disk_info = build_runtime_snapshot_disk(ep, runtime_state, offline_duration_seconds, capacity_snapshot); + disk_info.metrics = disk.metrics_snapshot(); + disk_info.uuid = cached_disk_id.map_or_else(String::new, |id| id.to_string()); + return disk_info; } - ret + match disk + .disk_info(&DiskInfoOptions { + metrics: true, + ..Default::default() + }) + .await + { + Ok(res) => { + disk.record_capacity_probe(res.total, res.used, res.free); + rustfs_madmin::Disk { + endpoint: ep.to_string(), + local: ep.is_local, + pool_index: ep.pool_idx, + set_index: ep.set_idx, + disk_index: ep.disk_idx, + state: "ok".to_owned(), + + root_disk: res.root_disk, + drive_path: res.mount_path.clone(), + healing: res.healing, + scanning: res.scanning, + runtime_state: Some(runtime_state.as_str().to_string()), + offline_duration_seconds, + capacity_observation_source: Some("live_probe".to_owned()), + capacity_observation_age_seconds: Some(0), + + uuid: res.id.map_or_else(|| "".to_string(), |id| id.to_string()), + major: res.major as u32, + minor: res.minor as u32, + model: None, + total_space: res.total, + used_space: res.used, + available_space: res.free, + physical_device_ids: (!res.physical_device_ids.is_empty()).then_some(res.physical_device_ids.clone()), + utilization: utilization_percent(res.total, res.used), + used_inodes: res.used_inodes, + free_inodes: res.free_inodes, + metrics: Some(res.metrics), + ..Default::default() + } + } + Err(err) => { + let mut disk_info = rustfs_madmin::Disk { + state: err.to_string(), + endpoint: ep.to_string(), + drive_path: ep.get_file_path(), + local: ep.is_local, + pool_index: ep.pool_idx, + set_index: ep.set_idx, + disk_index: ep.disk_idx, + runtime_state: Some(runtime_state.as_str().to_string()), + offline_duration_seconds, + metrics: disk.metrics_snapshot(), + uuid: cached_disk_id.map_or_else(String::new, |id| id.to_string()), + ..Default::default() + }; + if let Some((total, used, free, _)) = capacity_snapshot { + disk_info.total_space = total; + disk_info.used_space = used; + disk_info.available_space = free; + disk_info.utilization = utilization_percent(total, used); + disk_info.capacity_observation_source = Some("snapshot".to_owned()); + disk_info.capacity_observation_age_seconds = + capacity_snapshot.map(|(_, _, _, probe_unix_secs)| capacity_snapshot_age_seconds(probe_unix_secs)); + } else { + disk_info.capacity_observation_source = Some("missing".to_owned()); + disk_info.capacity_observation_age_seconds = Some(0); + } + disk_info + } + } } fn build_runtime_snapshot_disk( @@ -10691,6 +10692,41 @@ mod tests { ); } + #[tokio::test(start_paused = true)] + async fn test_get_disks_info_probes_drives_concurrently() { + use crate::disk::disk_store::DISK_INFO_PROBE_DELAY_FOR_TEST; + + let format = FormatV3::new(1, 4); + let mut temp_dirs = Vec::new(); + let mut endpoints = Vec::new(); + let mut disks = Vec::new(); + for disk_idx in 0..4 { + let (dir, endpoint, disk) = make_formatted_local_disk_for_info_test(disk_idx, &format).await; + temp_dirs.push(dir); + endpoints.push(endpoint); + disks.push(Some(disk)); + } + + let probe_delay = std::time::Duration::from_secs(2); + let started = tokio::time::Instant::now(); + let info = DISK_INFO_PROBE_DELAY_FOR_TEST + .scope(probe_delay, get_disks_info(&disks, &endpoints)) + .await; + let elapsed = started.elapsed(); + + assert_eq!(info.len(), 4); + assert!(info.iter().all(|disk| disk.state == "ok"), "every drive should still report a live probe"); + assert_eq!( + info.iter().map(|disk| disk.disk_index).collect::>(), + endpoints.iter().map(|ep| ep.disk_idx).collect::>(), + "concurrent probes must keep endpoint order" + ); + assert!( + elapsed < probe_delay * 2, + "four stalled drives must cost one probe delay, not four; took {elapsed:?}" + ); + } + #[tokio::test] async fn test_get_disks_info_preserves_remote_cached_disk_id_when_offline() { let (endpoint, disk) = make_remote_disk_for_info_test(0).await;