mirror of
https://github.com/rustfs/rustfs.git
synced 2026-09-07 04:25:54 +00:00
Compare commits
3 Commits
| Author | SHA1 | Date | |
|---|---|---|---|
| 022f1ecce5 | |||
| 672087ec0d | |||
| 9dbeae1b45 |
@@ -121,6 +121,22 @@ jobs:
|
||||
create_latest=false
|
||||
source_ref="$GITHUB_SHA"
|
||||
|
||||
# Pre-GA policy: until the first stable (vX.Y.Z) tag exists, every
|
||||
# prerelease (alpha/beta/rc) also moves `latest`, so users pulling
|
||||
# `latest` get the newest test build. Once a stable tag is published
|
||||
# this returns false and `latest` follows stable releases only.
|
||||
prerelease_moves_latest() {
|
||||
local stable_tags
|
||||
stable_tags=$(git ls-remote --tags --refs origin 2>/dev/null \
|
||||
| awk '{print $2}' \
|
||||
| grep -E '^refs/tags/v?[0-9]+\.[0-9]+\.[0-9]+$' || true)
|
||||
if [[ -z "$stable_tags" ]]; then
|
||||
return 0
|
||||
fi
|
||||
echo "ℹ️ Stable release tag(s) already exist; prereleases no longer update latest"
|
||||
return 1
|
||||
}
|
||||
|
||||
if [[ "${{ github.event_name }}" == "workflow_run" ]]; then
|
||||
# Triggered by build workflow completion
|
||||
echo "🔗 Triggered by build workflow completion"
|
||||
@@ -184,8 +200,8 @@ jobs:
|
||||
if [[ "$version" == *"alpha"* ]] || [[ "$version" == *"beta"* ]] || [[ "$version" == *"rc"* ]]; then
|
||||
build_type="prerelease"
|
||||
is_prerelease=true
|
||||
# Current policy: create latest tags for stable releases and selected prereleases (alpha/beta).
|
||||
if [[ "$version" == *"alpha"* ]] || [[ "$version" == *"beta"* ]]; then
|
||||
# Pre-GA policy: prereleases update latest until the first stable tag exists.
|
||||
if prerelease_moves_latest; then
|
||||
create_latest=true
|
||||
echo "🧪 Building Docker image for prerelease: $version (creating latest tag)"
|
||||
else
|
||||
@@ -243,8 +259,8 @@ jobs:
|
||||
v*alpha*|v*beta*|v*rc*|*alpha*|*beta*|*rc*)
|
||||
build_type="prerelease"
|
||||
is_prerelease=true
|
||||
# Current policy: create latest tags for stable releases and selected prereleases (alpha/beta).
|
||||
if [[ "$version" == *"alpha"* ]] || [[ "$version" == *"beta"* ]]; then
|
||||
# Pre-GA policy: prereleases update latest until the first stable tag exists.
|
||||
if prerelease_moves_latest; then
|
||||
create_latest=true
|
||||
echo "🧪 Building with prerelease version: $input_version (creating latest tag)"
|
||||
else
|
||||
@@ -394,11 +410,13 @@ jobs:
|
||||
TAG_BASE="${VERSION}${VARIANT_SUFFIX}"
|
||||
TAGS="${{ env.REGISTRY_DOCKERHUB }}:$TAG_BASE,${{ env.REGISTRY_GHCR }}:$TAG_BASE,${{ env.REGISTRY_QUAY }}:$TAG_BASE"
|
||||
|
||||
# Add channel tags for prereleases and latest for stable
|
||||
# Add latest when requested (stable releases, and prereleases before GA)
|
||||
if [[ "$CREATE_LATEST" == "true" ]]; then
|
||||
# Create latest tags for stable releases and selected prereleases when CREATE_LATEST=true.
|
||||
TAGS="$TAGS,${{ env.REGISTRY_DOCKERHUB }}:latest${VARIANT_SUFFIX},${{ env.REGISTRY_GHCR }}:latest${VARIANT_SUFFIX},${{ env.REGISTRY_QUAY }}:latest${VARIANT_SUFFIX}"
|
||||
elif [[ "$BUILD_TYPE" == "prerelease" ]]; then
|
||||
fi
|
||||
|
||||
# Always add the channel tag for prereleases, independent of latest
|
||||
if [[ "$BUILD_TYPE" == "prerelease" ]]; then
|
||||
# Prerelease channel tags (alpha, beta, rc)
|
||||
if [[ "$VERSION" == *"alpha"* ]]; then
|
||||
CHANNEL="alpha"
|
||||
@@ -555,7 +573,7 @@ jobs:
|
||||
"prerelease")
|
||||
echo "🧪 Prerelease Docker image has been built with ${VERSION} tags"
|
||||
echo "⚠️ This is a prerelease image - use with caution"
|
||||
# Create latest tags for stable releases and selected prereleases when CREATE_LATEST=true.
|
||||
# Prereleases move latest until the first stable tag exists (pre-GA policy).
|
||||
if [[ "$CREATE_LATEST" == "true" ]]; then
|
||||
echo "🏷️ Latest tag has been created for prerelease: $VERSION"
|
||||
else
|
||||
|
||||
Generated
+5
-5
@@ -6139,9 +6139,9 @@ checksum = "3eaf3ede3fee6db1a4c2ee091bf8a8b4dccdc6d17f656fb07896ee72867612f2"
|
||||
|
||||
[[package]]
|
||||
name = "libflate"
|
||||
version = "2.3.1"
|
||||
version = "2.3.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "a4da9b700e758e57152a1fd1c52cbdc5727c1aa6d8743dc1acda917398f1d76c"
|
||||
checksum = "561a8da1a50e1428d3c51321dafeca849df992a5bb67720c386131234caba82e"
|
||||
dependencies = [
|
||||
"adler32",
|
||||
"crc32fast",
|
||||
@@ -10943,9 +10943,9 @@ dependencies = [
|
||||
|
||||
[[package]]
|
||||
name = "rustfs-uring"
|
||||
version = "0.2.1"
|
||||
version = "0.2.2"
|
||||
source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "0486e62d0efe25db95c00aeacb2da84368adcba299216cda99fcb11328061c84"
|
||||
checksum = "b29bc57b4bd62a73f4fae408b536adf578332e50e464797d09dc2382c7cb68c2"
|
||||
dependencies = [
|
||||
"io-uring",
|
||||
"libc",
|
||||
@@ -12406,7 +12406,7 @@ source = "registry+https://github.com/rust-lang/crates.io-index"
|
||||
checksum = "32497e9a4c7b38532efcdebeef879707aa9f794296a4f0244f6f69e9bc8574bd"
|
||||
dependencies = [
|
||||
"fastrand",
|
||||
"getrandom 0.3.4",
|
||||
"getrandom 0.4.3",
|
||||
"once_cell",
|
||||
"rustix",
|
||||
"windows-sys 0.59.0",
|
||||
|
||||
@@ -226,7 +226,7 @@ metrics = { workspace = true }
|
||||
# crates.io. The guard scripts/check_no_tokio_io_uring.sh allows an explicit
|
||||
# io-uring integration; only the tokio "io-uring" runtime feature is banned.
|
||||
[target.'cfg(target_os = "linux")'.dependencies]
|
||||
rustfs-uring = "0.2.1"
|
||||
rustfs-uring = "0.2.2"
|
||||
|
||||
[target.'cfg(windows)'.dependencies]
|
||||
winapi-util.workspace = true
|
||||
|
||||
@@ -195,13 +195,6 @@ 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)]
|
||||
{
|
||||
@@ -2043,10 +2036,6 @@ 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
|
||||
|
||||
@@ -8708,6 +8708,34 @@ impl DiskAPI for LocalDisk {
|
||||
});
|
||||
}
|
||||
|
||||
let rollback_after_fsync_failure = |parent: &Path, current: Option<&[u8]>| -> std::io::Result<()> {
|
||||
match current {
|
||||
Some(previous) => {
|
||||
let rollback_temporary =
|
||||
parent.join(format!(".{}.{}.rollback.tmp", path.replace('/', "_"), Uuid::new_v4()));
|
||||
let rollback_result = (|| -> std::io::Result<()> {
|
||||
let mut staged = std::fs::OpenOptions::new()
|
||||
.create_new(true)
|
||||
.write(true)
|
||||
.open(&rollback_temporary)?;
|
||||
staged.write_all(previous)?;
|
||||
staged.sync_all()?;
|
||||
std::fs::rename(&rollback_temporary, &file_path)
|
||||
})();
|
||||
if let Err(err) = rollback_result {
|
||||
let _ = std::fs::remove_file(&rollback_temporary);
|
||||
return Err(err);
|
||||
}
|
||||
}
|
||||
None => match std::fs::remove_file(&file_path) {
|
||||
Ok(()) => {}
|
||||
Err(err) if err.kind() == ErrorKind::NotFound => {}
|
||||
Err(err) => return Err(err),
|
||||
},
|
||||
}
|
||||
os::fsync_dir_std(parent)
|
||||
};
|
||||
|
||||
match replacement {
|
||||
Some(replacement) => {
|
||||
let parent = file_path
|
||||
@@ -8727,14 +8755,19 @@ impl DiskAPI for LocalDisk {
|
||||
let _ = std::fs::remove_file(&temporary);
|
||||
return Err(err);
|
||||
}
|
||||
if sync_metadata {
|
||||
os::fsync_dir_std(parent)?;
|
||||
if sync_metadata && let Err(err) = os::fsync_dir_std(parent) {
|
||||
rollback_after_fsync_failure(parent, current.as_deref())?;
|
||||
return Err(err);
|
||||
}
|
||||
}
|
||||
None => {
|
||||
std::fs::remove_file(&file_path)?;
|
||||
if sync_metadata && let Some(parent) = file_path.parent() {
|
||||
os::fsync_dir_std(parent)?;
|
||||
if sync_metadata
|
||||
&& let Some(parent) = file_path.parent()
|
||||
&& let Err(err) = os::fsync_dir_std(parent)
|
||||
{
|
||||
rollback_after_fsync_failure(parent, current.as_deref())?;
|
||||
return Err(err);
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -22166,6 +22199,110 @@ mod test {
|
||||
));
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn conditional_file_update_dir_fsync_failure_restores_previous_bytes() {
|
||||
use tempfile::tempdir;
|
||||
|
||||
let dir = tempdir().expect("temp dir should be created");
|
||||
let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse");
|
||||
let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
|
||||
let previous = Bytes::from_static(b"previous-owner");
|
||||
let successor = Bytes::from_static(b"successor-owner");
|
||||
|
||||
assert_eq!(
|
||||
disk.compare_and_update_file(RUSTFS_META_BUCKET, HEALING_MARKER_PATH, None, Some(previous.clone()))
|
||||
.await
|
||||
.expect("previous owner should commit"),
|
||||
ConditionalFileUpdate::Updated
|
||||
);
|
||||
|
||||
let marker_path = disk
|
||||
.get_object_path(RUSTFS_META_BUCKET, HEALING_MARKER_PATH)
|
||||
.expect("marker path should resolve");
|
||||
let parent = marker_path.parent().expect("marker path should have a parent");
|
||||
os::fsync_dir_recorder::set_failure(parent, ErrorKind::Other);
|
||||
|
||||
let err = disk
|
||||
.compare_and_update_file(RUSTFS_META_BUCKET, HEALING_MARKER_PATH, Some(previous.clone()), Some(successor))
|
||||
.await
|
||||
.expect_err("directory fsync failure must fail the CAS update");
|
||||
assert!(matches!(err, DiskError::Io(ref err) if err.kind() == ErrorKind::Other));
|
||||
assert_eq!(
|
||||
disk.read_all(RUSTFS_META_BUCKET, HEALING_MARKER_PATH)
|
||||
.await
|
||||
.expect("previous bytes should remain readable after rollback"),
|
||||
previous
|
||||
);
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn conditional_file_update_dir_fsync_failure_removes_new_file_without_anchor() {
|
||||
use tempfile::tempdir;
|
||||
|
||||
let dir = tempdir().expect("temp dir should be created");
|
||||
let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse");
|
||||
let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
|
||||
ensure_test_volume(&disk, RUSTFS_META_BUCKET).await;
|
||||
let marker_path = disk
|
||||
.get_object_path(RUSTFS_META_BUCKET, HEALING_MARKER_PATH)
|
||||
.expect("marker path should resolve");
|
||||
let parent = marker_path.parent().expect("marker path should have a parent");
|
||||
os::fsync_dir_recorder::set_failure(parent, ErrorKind::Other);
|
||||
|
||||
let err = disk
|
||||
.compare_and_update_file(
|
||||
RUSTFS_META_BUCKET,
|
||||
HEALING_MARKER_PATH,
|
||||
None,
|
||||
Some(Bytes::from_static(b"successor-owner")),
|
||||
)
|
||||
.await
|
||||
.expect_err("directory fsync failure must fail the CAS create");
|
||||
assert!(matches!(err, DiskError::Io(ref err) if err.kind() == ErrorKind::Other));
|
||||
assert!(
|
||||
matches!(disk.read_all(RUSTFS_META_BUCKET, HEALING_MARKER_PATH).await, Err(DiskError::FileNotFound)),
|
||||
"uncommitted successor bytes must be removed when no previous anchor exists"
|
||||
);
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn conditional_file_delete_dir_fsync_failure_restores_previous_bytes() {
|
||||
use tempfile::tempdir;
|
||||
|
||||
let dir = tempdir().expect("temp dir should be created");
|
||||
let endpoint = Endpoint::try_from(dir.path().to_str().expect("temp dir should be utf8")).expect("endpoint should parse");
|
||||
let disk = LocalDisk::new(&endpoint, false).await.expect("local disk should be created");
|
||||
let previous = Bytes::from_static(b"previous-owner");
|
||||
|
||||
assert_eq!(
|
||||
disk.compare_and_update_file(RUSTFS_META_BUCKET, HEALING_MARKER_PATH, None, Some(previous.clone()))
|
||||
.await
|
||||
.expect("previous owner should commit"),
|
||||
ConditionalFileUpdate::Updated
|
||||
);
|
||||
|
||||
let marker_path = disk
|
||||
.get_object_path(RUSTFS_META_BUCKET, HEALING_MARKER_PATH)
|
||||
.expect("marker path should resolve");
|
||||
let parent = marker_path.parent().expect("marker path should have a parent");
|
||||
os::fsync_dir_recorder::set_failure(parent, ErrorKind::Other);
|
||||
|
||||
let err = disk
|
||||
.compare_and_update_file(RUSTFS_META_BUCKET, HEALING_MARKER_PATH, Some(previous.clone()), None)
|
||||
.await
|
||||
.expect_err("directory fsync failure must fail the CAS delete");
|
||||
assert!(matches!(err, DiskError::Io(ref err) if err.kind() == ErrorKind::Other));
|
||||
assert_eq!(
|
||||
disk.read_all(RUSTFS_META_BUCKET, HEALING_MARKER_PATH)
|
||||
.await
|
||||
.expect("previous bytes should be restored after failed delete"),
|
||||
previous
|
||||
);
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
#[tokio::test]
|
||||
async fn conditional_file_update_returns_would_block_when_marker_lock_is_contended() {
|
||||
|
||||
@@ -92,6 +92,9 @@ pub(crate) mod fsync_dir_recorder {
|
||||
static LIMITED: Mutex<Vec<PathBuf>> = Mutex::new(Vec::new());
|
||||
static GROUPED: Mutex<Vec<(PathBuf, usize)>> = Mutex::new(Vec::new());
|
||||
#[cfg(unix)]
|
||||
static FAILURES: std::sync::LazyLock<Mutex<HashMap<PathBuf, io::ErrorKind>>> =
|
||||
std::sync::LazyLock::new(|| Mutex::new(HashMap::new()));
|
||||
#[cfg(unix)]
|
||||
static BEFORE_LIMITED: std::sync::LazyLock<Mutex<HashMap<PathBuf, Hook>>> =
|
||||
std::sync::LazyLock::new(|| Mutex::new(HashMap::new()));
|
||||
static BEFORE_GROUP_BATCH: std::sync::LazyLock<Mutex<HashMap<PathBuf, Hook>>> =
|
||||
@@ -151,6 +154,19 @@ pub(crate) mod fsync_dir_recorder {
|
||||
contains_path(&RECORDED.lock().expect("fsync dir recorder poisoned"), dir)
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
pub(crate) fn set_failure(dir: &Path, kind: io::ErrorKind) {
|
||||
FAILURES
|
||||
.lock()
|
||||
.expect("fsync dir failure hook poisoned")
|
||||
.insert(dir.to_path_buf(), kind);
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
pub(crate) fn take_failure(dir: &Path) -> Option<io::ErrorKind> {
|
||||
remove_path_keyed(&FAILURES, dir, "fsync dir failure hook poisoned")
|
||||
}
|
||||
|
||||
#[cfg(unix)]
|
||||
pub(crate) fn record_limited(dir: &Path) {
|
||||
record_path(&LIMITED, dir, "limited fsync dir recorder");
|
||||
@@ -426,6 +442,10 @@ pub fn fsync_dir_std(dir: impl AsRef<Path>) -> io::Result<()> {
|
||||
fsync_dir_recorder::record(dir.as_ref());
|
||||
#[cfg(unix)]
|
||||
{
|
||||
#[cfg(test)]
|
||||
if let Some(kind) = fsync_dir_recorder::take_failure(dir.as_ref()) {
|
||||
return Err(io::Error::from(kind));
|
||||
}
|
||||
std::fs::File::open(dir.as_ref())?.sync_all()?;
|
||||
}
|
||||
#[cfg(not(unix))]
|
||||
|
||||
@@ -6463,114 +6463,113 @@ 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> {
|
||||
join_all(disks.iter().zip(eps).map(|(disk, ep)| disk_admin_info(disk.as_ref(), ep))).await
|
||||
}
|
||||
let mut ret = Vec::new();
|
||||
|
||||
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 {
|
||||
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(),
|
||||
|
||||
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;
|
||||
}
|
||||
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),
|
||||
|
||||
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));
|
||||
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 {
|
||||
disk_info.capacity_observation_source = Some("missing".to_owned());
|
||||
disk_info.capacity_observation_age_seconds = Some(0);
|
||||
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);
|
||||
}
|
||||
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(
|
||||
@@ -10692,41 +10691,6 @@ 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