gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-13 00:26:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(admin): stabilize cluster capacity usage reporting (#5053)

Browse Source 
* fix(admin): stabilize cluster capacity usage reporting

* test(admin): strengthen capacity usage regressions
This commit is contained in:
cxymds
2026-07-20 17:42:03 +08:00
committed by GitHub
parent fe67af3524
commit bd978bed2d
7 changed files with 132 additions and 294 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/capacity/service.rs
+3 -260
View File
@@ -13,14 +13,9 @@
// limitations under the License.
use crate::storage_api::capacity::service::{all_local_disk, disk_drive_path, disk_endpoint};
use rustfs_io_metrics::capacity_metrics::{
record_capacity_cache_hit, record_capacity_cache_miss, record_capacity_cache_served, record_capacity_refresh_request,
record_capacity_scan_mode,
};
use rustfs_object_capacity::{CapacityDiskRef, capacity_manager, scan};
use std::sync::Arc;
use std::time::Instant;
use tracing::{debug, info, warn};
use rustfs_io_metrics::capacity_metrics::{record_capacity_cache_hit, record_capacity_cache_miss};
use rustfs_object_capacity::{CapacityDiskRef, capacity_manager};
use tracing::{info, warn};
const LOG_COMPONENT_CAPACITY: &str = "capacity";
const LOG_SUBSYSTEM_CAPACITY: &str = "capacity";
@@ -32,237 +27,12 @@ pub fn capacity_disk_ref(endpoint: impl Into<String>, drive_path: impl Into<Stri
}
}
fn capacity_disk_refs(disks: &[rustfs_madmin::Disk]) -> Vec<CapacityDiskRef> {
// Admin callers pass cluster-wide `storage_info` disks. The scan walks
// `drive_path` on the local filesystem, so remote disks must be excluded:
// scanning them either double-counts local bytes (same mount layout on every
// node) or fails with NotFound (per-node layouts), and both poison the
// per-disk cache that the scheduled local-only refresh relies on.
disks
.iter()
.filter(|disk| disk.local)
.map(|disk| capacity_disk_ref(disk.endpoint.clone(), disk.drive_path.clone()))
.collect()
}
async fn refresh_admin_disks_with_subset_fallback(
capacity_manager: &capacity_manager::HybridCapacityManager,
all_disks: Vec<CapacityDiskRef>,
allow_dirty_subset: bool,
) -> Result<capacity_manager::CapacityUpdate, String> {
let (refresh_disks, dirty_subset) = if allow_dirty_subset {
scan::select_capacity_refresh_disks(capacity_manager, &all_disks).await
} else {
(all_disks.clone(), false)
};
match scan::refresh_capacity_with_scope(refresh_disks.clone(), dirty_subset).await {
Ok(update) => Ok(update),
Err(err) if dirty_subset => {
warn!(
component = LOG_COMPONENT_CAPACITY,
subsystem = LOG_SUBSYSTEM_CAPACITY,
event = "capacity_refresh_retry",
scope = "dirty_subset",
fallback_scope = "full_disk",
error = %err,
"Capacity refresh failed and will retry with full-disk scope"
);
scan::refresh_capacity_with_scope(all_disks, false).await
}
Err(err) => Err(err),
}
}
pub async fn refresh_or_join_admin_disks(
capacity_manager: Arc<capacity_manager::HybridCapacityManager>,
source: capacity_manager::DataSource,
disks: &[rustfs_madmin::Disk],
allow_dirty_subset: bool,
) -> Result<capacity_manager::CapacityUpdate, String> {
let all_disks = capacity_disk_refs(disks);
let refresh_manager = capacity_manager.clone();
capacity_manager
.refresh_or_join(source, move || {
let capacity_manager = refresh_manager.clone();
let all_disks = all_disks.clone();
async move {
refresh_admin_disks_with_subset_fallback(capacity_manager.as_ref(), all_disks, allow_dirty_subset).await
}
})
.await
}
pub async fn spawn_refresh_if_needed_admin_disks(
capacity_manager: Arc<capacity_manager::HybridCapacityManager>,
source: capacity_manager::DataSource,
disks: &[rustfs_madmin::Disk],
allow_dirty_subset: bool,
) -> bool {
let all_disks = capacity_disk_refs(disks);
let refresh_manager = capacity_manager.clone();
capacity_manager
.spawn_refresh_if_needed(source, move || async move {
refresh_admin_disks_with_subset_fallback(refresh_manager.as_ref(), all_disks, allow_dirty_subset).await
})
.await
}
pub async fn record_capacity_write(scope_token: Option<uuid::Uuid>) {
capacity_manager::get_capacity_manager()
.record_write_operation_with_scope_token(scope_token)
.await;
}
pub async fn resolve_admin_used_capacity(disks: &[rustfs_madmin::Disk], fallback_used_capacity: u64) -> u64 {
let capacity_manager = capacity_manager::get_capacity_manager();
if let Some(cached) = capacity_manager.get_capacity().await {
record_capacity_cache_hit();
let cache_age = cached.last_update.elapsed();
let fast_update_threshold = capacity_manager.get_config().fast_update_threshold;
if cache_age < fast_update_threshold {
record_capacity_cache_served("fresh");
debug!(
"Using cached capacity: {} bytes (age: {:?}, source: {:?}, files={}, estimated={}, degraded={})",
cached.total_used, cache_age, cached.source, cached.file_count, cached.is_estimated, cached.degraded
);
return cached.total_used;
}
let needs_update = capacity_manager.needs_fast_update().await;
let should_block = capacity_manager.should_block_on_refresh(cache_age);
if needs_update && should_block {
let start = Instant::now();
record_capacity_refresh_request("blocking", capacity_manager::DataSource::WriteTriggered.as_metric_label());
return match refresh_or_join_admin_disks(
capacity_manager.clone(),
capacity_manager::DataSource::WriteTriggered,
disks,
true,
)
.await
{
Ok(update) => {
let elapsed = start.elapsed();
debug!(
component = LOG_COMPONENT_CAPACITY,
subsystem = LOG_SUBSYSTEM_CAPACITY,
event = "capacity_refresh_completed",
mode = "foreground",
duration_ms = elapsed.as_millis() as u64,
file_count = update.file_count,
is_estimated = update.is_estimated,
degraded = update.degraded,
"Capacity refresh completed"
);
update.total_used
}
Err(err) => {
warn!(
component = LOG_COMPONENT_CAPACITY,
subsystem = LOG_SUBSYSTEM_CAPACITY,
event = "capacity_refresh_failed",
mode = "foreground",
fallback = "cached_value",
error = %err,
"Capacity refresh failed"
);
record_capacity_cache_served("stale");
cached.total_used
}
};
}
record_capacity_cache_served("stale");
debug!(
component = LOG_COMPONENT_CAPACITY,
subsystem = LOG_SUBSYSTEM_CAPACITY,
event = "capacity_cache_served",
cache_state = "stale",
total_used = cached.total_used,
age_ms = cache_age.as_millis() as u64,
source = ?cached.source,
file_count = cached.file_count,
is_estimated = cached.is_estimated,
degraded = cached.degraded,
needs_update,
blocking = should_block,
"Served cached capacity"
);
record_capacity_refresh_request("background", capacity_manager::DataSource::Scheduled.as_metric_label());
if spawn_refresh_if_needed_admin_disks(capacity_manager.clone(), capacity_manager::DataSource::Scheduled, disks, true)
.await
{
debug!(
component = LOG_COMPONENT_CAPACITY,
subsystem = LOG_SUBSYSTEM_CAPACITY,
event = "capacity_refresh_background",
state = "started",
"Background capacity refresh state changed"
);
} else {
debug!(
component = LOG_COMPONENT_CAPACITY,
subsystem = LOG_SUBSYSTEM_CAPACITY,
event = "capacity_refresh_background",
state = "skipped",
reason = "already_running",
"Background capacity refresh state changed"
);
}
return cached.total_used;
}
let start = Instant::now();
record_capacity_cache_miss();
record_capacity_refresh_request("initial", capacity_manager::DataSource::RealTime.as_metric_label());
match refresh_or_join_admin_disks(capacity_manager.clone(), capacity_manager::DataSource::RealTime, disks, false).await {
Ok(update) => {
let elapsed = start.elapsed();
info!(
component = LOG_COMPONENT_CAPACITY,
subsystem = LOG_SUBSYSTEM_CAPACITY,
event = "capacity_refresh_completed",
mode = "initial",
total_used = update.total_used,
duration_ms = elapsed.as_millis() as u64,
file_count = update.file_count,
is_estimated = update.is_estimated,
degraded = update.degraded,
"Capacity refresh completed"
);
update.total_used
}
Err(err) => {
warn!(
component = LOG_COMPONENT_CAPACITY,
subsystem = LOG_SUBSYSTEM_CAPACITY,
event = "capacity_refresh_failed",
mode = "initial",
fallback = "disk_used_capacity",
error = %err,
"Capacity refresh failed"
);
record_capacity_cache_served("fallback");
record_capacity_scan_mode("fallback");
capacity_manager
.update_capacity(
capacity_manager::CapacityUpdate::fallback(fallback_used_capacity),
capacity_manager::DataSource::Fallback,
)
.await;
fallback_used_capacity
}
}
}
pub async fn init_capacity_management_for_local_disks() {
info!(
component = LOG_COMPONENT_CAPACITY,
@@ -336,30 +106,3 @@ fn capacity_source_label(source: capacity_manager::DataSource) -> &'static str {
capacity_manager::DataSource::Fallback => "fallback",
}
}
#[cfg(test)]
mod tests {
use super::*;
#[test]
fn capacity_disk_refs_excludes_remote_disks() {
let disks = vec![
rustfs_madmin::Disk {
endpoint: "http://node1:9000/data/rustfs0".to_string(),
drive_path: "/data/rustfs0".to_string(),
local: true,
..Default::default()
},
rustfs_madmin::Disk {
endpoint: "http://node2:9000/data/rustfs0".to_string(),
drive_path: "/data/rustfs0".to_string(),
local: false,
..Default::default()
},
];
let refs = capacity_disk_refs(&disks);
assert_eq!(refs.len(), 1);
assert_eq!(refs[0].endpoint, "http://node1:9000/data/rustfs0");
}
}