lywsvip/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-05-06 22:28:16 +08:00
Code Issues Packages Projects Releases Wiki Activity

fix: deduplicate disks in capacity calculation to prevent inflation (#1656)

Browse Source
This commit is contained in:
houseme
2026-01-30 00:03:21 +08:00
committed by GitHub
parent 022e3dfc21
commit 2ee81496b0
17 changed files with 631 additions and 275 deletions
Download Patch File Download Diff File Expand all files Collapse all files

21
Cargo.lock generated
View File

@@ -5683,18 +5683,6 @@ dependencies = [
"memoffset",
]
[[package]]
name = "nix"
version = "0.31.1"
source = "registry+https://github.com/rust-lang/crates.io-index"
checksum = "225e7cfe711e0ba79a68baeddb2982723e4235247aefce1482f2f16c27865b66"
dependencies = [
"bitflags 2.10.0",
"cfg-if",
"cfg_aliases",
"libc",
]
[[package]]
name = "nom"
version = "7.1.3"
@@ -7727,7 +7715,6 @@ dependencies = [
"base64",
"base64-simd",
"bytes",
"chrono",
"clap",
"const-str",
"datafusion",
@@ -8325,7 +8312,6 @@ dependencies = [
name = "rustfs-scanner"
version = "0.0.5"
dependencies = [
"anyhow",
"async-trait",
"chrono",
"futures",
@@ -8342,16 +8328,11 @@ dependencies = [
"s3s",
"serde",
"serde_json",
"serial_test",
"tempfile",
"thiserror 2.0.18",
"time",
"tokio",
"tokio-test",
"tokio-util",
"tracing",
"tracing-subscriber",
"uuid",
]
[[package]]
@@ -8412,10 +8393,10 @@ dependencies = [
"lz4",
"md-5 0.11.0-rc.3",
"netif",
"nix 0.31.1",
"rand 0.10.0-rc.6",
"regex",
"rustfs-config",
"rustix 1.1.3",
"rustls",
"rustls-pemfile",
"rustls-pki-types",

2
Cargo.toml
View File

@@ -210,7 +210,6 @@ md5 = "0.8.0"
mime_guess = "2.0.5"
moka = { version = "0.12.13", features = ["future"] }
netif = "0.1.6"
nix = { version = "0.31.1", features = ["fs"] }
nu-ansi-term = "0.50.3"
num_cpus = { version = "1.17.0" }
nvml-wrapper = "0.11.0"
@@ -225,6 +224,7 @@ rayon = "1.11.0"
reed-solomon-simd = { version = "3.1.0" }
regex = { version = "1.12.2" }
rumqttc = { version = "0.25.1" }
rustix = { version = "1.1.3", features = ["fs"] }
rust-embed = { version = "8.11.0" }
rustc-hash = { version = "2.1.1" }
s3s = { version = "0.13.0-alpha.2", features = ["minio"], git = "https://github.com/s3s-project/s3s.git", branch = "main" }

10
crates/ecstore/src/data_usage.rs
View File

@@ -28,7 +28,7 @@ use rustfs_common::data_usage::{
};
use rustfs_utils::path::SLASH_SEPARATOR;
use std::{
collections::{HashMap, hash_map::Entry},
collections::{HashMap, HashSet, hash_map::Entry},
sync::{Arc, OnceLock},
time::{Duration, SystemTime},
};
@@ -223,6 +223,7 @@ pub async fn aggregate_local_snapshots(store: Arc<ECStore>) -> Result<(Vec<DiskU
let mut aggregated = DataUsageInfo::default();
let mut latest_update: Option<SystemTime> = None;
let mut statuses: Vec<DiskUsageStatus> = Vec::new();
let mut processed_disks: HashSet<String> = HashSet::new();
for (pool_idx, pool) in store.pools.iter().enumerate() {
for set_disks in pool.disk_set.iter() {
@@ -246,6 +247,13 @@ pub async fn aggregate_local_snapshots(store: Arc<ECStore>) -> Result<(Vec<DiskU
};
let root = disk.path();
let disk_key = format!("{}|{}", disk.endpoint(), root.display());
// Skip if we've already processed this physical disk
if !processed_disks.insert(disk_key.clone()) {
continue;
}
let mut status = DiskUsageStatus {
disk_id: disk_id.clone(),
pool_index: Some(pool_idx),

4
crates/ecstore/src/lib.rs
View File

@@ -47,6 +47,10 @@ pub mod store_utils;
pub mod client;
pub mod event;
pub mod event_notification;
#[cfg(test)]
mod pools_test;
#[cfg(test)]
mod store_test;
pub mod tier;
pub use global::new_object_layer_fn;

112
crates/ecstore/src/pools.rs
View File

@@ -41,7 +41,7 @@ use rustfs_rio::{HashReader, WarpReader};
use rustfs_utils::path::{SLASH_SEPARATOR, encode_dir_object, path_join};
use rustfs_workers::workers::Workers;
use serde::{Deserialize, Serialize};
use std::collections::HashMap;
use std::collections::{HashMap, HashSet};
use std::fmt::Display;
use std::io::{Cursor, Write};
use std::path::PathBuf;
@@ -49,7 +49,7 @@ use std::sync::Arc;
use time::{Duration, OffsetDateTime};
use tokio::io::{AsyncReadExt, BufReader};
use tokio_util::sync::CancellationToken;
use tracing::{error, info, warn};
use tracing::{debug, error, info, warn};
pub const POOL_META_NAME: &str = "pool.bin";
pub const POOL_META_FORMAT: u16 = 1;
@@ -1404,30 +1404,26 @@ fn is_disk_online_state(state: &str) -> bool {
true
}
#[deprecated(since = "0.1.0", note = "Use fallback_total_capacity_dedup instead")]
#[allow(dead_code)]
fn fallback_total_capacity(disks: &[rustfs_madmin::Disk]) -> usize {
disks
.iter()
.filter(|d| is_disk_online_state(&d.state))
.map(|d| d.total_space as usize)
.sum()
fallback_total_capacity_dedup(disks)
}
#[deprecated(since = "0.1.0", note = "Use fallback_free_capacity_dedup instead")]
#[allow(dead_code)]
fn fallback_free_capacity(disks: &[rustfs_madmin::Disk]) -> usize {
disks
.iter()
.filter(|d| is_disk_online_state(&d.state))
.map(|d| d.available_space as usize)
.sum()
fallback_free_capacity_dedup(disks)
}
pub fn get_total_usable_capacity(disks: &[rustfs_madmin::Disk], info: &rustfs_madmin::StorageInfo) -> usize {
// If backend info is missing or inconsistent, do a safe fallback to avoid reporting nonsense.
if info.backend.standard_sc_data.is_empty() {
return fallback_total_capacity(disks);
return fallback_total_capacity_dedup(disks);
}
let mut capacity = 0usize;
let mut matched_any = false;
let mut counted_disks: HashSet<String> = HashSet::new();
for disk in disks.iter() {
if disk.pool_index < 0 {
@@ -1444,8 +1440,27 @@ pub fn get_total_usable_capacity(disks: &[rustfs_madmin::Disk], info: &rustfs_ma
}
if (disk.disk_index as usize) < usable_disks_per_set {
matched_any = true;
capacity += disk.total_space as usize;
// 🔧 Generate a unique identity using a combination of fields
let disk_key = format!(
"{}|{}|p{}s{}d{}",
disk.endpoint, // Node address
disk.drive_path, // mount path
disk.pool_index, // Pool index
disk.set_index, // Collection index
disk.disk_index // Disk index
);
debug!("get_total_usable_capacity disk_key: {}", disk_key);
// 🔧 Only disks that have not been counted are counted towards capacity
if counted_disks.insert(disk_key) {
matched_any = true;
capacity += disk.total_space as usize;
} else {
// Log duplicate disks: this likely indicates a configuration issue and should always be visible.
warn!(
"Duplicate disk detected in capacity calculation: {} at {}",
disk.endpoint, disk.drive_path
);
}
}
}
@@ -1454,17 +1469,18 @@ pub fn get_total_usable_capacity(disks: &[rustfs_madmin::Disk], info: &rustfs_ma
} else {
// Even if standard_sc_data exists, it might not match disk indexes due to upstream bugs.
// Fallback to summing all online disks to prevent under-reporting.
fallback_total_capacity(disks)
fallback_total_capacity_dedup(disks)
}
}
pub fn get_total_usable_capacity_free(disks: &[rustfs_madmin::Disk], info: &rustfs_madmin::StorageInfo) -> usize {
if info.backend.standard_sc_data.is_empty() {
return fallback_free_capacity(disks);
return fallback_free_capacity_dedup(disks);
}
let mut capacity = 0usize;
let mut matched_any = false;
let mut counted_disks: HashSet<String> = HashSet::new();
for disk in disks.iter() {
if disk.pool_index < 0 {
@@ -1481,14 +1497,68 @@ pub fn get_total_usable_capacity_free(disks: &[rustfs_madmin::Disk], info: &rust
}
if (disk.disk_index as usize) < usable_disks_per_set {
matched_any = true;
capacity += disk.available_space as usize;
let disk_key = format!(
"{}|{}|p{}s{}d{}",
disk.endpoint, disk.drive_path, disk.pool_index, disk.set_index, disk.disk_index
);
if counted_disks.insert(disk_key) {
matched_any = true;
capacity += disk.available_space as usize;
}
}
}
if matched_any {
capacity
} else {
fallback_free_capacity(disks)
fallback_free_capacity_dedup(disks)
}
}
/// Total fallback capacity calculation with deweight
///
/// Replace original function: fallback_total_capacity()
pub(crate) fn fallback_total_capacity_dedup(disks: &[rustfs_madmin::Disk]) -> usize {
let mut counted_disks: HashSet<String> = HashSet::new();
let mut total = 0usize;
for disk in disks.iter() {
// Only online disks are counted
if !is_disk_online_state(&disk.state) {
continue;
}
// Use endpoint + drive_path as a unique identifier
let disk_key = format!("{}|{}", disk.endpoint, disk.drive_path);
// Capacity is counted only when the disk is encountered for the first time
if counted_disks.insert(disk_key) {
total += disk.total_space as usize;
}
}
total
}
/// Remove the heavy fallback idle capacity calculation
///
/// Replace original function: fallback_free_capacity()
pub(crate) fn fallback_free_capacity_dedup(disks: &[rustfs_madmin::Disk]) -> usize {
let mut counted_disks: HashSet<String> = HashSet::new();
let mut total = 0usize;
for disk in disks.iter() {
if !is_disk_online_state(&disk.state) {
continue;
}
let disk_key = format!("{}|{}", disk.endpoint, disk.drive_path);
if counted_disks.insert(disk_key) {
total += disk.available_space as usize;
}
}
total
}

177
crates/ecstore/src/pools_test.rs Normal file
View File

@@ -0,0 +1,177 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#[cfg(test)]
mod capacity_dedup_tests {
use crate::pools::{
fallback_free_capacity_dedup, fallback_total_capacity_dedup, get_total_usable_capacity, get_total_usable_capacity_free,
};
#[test]
fn test_single_disk_no_duplication() {
let disks = vec![rustfs_madmin::Disk {
endpoint: "node1".to_string(),
drive_path: "/mnt/disk1".to_string(),
pool_index: 0,
set_index: 0,
disk_index: 0,
total_space: 2_000_000_000_000, // 2TB
available_space: 500_000_000_000,
used_space: 1_500_000_000_000,
state: "ok".to_string(),
..Default::default()
}];
let info = rustfs_madmin::StorageInfo {
backend: rustfs_madmin::BackendInfo {
standard_sc_data: vec![1],
..Default::default()
},
disks: disks.clone(),
};
let total = get_total_usable_capacity(&disks, &info);
let free = get_total_usable_capacity_free(&disks, &info);
assert_eq!(total, 2_000_000_000_000, "Total capacity should be 2TB");
assert_eq!(free, 500_000_000_000, "Free capacity should be 500GB");
}
#[test]
fn test_duplicate_disk_entries_deduped() {
// Simulate the same disk appearing 232 times
let mut disks = Vec::new();
for _ in 0..232 {
disks.push(rustfs_madmin::Disk {
endpoint: "node1".to_string(),
drive_path: "/mnt/disk1".to_string(),
pool_index: 0,
set_index: 0,
disk_index: 0,
total_space: 2_000_000_000_000,
available_space: 500_000_000_000,
used_space: 1_500_000_000_000,
state: "ok".to_string(),
..Default::default()
});
}
let info = rustfs_madmin::StorageInfo {
backend: rustfs_madmin::BackendInfo {
standard_sc_data: vec![1],
..Default::default()
},
disks: disks.clone(),
};
let total = get_total_usable_capacity(&disks, &info);
let free = get_total_usable_capacity_free(&disks, &info);
// Should only be counted once, not 232 times
assert_eq!(total, 2_000_000_000_000, "Duplicate disks should be counted only once");
assert_eq!(free, 500_000_000_000, "Free capacity should not be multiplied");
// If not deduplicated, the result would be:
// total = 2TB × 232 = 464TB ❌
}
#[test]
fn test_four_disk_erasure_coding() {
// 4-disk erasure coding: 2 data disks + 2 parity disks
let disks = vec![
rustfs_madmin::Disk {
endpoint: "node1".to_string(),
drive_path: "/mnt/disk1".to_string(),
pool_index: 0,
set_index: 0,
disk_index: 0,
total_space: 1_000_000_000_000, // 1TB
available_space: 250_000_000_000,
state: "ok".to_string(),
..Default::default()
},
rustfs_madmin::Disk {
endpoint: "node1".to_string(),
drive_path: "/mnt/disk2".to_string(),
pool_index: 0,
set_index: 0,
disk_index: 1,
total_space: 1_000_000_000_000,
available_space: 250_000_000_000,
state: "ok".to_string(),
..Default::default()
},
rustfs_madmin::Disk {
endpoint: "node1".to_string(),
drive_path: "/mnt/disk3".to_string(),
pool_index: 0,
set_index: 0,
disk_index: 2,
total_space: 1_000_000_000_000,
available_space: 250_000_000_000,
state: "ok".to_string(),
..Default::default()
},
rustfs_madmin::Disk {
endpoint: "node1".to_string(),
drive_path: "/mnt/disk4".to_string(),
pool_index: 0,
set_index: 0,
disk_index: 3,
total_space: 1_000_000_000_000,
available_space: 250_000_000_000,
state: "ok".to_string(),
..Default::default()
},
];
let info = rustfs_madmin::StorageInfo {
backend: rustfs_madmin::BackendInfo {
standard_sc_data: vec![2], // 2 data disks
standard_sc_parity: Some(2), // 2 parity disks
..Default::default()
},
disks: disks.clone(),
};
let total = get_total_usable_capacity(&disks, &info);
// Only count data disks (disk_index < 2)
assert_eq!(total, 2_000_000_000_000, "Should count only data disks (2 × 1TB)");
}
#[test]
fn test_fallback_dedup() {
// Test deduplication capability of fallback functions
let mut disks = Vec::new();
// Add duplicate disks
for _ in 0..100 {
disks.push(rustfs_madmin::Disk {
endpoint: "node1".to_string(),
drive_path: "/mnt/disk1".to_string(),
total_space: 2_000_000_000_000,
available_space: 500_000_000_000,
state: "ok".to_string(),
..Default::default()
});
}
let total = fallback_total_capacity_dedup(&disks);
let free = fallback_free_capacity_dedup(&disks);
assert_eq!(total, 2_000_000_000_000);
assert_eq!(free, 500_000_000_000);
}
}

68
crates/ecstore/src/store.rs
View File

@@ -1010,6 +1010,46 @@ impl ECStore {
// *self.pool_meta.write().unwrap() = meta;
Ok(())
}
/// Disk information deduplication function
///
/// Use multiple field combinations to ensure uniqueness:
/// - endpoint (node address)
/// - drive_path (mount path)
/// - pool_index (pool index)
/// - set_index (Collection Index)
/// - disk_index (disk index)
pub(crate) fn deduplicate_disks(disks: Vec<rustfs_madmin::Disk>) -> Vec<rustfs_madmin::Disk> {
use std::collections::HashMap;
let mut unique_disks: HashMap<String, rustfs_madmin::Disk> = HashMap::new();
let mut duplicate_count = 0;
for disk in disks {
// Generate a compound unique key
let key = format!(
"{}|{}|p{}s{}d{}",
disk.endpoint, disk.drive_path, disk.pool_index, disk.set_index, disk.disk_index
);
// Use the entry API to avoid duplicate inserts
use std::collections::hash_map::Entry;
match unique_disks.entry(key) {
Entry::Vacant(e) => {
e.insert(disk);
}
Entry::Occupied(_) => {
duplicate_count += 1;
}
}
}
if duplicate_count > 0 {
debug!("Deduplicated {} duplicate disk entries", duplicate_count);
}
unique_disks.into_values().collect()
}
}
pub async fn find_local_disk(disk_path: &String) -> Option<DiskStore> {
@@ -1259,7 +1299,23 @@ impl StorageAPI for ECStore {
return rustfs_madmin::StorageInfo::default();
};
notification_sy.storage_info(self).await
let mut info = notification_sy.storage_info(self).await;
// 🔧 Defensive deduplication: This protection mechanism is retained even if the upstream is fixed
let original_count = info.disks.len();
info.disks = Self::deduplicate_disks(info.disks);
let final_count = info.disks.len();
if original_count != final_count {
warn!(
"Storage info deduplication: removed {} duplicate disk entries ({} -> {})",
original_count - final_count,
original_count,
final_count
);
}
info
}
#[instrument(skip(self))]
async fn local_storage_info(&self) -> rustfs_madmin::StorageInfo {
@@ -1277,6 +1333,16 @@ impl StorageAPI for ECStore {
disks.extend_from_slice(&res.disks);
}
// 🔧 Defensive deduplication: when aggregating disks from all pools, drop duplicate
// entries that may be reported multiple times by backends; this extra layer is kept
// even if the upstream reporting is later fixed.
let original_count = disks.len();
disks = Self::deduplicate_disks(disks);
if original_count != disks.len() {
warn!("Local storage info deduplication: {} -> {}", original_count, disks.len());
}
let backend = self.backend_info().await;
rustfs_madmin::StorageInfo { backend, disks }
}

91
crates/ecstore/src/store_test.rs Normal file
View File

@@ -0,0 +1,91 @@
// Copyright 2024 RustFS Team
//
// Licensed under the Apache License, Version 2.0 (the "License");
// you may not use this file except in compliance with the License.
// You may obtain a copy of the License at
//
// http://www.apache.org/licenses/LICENSE-2.0
//
// Unless required by applicable law or agreed to in writing, software
// distributed under the License is distributed on an "AS IS" BASIS,
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
// See the License for the specific language governing permissions and
// limitations under the License.
#[cfg(test)]
mod store_dedup_tests {
use crate::store::ECStore;
#[test]
fn test_deduplicate_disks() {
let mut disks = Vec::new();
// Add original disk
disks.push(rustfs_madmin::Disk {
endpoint: "node1".to_string(),
drive_path: "/mnt/disk1".to_string(),
pool_index: 0,
set_index: 0,
disk_index: 0,
total_space: 1_000_000_000_000,
..Default::default()
});
// Add 231 duplicates
for _ in 0..231 {
disks.push(rustfs_madmin::Disk {
endpoint: "node1".to_string(),
drive_path: "/mnt/disk1".to_string(),
pool_index: 0,
set_index: 0,
disk_index: 0,
total_space: 1_000_000_000_000,
..Default::default()
});
}
assert_eq!(disks.len(), 232);
let deduped = ECStore::deduplicate_disks(disks);
assert_eq!(deduped.len(), 1, "Should deduplicate to 1 unique disk");
}
#[test]
fn test_deduplicate_multiple_unique_disks() {
let disks = vec![
rustfs_madmin::Disk {
endpoint: "node1".to_string(),
drive_path: "/mnt/disk1".to_string(),
pool_index: 0,
set_index: 0,
disk_index: 0,
total_space: 1_000_000_000_000,
..Default::default()
},
rustfs_madmin::Disk {
endpoint: "node1".to_string(),
drive_path: "/mnt/disk2".to_string(),
pool_index: 0,
set_index: 0,
disk_index: 1,
total_space: 1_000_000_000_000,
..Default::default()
},
// Duplicate disk1
rustfs_madmin::Disk {
endpoint: "node1".to_string(),
drive_path: "/mnt/disk1".to_string(),
pool_index: 0,
set_index: 0,
disk_index: 0,
total_space: 1_000_000_000_000,
..Default::default()
},
];
let deduped = ECStore::deduplicate_disks(disks);
assert_eq!(deduped.len(), 2, "Should keep 2 unique disks");
}
}

6
crates/scanner/Cargo.toml
View File

@@ -38,8 +38,6 @@ tracing = { workspace = true }
serde = { workspace = true, features = ["derive"] }
serde_json = { workspace = true }
thiserror = { workspace = true }
uuid = { workspace = true, features = ["v4", "serde"] }
anyhow = { workspace = true }
async-trait = { workspace = true }
futures = { workspace = true }
time = { workspace = true }
@@ -55,7 +53,3 @@ rand = { workspace = true }
s3s = { workspace = true }
[dev-dependencies]
tokio-test = { workspace = true }
tracing-subscriber = { workspace = true }
tempfile = { workspace = true }
serial_test = { workspace = true }

4
crates/utils/Cargo.toml
View File

@@ -43,9 +43,9 @@ local-ip-address = { workspace = true, optional = true }
lz4 = { workspace = true, optional = true }
md-5 = { workspace = true, optional = true }
netif = { workspace = true, optional = true }
nix = { workspace = true, optional = true }
rand = { workspace = true, optional = true }
regex = { workspace = true, optional = true }
rustix = { workspace = true, optional = true }
rustfs-config = { workspace = true, features = ["constants"] }
rustls = { workspace = true, optional = true }
rustls-pemfile = { workspace = true, optional = true }
@@ -89,7 +89,7 @@ compress = ["dep:flate2", "dep:brotli", "dep:snap", "dep:lz4", "dep:zstd"]
string = ["dep:regex"]
crypto = ["dep:base64-simd", "dep:hex-simd", "dep:hmac", "dep:hyper", "dep:sha1"]
hash = ["dep:highway", "dep:md-5", "dep:sha2", "dep:blake3", "dep:serde", "dep:siphasher", "dep:hex-simd", "dep:crc-fast"]
os = ["dep:nix", "dep:tempfile", "dep:windows"] # operating system utilities
os = ["dep:rustix", "dep:tempfile", "dep:windows"] # operating system utilities
integration = [] # integration test features
sys = ["dep:sysinfo"] # system information features
http = ["dep:convert_case", "dep:http", "dep:regex"]

85
crates/utils/src/os/linux.rs
View File

@@ -12,29 +12,40 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use nix::sys::stat::{self, stat};
use nix::sys::statfs::{self, FsType, statfs};
use super::{DiskInfo, IOStats};
use rustix::fs::statfs;
use std::fs::File;
use std::io::{self, BufRead, Error, ErrorKind};
use std::path::Path;
use super::{DiskInfo, IOStats};
/// Returns total and free bytes available in a directory, e.g. `/`.
pub fn get_info(p: impl AsRef<Path>) -> std::io::Result<DiskInfo> {
let path_display = p.as_ref().display();
let stat_fs = statfs(p.as_ref())?;
// Use statfs on Linux to get access to f_type (filesystem magic number)
let stat = statfs(p.as_ref())?;
let bsize = stat_fs.block_size() as u64;
let bfree = stat_fs.blocks_free() as u64;
let bavail = stat_fs.blocks_available() as u64;
let blocks = stat_fs.blocks() as u64;
// Linux statfs:
// f_bsize: Optimal transfer block size
// f_blocks: Total data blocks in file system
// f_frsize: Fragment size (since Linux 2.6) - unit for blocks
//
// If f_frsize is > 0, it is the unit for f_blocks, f_bfree, f_bavail.
// Otherwise f_bsize is used.
let bsize = if stat.f_frsize > 0 {
stat.f_frsize as u64
} else {
stat.f_bsize as u64
};
let bfree = stat.f_bfree as u64;
let bavail = stat.f_bavail as u64;
let blocks = stat.f_blocks as u64;
let reserved = match bfree.checked_sub(bavail) {
Some(reserved) => reserved,
None => {
return Err(Error::other(format!(
"detected f_bavail space ({bavail}) > f_bfree space ({bfree}), fs corruption at ({path_display}). please run 'fsck'"
"detected f_bavail space ({bavail}) > f_bfree space ({bfree}), fs corruption at ({path_display}). please run 'fsck'",
)));
}
};
@@ -43,7 +54,7 @@ pub fn get_info(p: impl AsRef<Path>) -> std::io::Result<DiskInfo> {
Some(total) => total * bsize,
None => {
return Err(Error::other(format!(
"detected reserved space ({reserved}) > blocks space ({blocks}), fs corruption at ({path_display}). please run 'fsck'"
"detected reserved space ({reserved}) > blocks space ({blocks}), fs corruption at ({path_display}). please run 'fsck'",
)));
}
};
@@ -58,17 +69,17 @@ pub fn get_info(p: impl AsRef<Path>) -> std::io::Result<DiskInfo> {
}
};
let st = stat(p.as_ref())?;
let st = rustix::fs::stat(p.as_ref())?;
Ok(DiskInfo {
total,
free,
used,
files: stat_fs.files(),
ffree: stat_fs.files_free(),
fstype: get_fs_type(stat_fs.filesystem_type()).to_string(),
major: stat::major(st.st_dev),
minor: stat::minor(st.st_dev),
files: stat.f_files as u64,
ffree: stat.f_ffree as u64,
fstype: get_fs_type(stat.f_type as u64).to_string(),
major: rustix::fs::major(st.st_dev) as u64,
minor: rustix::fs::minor(st.st_dev) as u64,
..Default::default()
})
}
@@ -85,23 +96,26 @@ pub fn get_info(p: impl AsRef<Path>) -> std::io::Result<DiskInfo> {
/// "2fc12fc1" => "zfs",
/// "ff534d42" => "cifs",
/// "53464846" => "wslfs",
fn get_fs_type(fs_type: FsType) -> &'static str {
fn get_fs_type(fs_type: u64) -> &'static str {
// Magic numbers for various filesystems
match fs_type {
statfs::TMPFS_MAGIC => "TMPFS",
statfs::MSDOS_SUPER_MAGIC => "MSDOS",
// statfs::XFS_SUPER_MAGIC => "XFS",
statfs::NFS_SUPER_MAGIC => "NFS",
statfs::EXT4_SUPER_MAGIC => "EXT4",
statfs::ECRYPTFS_SUPER_MAGIC => "ecryptfs",
statfs::OVERLAYFS_SUPER_MAGIC => "overlayfs",
statfs::REISERFS_SUPER_MAGIC => "REISERFS",
0x01021994 => "TMPFS",
0x4d44 => "MSDOS",
0x6969 => "NFS",
0xEF53 => "EXT4",
0xf15f => "ecryptfs",
0x794c7630 => "overlayfs",
0x52654973 => "REISERFS",
// Additional common ones can be added here:
// 0x58465342 => "XFS",
// 0x9123683E => "BTRFS",
_ => "UNKNOWN",
}
}
pub fn same_disk(disk1: &str, disk2: &str) -> std::io::Result<bool> {
let stat1 = stat(disk1)?;
let stat2 = stat(disk2)?;
let stat1 = rustix::fs::stat(disk1)?;
let stat2 = rustix::fs::stat(disk2)?;
Ok(stat1.st_dev == stat2.st_dev)
}
@@ -166,18 +180,3 @@ fn read_stat(file_name: &str) -> std::io::Result<Vec<u64>> {
Ok(stats)
}
#[cfg(test)]
mod test {
use super::get_drive_stats;
use tracing::debug;
#[ignore] // FIXME: failed in github actions
#[test]
fn test_stats() {
let major = 7;
let minor = 11;
let s = get_drive_stats(major, minor).unwrap();
debug!("Drive stats for major: {}, minor: {} - {:?}", major, minor, s);
}
}

18
crates/utils/src/os/mod.rs
View File

@@ -22,10 +22,10 @@ mod windows;
#[cfg(target_os = "linux")]
pub use linux::{get_drive_stats, get_info, same_disk};
// pub use linux::same_disk;
#[cfg(all(unix, not(target_os = "linux")))]
pub use unix::{get_drive_stats, get_info, same_disk};
#[cfg(target_os = "windows")]
pub use windows::{get_drive_stats, get_info, same_disk};
@@ -79,14 +79,19 @@ mod tests {
assert!(info.total > 0);
assert!(info.free > 0);
assert!(info.used > 0);
assert!(info.files > 0);
assert!(info.ffree > 0);
// Files count might be 0 on some systems/filesystems or if empty
// assert!(info.files > 0);
// assert!(info.ffree > 0);
assert!(info.total >= info.free);
}
#[test]
fn test_get_info_invalid_path() {
#[cfg(unix)]
let invalid_path = PathBuf::from("/invalid/path");
#[cfg(windows)]
let invalid_path = PathBuf::from("Z:\\invalid\\path");
let result = get_info(&invalid_path);
assert!(result.is_err());
@@ -118,7 +123,10 @@ mod tests {
#[ignore] // FIXME: failed in github actions
#[test]
fn test_get_drive_stats_default() {
let stats = get_drive_stats(0, 0).unwrap();
assert_eq!(stats, IOStats::default());
#[cfg(not(target_os = "linux"))]
{
let stats = get_drive_stats(0, 0).unwrap();
assert_eq!(stats, IOStats::default());
}
}
}

41
crates/utils/src/os/unix.rs
View File

@@ -13,7 +13,7 @@
// limitations under the License.
use super::{DiskInfo, IOStats};
use nix::sys::{stat::stat, statvfs::statvfs};
use rustix::fs::{StatVfs, statvfs};
use std::io::Error;
use std::path::Path;
@@ -22,10 +22,22 @@ pub fn get_info(p: impl AsRef<Path>) -> std::io::Result<DiskInfo> {
let path_display = p.as_ref().display();
let stat = statvfs(p.as_ref())?;
let bsize = stat.block_size();
let bfree = stat.blocks_free() as u64;
let bavail = stat.blocks_available() as u64;
let blocks = stat.blocks() as u64;
// According to POSIX statvfs definition:
// f_bsize: File system block size.
// f_frsize: Fundamental file system block size.
// f_blocks: Total number of blocks on file system in units of f_frsize.
//
// We should use f_frsize to calculate the size in bytes.
// If f_frsize is 0 (which shouldn't happen on compliant systems), fallback to f_bsize.
let bsize = if stat.f_frsize > 0 {
stat.f_frsize as u64
} else {
stat.f_bsize as u64
};
let bfree = stat.f_bfree as u64;
let bavail = stat.f_bavail as u64;
let blocks = stat.f_blocks as u64;
let reserved = match bfree.checked_sub(bavail) {
Some(reserved) => reserved,
@@ -55,24 +67,33 @@ pub fn get_info(p: impl AsRef<Path>) -> std::io::Result<DiskInfo> {
}
};
let st = rustix::fs::stat(p.as_ref())?;
Ok(DiskInfo {
total,
free,
used,
files: stat.files() as u64,
ffree: stat.files_free() as u64,
// Statvfs does not provide a way to return the filesystem as name.
files: stat.f_files,
ffree: stat.f_ffree,
fstype: get_fs_type(&stat).to_string(),
major: rustix::fs::major(st.st_dev) as u64,
minor: rustix::fs::minor(st.st_dev) as u64,
..Default::default()
})
}
fn get_fs_type(_stat: &StatVfs) -> &'static str {
"UNKNOWN"
}
pub fn same_disk(disk1: &str, disk2: &str) -> std::io::Result<bool> {
let stat1 = stat(disk1)?;
let stat2 = stat(disk2)?;
let stat1 = rustix::fs::stat(disk1)?;
let stat2 = rustix::fs::stat(disk2)?;
Ok(stat1.st_dev == stat2.st_dev)
}
#[cfg(not(target_os = "linux"))]
pub fn get_drive_stats(_major: u32, _minor: u32) -> std::io::Result<IOStats> {
Ok(IOStats::default())
}

150
crates/utils/src/os/windows.rs
View File

@@ -12,9 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
#![allow(unsafe_code)] // TODO: audit unsafe code
use crate::os::{DiskInfo, IOStats};
use super::{DiskInfo, IOStats};
use std::io::Error;
use std::path::Path;
use windows::Win32::Foundation::MAX_PATH;
@@ -75,51 +73,12 @@ pub fn get_info(p: impl AsRef<Path>) -> std::io::Result<DiskInfo> {
used: total - free,
files: total_number_of_clusters as u64,
ffree: number_of_free_clusters as u64,
fstype: get_fs_type(&path_wide).unwrap_or_default(),
fstype: get_windows_fs_type(&path_wide).unwrap_or_default(),
..Default::default()
})
}
/// Returns leading volume name.
///
/// # Arguments
/// * `v` - A slice of u16 representing the path in UTF-16 encoding
///
/// # Returns
/// * `Ok(Vec<u16>)` containing the volume name in UTF-16 encoding.
/// * `Err` if an error occurs during the operation.
#[allow(dead_code)]
fn get_volume_name(v: &[u16]) -> std::io::Result<Vec<u16>> {
let mut volume_name_buffer = [0u16; MAX_PATH as usize];
unsafe {
GetVolumePathNameW(windows::core::PCWSTR::from_raw(v.as_ptr()), &mut volume_name_buffer)
.map_err(|e| Error::from_raw_os_error(e.code().0 as i32))?;
}
let len = volume_name_buffer
.iter()
.position(|&x| x == 0)
.unwrap_or(volume_name_buffer.len());
Ok(volume_name_buffer[..len].to_vec())
}
#[allow(dead_code)]
fn utf16_to_string(v: &[u16]) -> String {
let len = v.iter().position(|&x| x == 0).unwrap_or(v.len());
String::from_utf16_lossy(&v[..len])
}
/// Returns the filesystem type of the underlying mounted filesystem
///
/// # Arguments
/// * `p` - A slice of u16 representing the path in UTF-16 encoding
///
/// # Returns
/// * `Ok(String)` containing the filesystem type (e.g., "NTFS", "FAT32").
/// * `Err` if an error occurs during the operation.
#[allow(dead_code)]
fn get_fs_type(p: &[u16]) -> std::io::Result<String> {
fn get_windows_fs_type(p: &[u16]) -> std::io::Result<String> {
let path = get_volume_name(p)?;
let mut volume_serial_number = 0u32;
@@ -143,16 +102,26 @@ fn get_fs_type(p: &[u16]) -> std::io::Result<String> {
Ok(utf16_to_string(&file_system_name_buffer))
}
/// Determines if two paths are on the same disk.
///
/// # Arguments
/// * `disk1` - The first disk path as a string slice.
/// * `disk2` - The second disk path as a string slice.
///
/// # Returns
/// * `Ok(true)` if both paths are on the same disk.
/// * `Ok(false)` if both paths are on different disks.
/// * `Err` if an error occurs during the operation.
fn get_volume_name(v: &[u16]) -> std::io::Result<Vec<u16>> {
let mut volume_name_buffer = [0u16; MAX_PATH as usize];
unsafe {
GetVolumePathNameW(windows::core::PCWSTR::from_raw(v.as_ptr()), &mut volume_name_buffer)
.map_err(|e| Error::from_raw_os_error(e.code().0 as i32))?;
}
let len = volume_name_buffer
.iter()
.position(|&x| x == 0)
.unwrap_or(volume_name_buffer.len());
Ok(volume_name_buffer[..len].to_vec())
}
fn utf16_to_string(v: &[u16]) -> String {
let len = v.iter().position(|&x| x == 0).unwrap_or(v.len());
String::from_utf16_lossy(&v[..len])
}
pub fn same_disk(disk1: &str, disk2: &str) -> std::io::Result<bool> {
let path1_wide: Vec<u16> = disk1.encode_utf16().chain(std::iter::once(0)).collect();
let path2_wide: Vec<u16> = disk2.encode_utf16().chain(std::iter::once(0)).collect();
@@ -163,79 +132,6 @@ pub fn same_disk(disk1: &str, disk2: &str) -> std::io::Result<bool> {
Ok(volume1 == volume2)
}
/// Retrieves I/O statistics for a drive identified by its major and minor numbers.
///
/// # Arguments
/// * `major` - The major number of the drive.
/// * `minor` - The minor number of the drive.
///
/// # Returns
/// * `Ok(IOStats)` containing the I/O statistics.
/// * `Err` if an error occurs during the operation.
pub fn get_drive_stats(_major: u32, _minor: u32) -> std::io::Result<IOStats> {
// Windows does not provide direct IO stats via simple API; this is a stub
// For full implementation, consider using PDH or WMI, but that adds complexity
Ok(IOStats::default())
}
#[cfg(test)]
mod tests {
#[cfg(target_os = "windows")]
#[test]
fn test_get_info_valid_path() {
let temp_dir = tempfile::tempdir().unwrap();
let info = get_info(temp_dir.path()).unwrap();
// Verify disk info is valid
assert!(info.total > 0);
assert!(info.free > 0);
assert!(info.used > 0);
assert!(info.files > 0);
assert!(info.ffree > 0);
assert!(!info.fstype.is_empty());
}
#[cfg(target_os = "windows")]
#[test]
fn test_get_info_invalid_path() {
use std::path::PathBuf;
let invalid_path = PathBuf::from("Z:\\invalid\\path");
let result = get_info(&invalid_path);
assert!(result.is_err());
}
#[cfg(target_os = "windows")]
#[test]
fn test_same_disk_same_path() {
let temp_dir = tempfile::tempdir().unwrap();
let path = temp_dir.path().to_str().unwrap();
let result = same_disk(path, path).unwrap();
assert!(result);
}
#[cfg(target_os = "windows")]
#[test]
fn test_same_disk_different_paths() {
let temp_dir1 = tempfile::tempdir().unwrap();
let temp_dir2 = tempfile::tempdir().unwrap();
let path1 = temp_dir1.path().to_str().unwrap();
let path2 = temp_dir2.path().to_str().unwrap();
let _result = same_disk(path1, path2).unwrap();
// Since both temporary directories are created in the same file system,
// they should be on the same disk in most cases
// Test passes if the function doesn't panic - the actual result depends on test environment
}
#[cfg(target_os = "windows")]
#[test]
fn get_info_with_root_drive() {
let info = get_info("C:\\").unwrap();
assert!(info.total > 0);
assert!(info.free > 0);
assert!(info.used > 0);
assert!(info.files > 0);
assert!(info.ffree > 0);
assert!(!info.fstype.is_empty());
}
}

3
rustfs/Cargo.toml
View File

@@ -96,9 +96,6 @@ serde_urlencoded = { workspace = true }
rustls = { workspace = true }
subtle = { workspace = true }
rustls-pemfile = { workspace = true }
# Time and Date
chrono = { workspace = true }
jiff = { workspace = true }
time = { workspace = true, features = ["parsing", "formatting", "serde"] }

112
rustfs/src/admin/handlers.rs
View File

@@ -34,9 +34,7 @@ use rustfs_ecstore::bucket::metadata_sys;
use rustfs_ecstore::bucket::target::BucketTarget;
use rustfs_ecstore::bucket::utils::is_valid_object_prefix;
use rustfs_ecstore::bucket::versioning_sys::BucketVersioningSys;
use rustfs_ecstore::data_usage::{
aggregate_local_snapshots, compute_bucket_usage, load_data_usage_from_backend, store_data_usage_in_backend,
};
use rustfs_ecstore::data_usage::{compute_bucket_usage, load_data_usage_from_backend, store_data_usage_in_backend};
use rustfs_ecstore::error::StorageError;
use rustfs_ecstore::global::global_rustfs_port;
use rustfs_ecstore::metrics_realtime::{CollectMetricsOpts, MetricType, collect_local_metrics};
@@ -517,30 +515,10 @@ impl Operation for DataUsageInfoHandler {
return Err(S3Error::with_message(S3ErrorCode::InternalError, "Not init".to_string()));
};
let (disk_statuses, mut info) = match aggregate_local_snapshots(store.clone()).await {
Ok((statuses, usage)) => (statuses, usage),
Err(err) => {
warn!("aggregate_local_snapshots failed: {:?}", err);
(
Vec::new(),
load_data_usage_from_backend(store.clone()).await.map_err(|e| {
error!("load_data_usage_from_backend failed {:?}", e);
s3_error!(InternalError, "load_data_usage_from_backend failed")
})?,
)
}
};
let snapshots_available = disk_statuses.iter().any(|status| status.snapshot_exists);
if !snapshots_available {
if let Ok(fallback) = load_data_usage_from_backend(store.clone()).await {
let mut fallback_info = fallback;
fallback_info.disk_usage_status = disk_statuses.clone();
info = fallback_info;
}
} else {
info.disk_usage_status = disk_statuses.clone();
}
let mut info = load_data_usage_from_backend(store.clone()).await.map_err(|e| {
error!("load_data_usage_from_backend failed {:?}", e);
s3_error!(InternalError, "load_data_usage_from_backend failed")
})?;
let last_update_age = info.last_update.and_then(|ts| ts.elapsed().ok());
let data_missing = info.objects_total_count == 0 && info.buckets_count == 0;
@@ -576,16 +554,82 @@ impl Operation for DataUsageInfoHandler {
});
}
info.disk_usage_status = disk_statuses;
// Set capacity information
let sinfo = store.storage_info().await;
info.total_capacity = get_total_usable_capacity(&sinfo.disks, &sinfo) as u64;
info.total_free_capacity = get_total_usable_capacity_free(&sinfo.disks, &sinfo) as u64;
if info.total_capacity > info.total_free_capacity {
info.total_used_capacity = info.total_capacity - info.total_free_capacity;
// 🔧 Use the fixed capacity calculation function (built-in deduplication)
let raw_total = get_total_usable_capacity(&sinfo.disks, &sinfo);
let raw_free = get_total_usable_capacity_free(&sinfo.disks, &sinfo);
// 🔧 Add a plausibility check (extra layer of protection)
const MAX_REASONABLE_CAPACITY: u64 = 100_000 * 1024 * 1024 * 1024 * 1024; // 100 PiB
const MIN_REASONABLE_CAPACITY: u64 = 1024 * 1024 * 1024; // 1 GiB
let total_u64 = raw_total as u64;
let free_u64 = raw_free as u64;
// Detect outliers
if total_u64 > MAX_REASONABLE_CAPACITY {
error!(
"Abnormal total capacity detected: {} bytes ({:.2} TiB), capping to physical capacity",
total_u64,
total_u64 as f64 / (1024.0_f64.powi(4))
);
// Use the number of disks and the average capacity to estimate reasonable values
let disk_count = sinfo.disks.len();
if disk_count > 0 {
// Calculate the actual number of disks after deduplication (via endpoint + drive_path)
use std::collections::HashSet;
let unique_disks: HashSet<String> = sinfo
.disks
.iter()
.map(|d| format!("{}|{}", d.endpoint, d.drive_path))
.collect();
let actual_disk_count = unique_disks.len();
// Use the capacity of the first disk as a reference
if let Some(first_disk) = sinfo.disks.first() {
info.total_capacity = first_disk.total_space * actual_disk_count as u64;
info.total_free_capacity = first_disk.available_space * actual_disk_count as u64;
info!(
"Applied capacity correction: {} unique disks, capacity per disk: {} bytes",
actual_disk_count, first_disk.total_space
);
} else {
info.total_capacity = 0;
info.total_free_capacity = 0;
}
} else {
info.total_capacity = 0;
info.total_free_capacity = 0;
}
} else if total_u64 < MIN_REASONABLE_CAPACITY && total_u64 > 0 {
warn!(
"Unusually small total capacity: {} bytes ({:.2} GiB)",
total_u64,
total_u64 as f64 / (1024.0_f64.powi(3))
);
info.total_capacity = total_u64;
info.total_free_capacity = free_u64;
} else {
// Normal
info.total_capacity = total_u64;
info.total_free_capacity = free_u64;
}
// Calculate the used capacity
info.total_used_capacity = info.total_capacity.saturating_sub(info.total_free_capacity);
// Record the final statistical results
debug!(
"Capacity statistics: total={:.2} TiB, free={:.2} TiB, used={:.2} TiB",
info.total_capacity as f64 / (1024.0_f64.powi(4)),
info.total_free_capacity as f64 / (1024.0_f64.powi(4)),
info.total_used_capacity as f64 / (1024.0_f64.powi(4))
);
let data = serde_json::to_vec(&info)
.map_err(|_e| S3Error::with_message(S3ErrorCode::InternalError, "parse DataUsageInfo failed"))?;

2
scripts/run.sh
View File

@@ -170,7 +170,7 @@ export RUSTFS_NS_SCANNER_INTERVAL=60 # Object scanning interval in seconds
#export RUSTFS_REGION="us-east-1"
export RUSTFS_ENABLE_SCANNER=false
export RUSTFS_ENABLE_SCANNER=true
export RUSTFS_ENABLE_HEAL=false