mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-07 04:25:54 +00:00
Compare commits
1 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| d59a2b5ac8 |
@@ -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
|
||||
|
||||
@@ -6463,16 +6463,43 @@ 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<DiskStore>], eps: &[Endpoint]) -> Vec<rustfs_madmin::Disk> {
|
||||
let mut ret = Vec::new();
|
||||
join_all(disks.iter().zip(eps).map(|(disk, ep)| disk_admin_info(disk.as_ref(), ep))).await
|
||||
}
|
||||
|
||||
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()
|
||||
};
|
||||
};
|
||||
|
||||
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 {
|
||||
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;
|
||||
}
|
||||
|
||||
match disk
|
||||
.disk_info(&DiskInfoOptions {
|
||||
metrics: true,
|
||||
@@ -6482,12 +6509,12 @@ async fn get_disks_info(disks: &[Option<DiskStore>], eps: &[Endpoint]) -> Vec<ru
|
||||
{
|
||||
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,
|
||||
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,
|
||||
@@ -6512,17 +6539,17 @@ async fn get_disks_info(disks: &[Option<DiskStore>], eps: &[Endpoint]) -> Vec<ru
|
||||
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,
|
||||
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(),
|
||||
@@ -6535,41 +6562,15 @@ async fn get_disks_info(disks: &[Option<DiskStore>], eps: &[Endpoint]) -> Vec<ru
|
||||
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));
|
||||
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);
|
||||
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()
|
||||
})
|
||||
}
|
||||
}
|
||||
|
||||
ret
|
||||
}
|
||||
|
||||
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::<Vec<_>>(),
|
||||
endpoints.iter().map(|ep| ep.disk_idx).collect::<Vec<_>>(),
|
||||
"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;
|
||||
|
||||
Reference in New Issue
Block a user