mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-26 16:28:15 +00:00
perf(capacity): tune default capacity settings, sync docs, and fix refresh/metrics correctness (#2336)
Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: houseme <housemecn@gmail.com>
This commit is contained in:
houseme
@@ -62,32 +62,32 @@ pub const ENV_CAPACITY_STALL_TIMEOUT: &str = "RUSTFS_CAPACITY_STALL_TIMEOUT";
|
||||
// ============================================================================
|
||||
|
||||
/// Scheduled update interval in seconds
|
||||
/// Default: 300 seconds (5 minutes)
|
||||
pub const DEFAULT_SCHEDULED_UPDATE_INTERVAL_SECS: u64 = 300;
|
||||
/// Default: 120 seconds (2 minutes)
|
||||
pub const DEFAULT_SCHEDULED_UPDATE_INTERVAL_SECS: u64 = 120;
|
||||
|
||||
/// Write trigger delay in seconds
|
||||
/// Default: 10 seconds
|
||||
pub const DEFAULT_WRITE_TRIGGER_DELAY_SECS: u64 = 10;
|
||||
/// Default: 5 seconds
|
||||
pub const DEFAULT_WRITE_TRIGGER_DELAY_SECS: u64 = 5;
|
||||
|
||||
/// Write frequency threshold (writes per minute)
|
||||
/// Default: 10 writes/minute
|
||||
pub const DEFAULT_WRITE_FREQUENCY_THRESHOLD: usize = 10;
|
||||
/// Default: 5 writes/minute
|
||||
pub const DEFAULT_WRITE_FREQUENCY_THRESHOLD: usize = 5;
|
||||
|
||||
/// Fast update threshold in seconds
|
||||
/// Default: 60 seconds
|
||||
pub const DEFAULT_FAST_UPDATE_THRESHOLD_SECS: u64 = 60;
|
||||
/// Default: 30 seconds
|
||||
pub const DEFAULT_FAST_UPDATE_THRESHOLD_SECS: u64 = 30;
|
||||
|
||||
/// Maximum files threshold for sampling
|
||||
/// Default: 1,000,000 files
|
||||
pub const DEFAULT_MAX_FILES_THRESHOLD: usize = 1_000_000;
|
||||
/// Default: 200,000 files
|
||||
pub const DEFAULT_MAX_FILES_THRESHOLD: usize = 200_000;
|
||||
|
||||
/// Statistics timeout in seconds
|
||||
/// Default: 5 seconds
|
||||
pub const DEFAULT_STAT_TIMEOUT_SECS: u64 = 5;
|
||||
/// Default: 3 seconds
|
||||
pub const DEFAULT_STAT_TIMEOUT_SECS: u64 = 3;
|
||||
|
||||
/// Sampling rate (1 in every N files)
|
||||
/// Default: 100
|
||||
pub const DEFAULT_SAMPLE_RATE: usize = 100;
|
||||
/// Default: 200
|
||||
pub const DEFAULT_SAMPLE_RATE: usize = 200;
|
||||
|
||||
/// Follow symbolic links during capacity calculation
|
||||
/// Default: false (disabled for safety)
|
||||
@@ -102,16 +102,16 @@ pub const DEFAULT_CAPACITY_MAX_SYMLINK_DEPTH: u8 = 3;
|
||||
pub const DEFAULT_CAPACITY_ENABLE_DYNAMIC_TIMEOUT: bool = true;
|
||||
|
||||
/// Minimum capacity calculation timeout in seconds
|
||||
/// Default: 5 seconds
|
||||
pub const DEFAULT_CAPACITY_MIN_TIMEOUT_SECS: u64 = 5;
|
||||
/// Default: 2 seconds
|
||||
pub const DEFAULT_CAPACITY_MIN_TIMEOUT_SECS: u64 = 2;
|
||||
|
||||
/// Maximum capacity calculation timeout in seconds
|
||||
/// Default: 60 seconds
|
||||
pub const DEFAULT_CAPACITY_MAX_TIMEOUT_SECS: u64 = 60;
|
||||
/// Default: 15 seconds
|
||||
pub const DEFAULT_CAPACITY_MAX_TIMEOUT_SECS: u64 = 15;
|
||||
|
||||
/// Progress stall detection timeout in seconds
|
||||
/// Default: 1 second (no progress for 1 second = stall)
|
||||
pub const DEFAULT_CAPACITY_STALL_TIMEOUT_SECS: u64 = 1;
|
||||
/// Default: 20 seconds
|
||||
pub const DEFAULT_CAPACITY_STALL_TIMEOUT_SECS: u64 = 20;
|
||||
|
||||
// ============================================================================
|
||||
// Tests
|
||||
@@ -140,16 +140,16 @@ mod tests {
|
||||
|
||||
#[test]
|
||||
fn test_default_values() {
|
||||
assert_eq!(DEFAULT_SCHEDULED_UPDATE_INTERVAL_SECS, 300);
|
||||
assert_eq!(DEFAULT_WRITE_TRIGGER_DELAY_SECS, 10);
|
||||
assert_eq!(DEFAULT_WRITE_FREQUENCY_THRESHOLD, 10);
|
||||
assert_eq!(DEFAULT_FAST_UPDATE_THRESHOLD_SECS, 60);
|
||||
assert_eq!(DEFAULT_MAX_FILES_THRESHOLD, 1_000_000);
|
||||
assert_eq!(DEFAULT_STAT_TIMEOUT_SECS, 5);
|
||||
assert_eq!(DEFAULT_SAMPLE_RATE, 100);
|
||||
assert_eq!(DEFAULT_SCHEDULED_UPDATE_INTERVAL_SECS, 120);
|
||||
assert_eq!(DEFAULT_WRITE_TRIGGER_DELAY_SECS, 5);
|
||||
assert_eq!(DEFAULT_WRITE_FREQUENCY_THRESHOLD, 5);
|
||||
assert_eq!(DEFAULT_FAST_UPDATE_THRESHOLD_SECS, 30);
|
||||
assert_eq!(DEFAULT_MAX_FILES_THRESHOLD, 200_000);
|
||||
assert_eq!(DEFAULT_STAT_TIMEOUT_SECS, 3);
|
||||
assert_eq!(DEFAULT_SAMPLE_RATE, 200);
|
||||
assert_eq!(DEFAULT_CAPACITY_MAX_SYMLINK_DEPTH, 3);
|
||||
assert_eq!(DEFAULT_CAPACITY_MIN_TIMEOUT_SECS, 5);
|
||||
assert_eq!(DEFAULT_CAPACITY_MAX_TIMEOUT_SECS, 60);
|
||||
assert_eq!(DEFAULT_CAPACITY_STALL_TIMEOUT_SECS, 1);
|
||||
assert_eq!(DEFAULT_CAPACITY_MIN_TIMEOUT_SECS, 2);
|
||||
assert_eq!(DEFAULT_CAPACITY_MAX_TIMEOUT_SECS, 15);
|
||||
assert_eq!(DEFAULT_CAPACITY_STALL_TIMEOUT_SECS, 20);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -0,0 +1,92 @@
|
||||
// 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.
|
||||
|
||||
//! Capacity metrics recording helpers.
|
||||
|
||||
use metrics::{counter, gauge, histogram};
|
||||
use std::time::Duration;
|
||||
|
||||
/// Record capacity cache hit.
|
||||
#[inline(always)]
|
||||
pub fn record_capacity_cache_hit() {
|
||||
counter!("rustfs.capacity.cache.hits").increment(1);
|
||||
}
|
||||
|
||||
/// Record capacity cache miss.
|
||||
#[inline(always)]
|
||||
pub fn record_capacity_cache_miss() {
|
||||
counter!("rustfs.capacity.cache.misses").increment(1);
|
||||
}
|
||||
|
||||
/// Record current capacity gauge.
|
||||
#[inline(always)]
|
||||
pub fn record_capacity_current_bytes(used_bytes: u64) {
|
||||
gauge!("rustfs.capacity.current").set(used_bytes as f64);
|
||||
}
|
||||
|
||||
/// Record capacity update completion.
|
||||
#[inline(always)]
|
||||
pub fn record_capacity_update_completed(source: &str, duration: Duration, used_bytes: u64, is_estimated: bool) {
|
||||
counter!("rustfs.capacity.update.total", "source" => source.to_string()).increment(1);
|
||||
histogram!("rustfs.capacity.update.duration.seconds", "source" => source.to_string()).record(duration.as_secs_f64());
|
||||
histogram!("rustfs.capacity.update.bytes", "source" => source.to_string()).record(used_bytes as f64);
|
||||
counter!("rustfs.capacity.update.estimated.total", "source" => source.to_string(), "estimated" => is_estimated.to_string())
|
||||
.increment(1);
|
||||
}
|
||||
|
||||
/// Record failed capacity update.
|
||||
#[inline(always)]
|
||||
pub fn record_capacity_update_failed(source: &str) {
|
||||
counter!("rustfs.capacity.update.failures", "source" => source.to_string()).increment(1);
|
||||
}
|
||||
|
||||
/// Record capacity write activity.
|
||||
#[inline(always)]
|
||||
pub fn record_capacity_write_operation(write_frequency: usize) {
|
||||
counter!("rustfs.capacity.write.operations").increment(1);
|
||||
gauge!("rustfs.capacity.write.frequency").set(write_frequency as f64);
|
||||
}
|
||||
|
||||
/// Record symlink accounting.
|
||||
#[inline(always)]
|
||||
pub fn record_capacity_symlink(size_bytes: u64) {
|
||||
counter!("rustfs.capacity.symlinks.encountered").increment(1);
|
||||
histogram!("rustfs.capacity.symlinks.size.bytes").record(size_bytes as f64);
|
||||
}
|
||||
|
||||
/// Record timeout fallback event.
|
||||
#[inline(always)]
|
||||
pub fn record_capacity_timeout_fallback() {
|
||||
counter!("rustfs.capacity.timeout.fallback").increment(1);
|
||||
}
|
||||
|
||||
/// Record stall detection event.
|
||||
#[inline(always)]
|
||||
pub fn record_capacity_stall_detected() {
|
||||
counter!("rustfs.capacity.timeout.stall").increment(1);
|
||||
}
|
||||
|
||||
/// Record dynamic timeout usage.
|
||||
#[inline(always)]
|
||||
pub fn record_capacity_dynamic_timeout(timeout: Duration) {
|
||||
counter!("rustfs.capacity.timeout.dynamic").increment(1);
|
||||
histogram!("rustfs.capacity.timeout.dynamic.seconds").record(timeout.as_secs_f64());
|
||||
}
|
||||
|
||||
/// Record scan sampling outcome.
|
||||
#[inline(always)]
|
||||
pub fn record_capacity_scan_sampling(sampled_count: usize, estimated: bool) {
|
||||
histogram!("rustfs.capacity.scan.sampled.count").record(sampled_count as f64);
|
||||
counter!("rustfs.capacity.scan.estimated.total", "estimated" => estimated.to_string()).increment(1);
|
||||
}
|
||||
@@ -52,6 +52,7 @@ pub mod adaptive_ttl;
|
||||
pub mod autotuner;
|
||||
pub mod backpressure_metrics;
|
||||
pub mod cache_config;
|
||||
pub mod capacity_metrics;
|
||||
pub mod collector;
|
||||
pub mod config;
|
||||
pub mod deadlock_metrics;
|
||||
@@ -71,6 +72,13 @@ pub use adaptive_ttl::{
|
||||
record_ttl_expiration,
|
||||
};
|
||||
|
||||
// Capacity metrics exports
|
||||
pub use capacity_metrics::{
|
||||
record_capacity_cache_hit, record_capacity_cache_miss, record_capacity_current_bytes, record_capacity_dynamic_timeout,
|
||||
record_capacity_scan_sampling, record_capacity_stall_detected, record_capacity_symlink, record_capacity_timeout_fallback,
|
||||
record_capacity_update_completed, record_capacity_update_failed, record_capacity_write_operation,
|
||||
};
|
||||
|
||||
// I/O metrics exports
|
||||
pub use io_metrics::{
|
||||
IoSchedulerStats, record_bandwidth_observation, record_buffer_size_adjustment, record_io_priority_decision,
|
||||
|
||||
+231
-113
@@ -16,10 +16,9 @@
|
||||
|
||||
use crate::app::context::{AppContext, get_global_app_context};
|
||||
use crate::capacity::capacity_manager::{
|
||||
DataSource, get_capacity_manager, get_enable_dynamic_timeout, get_follow_symlinks, get_max_files_threshold,
|
||||
CapacityUpdate, DataSource, get_capacity_manager, get_enable_dynamic_timeout, get_follow_symlinks, get_max_files_threshold,
|
||||
get_max_symlink_depth, get_max_timeout, get_min_timeout, get_sample_rate, get_stall_timeout, get_stat_timeout,
|
||||
};
|
||||
use crate::capacity::capacity_metrics::get_capacity_metrics;
|
||||
use crate::error::ApiError;
|
||||
use rustfs_common::data_usage::DataUsageInfo;
|
||||
use rustfs_ecstore::admin_server_info::get_server_info;
|
||||
@@ -28,6 +27,10 @@ use rustfs_ecstore::endpoints::EndpointServerPools;
|
||||
use rustfs_ecstore::new_object_layer_fn;
|
||||
use rustfs_ecstore::pools::{PoolStatus, get_total_usable_capacity, get_total_usable_capacity_free};
|
||||
use rustfs_ecstore::store_api::StorageAPI;
|
||||
use rustfs_io_metrics::{
|
||||
record_capacity_dynamic_timeout, record_capacity_scan_sampling, record_capacity_stall_detected, record_capacity_symlink,
|
||||
record_capacity_timeout_fallback,
|
||||
};
|
||||
use rustfs_madmin::{InfoMessage, StorageInfo};
|
||||
use s3s::S3ErrorCode;
|
||||
use std::collections::HashSet;
|
||||
@@ -44,6 +47,31 @@ pub struct QueryServerInfoRequest {
|
||||
pub include_pools: bool,
|
||||
}
|
||||
|
||||
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
|
||||
pub(crate) struct CapacityScanResult {
|
||||
pub used_bytes: u64,
|
||||
pub file_count: usize,
|
||||
pub sampled_count: usize,
|
||||
pub is_estimated: bool,
|
||||
pub scan_duration: Duration,
|
||||
pub had_partial_errors: bool,
|
||||
}
|
||||
|
||||
impl CapacityScanResult {
|
||||
fn with_partial_errors(mut self) -> Self {
|
||||
self.had_partial_errors = true;
|
||||
self
|
||||
}
|
||||
|
||||
pub(crate) fn to_capacity_update(self) -> CapacityUpdate {
|
||||
if self.is_estimated {
|
||||
CapacityUpdate::estimated(self.used_bytes, self.file_count)
|
||||
} else {
|
||||
CapacityUpdate::exact(self.used_bytes, self.file_count)
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
pub struct QueryServerInfoResponse {
|
||||
pub info: InfoMessage,
|
||||
}
|
||||
@@ -69,47 +97,66 @@ pub struct QueryPoolStatusRequest {
|
||||
/// Calculate actual used capacity of all data directories
|
||||
pub(crate) async fn calculate_data_dir_used_capacity(
|
||||
disks: &[rustfs_madmin::Disk],
|
||||
) -> Result<u64, Box<dyn std::error::Error + Send + Sync>> {
|
||||
) -> Result<CapacityScanResult, Box<dyn std::error::Error + Send + Sync>> {
|
||||
let start = Instant::now();
|
||||
let mut total_used = 0u64;
|
||||
let mut total_files = 0usize;
|
||||
let mut total_sampled = 0usize;
|
||||
let mut has_failure = false;
|
||||
let mut has_success = false;
|
||||
let mut is_estimated = false;
|
||||
|
||||
for disk in disks {
|
||||
let path = Path::new(&disk.drive_path);
|
||||
|
||||
// Check if path exists
|
||||
if !path.exists() {
|
||||
warn!("Data directory does not exist: {}", disk.drive_path);
|
||||
has_failure = true;
|
||||
continue;
|
||||
}
|
||||
|
||||
// Asynchronously calculate directory size
|
||||
match get_dir_size_async(path).await {
|
||||
Ok(size) => {
|
||||
debug!("Data directory {} size: {} bytes", disk.drive_path, size);
|
||||
total_used += size;
|
||||
Ok(scan) => {
|
||||
debug!(
|
||||
"Data directory {} size: {} bytes, files={}, sampled={}, estimated={}",
|
||||
disk.drive_path, scan.used_bytes, scan.file_count, scan.sampled_count, scan.is_estimated
|
||||
);
|
||||
total_used += scan.used_bytes;
|
||||
total_files += scan.file_count;
|
||||
total_sampled += scan.sampled_count;
|
||||
is_estimated |= scan.is_estimated;
|
||||
has_failure |= scan.had_partial_errors;
|
||||
has_success = true;
|
||||
}
|
||||
Err(e) => {
|
||||
warn!("Failed to get size for directory {}: {:?}", disk.drive_path, e);
|
||||
has_failure = true;
|
||||
// Continue with other directories
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// If all directories failed, return error to trigger fallback
|
||||
if !has_success {
|
||||
return Err("All directories failed to calculate size".into());
|
||||
}
|
||||
|
||||
// Log warning if there were some failures
|
||||
if has_failure {
|
||||
warn!("Some directories failed to calculate size, result may be incomplete");
|
||||
}
|
||||
|
||||
Ok(total_used)
|
||||
let mut result = CapacityScanResult {
|
||||
used_bytes: total_used,
|
||||
file_count: total_files,
|
||||
sampled_count: total_sampled,
|
||||
is_estimated,
|
||||
scan_duration: start.elapsed(),
|
||||
had_partial_errors: false,
|
||||
};
|
||||
|
||||
if has_failure {
|
||||
result = result.with_partial_errors();
|
||||
}
|
||||
|
||||
Ok(result)
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
@@ -159,11 +206,7 @@ impl SymlinkTracker {
|
||||
self.visited.insert(path);
|
||||
self.symlink_count += 1;
|
||||
self.symlink_size += size;
|
||||
|
||||
// Record to metrics
|
||||
if let Ok(metrics) = std::panic::catch_unwind(std::panic::AssertUnwindSafe(get_capacity_metrics)) {
|
||||
metrics.record_symlink(size);
|
||||
}
|
||||
record_capacity_symlink(size);
|
||||
}
|
||||
|
||||
/// Get symlink statistics
|
||||
@@ -269,11 +312,8 @@ impl ProgressMonitor {
|
||||
files_processed, elapsed, dynamic_timeout
|
||||
);
|
||||
|
||||
// Record timeout to metrics
|
||||
if let Ok(metrics) = std::panic::catch_unwind(std::panic::AssertUnwindSafe(get_capacity_metrics))
|
||||
&& self.used_dynamic_timeout
|
||||
{
|
||||
metrics.record_dynamic_timeout();
|
||||
if self.enable_dynamic_timeout {
|
||||
record_capacity_dynamic_timeout(dynamic_timeout);
|
||||
}
|
||||
|
||||
return Err(std::io::Error::new(
|
||||
@@ -294,10 +334,7 @@ impl ProgressMonitor {
|
||||
self.stall_timeout, files_processed
|
||||
);
|
||||
|
||||
// Record stall to metrics
|
||||
if let Ok(metrics) = std::panic::catch_unwind(std::panic::AssertUnwindSafe(get_capacity_metrics)) {
|
||||
metrics.record_stall_detected();
|
||||
}
|
||||
record_capacity_stall_detected();
|
||||
|
||||
return Err(std::io::Error::new(
|
||||
std::io::ErrorKind::TimedOut,
|
||||
@@ -314,17 +351,14 @@ impl ProgressMonitor {
|
||||
|
||||
/// Record timeout fallback to sampling
|
||||
fn record_timeout_fallback(&self) {
|
||||
if let Ok(metrics) = std::panic::catch_unwind(std::panic::AssertUnwindSafe(get_capacity_metrics)) {
|
||||
metrics.record_timeout_fallback();
|
||||
}
|
||||
record_capacity_timeout_fallback();
|
||||
}
|
||||
}
|
||||
|
||||
/// Asynchronously get directory size with enhanced symlink handling and dynamic timeout
|
||||
async fn get_dir_size_async(path: &Path) -> Result<u64, std::io::Error> {
|
||||
async fn get_dir_size_async(path: &Path) -> Result<CapacityScanResult, std::io::Error> {
|
||||
let path = path.to_path_buf();
|
||||
|
||||
// Get configuration values
|
||||
let max_files_threshold = get_max_files_threshold();
|
||||
let base_timeout = get_stat_timeout();
|
||||
let min_timeout = get_min_timeout();
|
||||
@@ -335,7 +369,6 @@ async fn get_dir_size_async(path: &Path) -> Result<u64, std::io::Error> {
|
||||
let follow_symlinks = get_follow_symlinks();
|
||||
let max_symlink_depth = get_max_symlink_depth();
|
||||
|
||||
// Ensure sample_rate is never zero to avoid panics in is_multiple_of
|
||||
let effective_sample_rate = if sample_rate == 0 {
|
||||
warn!("Invalid sampling configuration: sample_rate=0. Clamping to 1 to avoid panic.");
|
||||
1
|
||||
@@ -343,7 +376,6 @@ async fn get_dir_size_async(path: &Path) -> Result<u64, std::io::Error> {
|
||||
sample_rate
|
||||
};
|
||||
|
||||
// Check if path exists before traversing
|
||||
if !path.exists() {
|
||||
return Err(std::io::Error::new(
|
||||
std::io::ErrorKind::NotFound,
|
||||
@@ -351,15 +383,14 @@ async fn get_dir_size_async(path: &Path) -> Result<u64, std::io::Error> {
|
||||
));
|
||||
}
|
||||
|
||||
// Use tokio::task::spawn_blocking to avoid blocking the async runtime
|
||||
tokio::task::spawn_blocking(move || {
|
||||
let start_time = Instant::now();
|
||||
let mut total_size = 0u64;
|
||||
let mut exact_prefix_bytes = 0u64;
|
||||
let mut overflow_sampled_bytes = 0u64;
|
||||
let mut file_count = 0usize;
|
||||
let mut sampled_size = 0u64;
|
||||
let mut sampled_count = 0usize;
|
||||
let mut had_partial_errors = false;
|
||||
|
||||
// Initialize symlink tracker and progress monitor
|
||||
let mut symlink_tracker = if follow_symlinks {
|
||||
Some(SymlinkTracker::new(max_symlink_depth))
|
||||
} else {
|
||||
@@ -369,7 +400,6 @@ async fn get_dir_size_async(path: &Path) -> Result<u64, std::io::Error> {
|
||||
let mut progress_monitor =
|
||||
ProgressMonitor::new(base_timeout, min_timeout, max_timeout, stall_timeout, enable_dynamic_timeout);
|
||||
|
||||
// Build WalkDir with appropriate settings
|
||||
let mut walker_builder = WalkDir::new(&path);
|
||||
if !follow_symlinks {
|
||||
walker_builder = walker_builder.follow_links(false);
|
||||
@@ -377,79 +407,87 @@ async fn get_dir_size_async(path: &Path) -> Result<u64, std::io::Error> {
|
||||
let walker = walker_builder.into_iter();
|
||||
|
||||
for entry_result in walker {
|
||||
// Propagate traversal errors instead of silently dropping them
|
||||
let entry = match entry_result {
|
||||
Ok(entry) => entry,
|
||||
Err(err) => {
|
||||
warn!("Failed to traverse directory entry under {:?}: {}", path, err);
|
||||
return Err(std::io::Error::other(err.to_string()));
|
||||
}
|
||||
};
|
||||
|
||||
// Get file metadata
|
||||
let metadata = match entry.metadata() {
|
||||
Ok(meta) => meta,
|
||||
Err(err) => {
|
||||
warn!("Failed to get metadata for {:?}: {}", entry.path(), err);
|
||||
had_partial_errors = true;
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
let metadata = match entry.metadata() {
|
||||
Ok(meta) => meta,
|
||||
Err(err) => {
|
||||
warn!("Failed to get metadata for {:?}: {}", entry.path(), err);
|
||||
had_partial_errors = true;
|
||||
continue;
|
||||
}
|
||||
};
|
||||
|
||||
// Handle symlinks if enabled
|
||||
if metadata.is_symlink() {
|
||||
if let Some(ref mut tracker) = symlink_tracker
|
||||
&& let Ok(target) = std::fs::read_link(entry.path())
|
||||
&& tracker.should_follow(&target, 0)
|
||||
{
|
||||
tracker.record_symlink(target, metadata.len());
|
||||
// Don't count symlink size itself, only target
|
||||
continue;
|
||||
}
|
||||
// If not following symlinks, skip
|
||||
continue;
|
||||
}
|
||||
|
||||
// Only count file sizes, ignore directories
|
||||
if !metadata.is_file() {
|
||||
continue;
|
||||
}
|
||||
|
||||
file_count += 1;
|
||||
let exact_count = file_count.min(max_files_threshold);
|
||||
let avg_size = if exact_count > 0 {
|
||||
exact_prefix_bytes / exact_count as u64
|
||||
} else {
|
||||
0
|
||||
};
|
||||
|
||||
// Update progress and check for timeout/stall
|
||||
let avg_size = if file_count > 0 { total_size / file_count as u64 } else { 0 };
|
||||
if let Err(e) = progress_monitor.update_and_check_timeout(file_count, avg_size) {
|
||||
// Timeout or stall detected
|
||||
if sampled_count > 0 {
|
||||
info!("Timeout/stall at {} files, using sampled estimate", file_count);
|
||||
let overflow_count = file_count.saturating_sub(max_files_threshold);
|
||||
let estimated_overflow = overflow_sampled_bytes.saturating_mul(overflow_count as u64) / sampled_count as u64;
|
||||
let estimated_total = exact_prefix_bytes.saturating_add(estimated_overflow);
|
||||
info!(
|
||||
"Timeout/stall at {} files, using sampled estimate: exact_prefix={} overflow_estimate={} sampled={}",
|
||||
file_count, exact_prefix_bytes, estimated_overflow, sampled_count
|
||||
);
|
||||
progress_monitor.record_timeout_fallback();
|
||||
return Ok(sampled_size * file_count as u64 / sampled_count as u64);
|
||||
record_capacity_scan_sampling(sampled_count, true);
|
||||
return Ok(CapacityScanResult {
|
||||
used_bytes: estimated_total,
|
||||
file_count,
|
||||
sampled_count,
|
||||
is_estimated: true,
|
||||
scan_duration: start_time.elapsed(),
|
||||
had_partial_errors,
|
||||
});
|
||||
}
|
||||
return Err(e);
|
||||
}
|
||||
|
||||
// When file count exceeds threshold, enable sampling
|
||||
if file_count > max_files_threshold {
|
||||
// Sampling: count 1 in every effective_sample_rate files
|
||||
if file_count.is_multiple_of(effective_sample_rate) {
|
||||
sampled_size += metadata.len();
|
||||
if file_count <= max_files_threshold {
|
||||
exact_prefix_bytes += metadata.len();
|
||||
} else {
|
||||
let overflow_index = file_count - max_files_threshold;
|
||||
if overflow_index.is_multiple_of(effective_sample_rate) {
|
||||
overflow_sampled_bytes += metadata.len();
|
||||
sampled_count += 1;
|
||||
}
|
||||
|
||||
// Log progress every 100k files
|
||||
if file_count.is_multiple_of(100_000) {
|
||||
debug!(
|
||||
"Processed {} files, sampled {} files, size: {} bytes",
|
||||
file_count, sampled_count, sampled_size
|
||||
"Processed {} files, exact_prefix_bytes={}, sampled_overflow={} files/{} bytes",
|
||||
file_count, exact_prefix_bytes, sampled_count, overflow_sampled_bytes
|
||||
);
|
||||
}
|
||||
} else {
|
||||
// Below threshold, full statistics
|
||||
total_size += metadata.len();
|
||||
}
|
||||
}
|
||||
|
||||
// Report symlink statistics if tracking was enabled
|
||||
if let Some(tracker) = symlink_tracker {
|
||||
let (count, size) = tracker.get_stats();
|
||||
if count > 0 {
|
||||
@@ -457,22 +495,68 @@ async fn get_dir_size_async(path: &Path) -> Result<u64, std::io::Error> {
|
||||
}
|
||||
}
|
||||
|
||||
// If sampling was enabled, return estimated value
|
||||
if file_count > max_files_threshold && sampled_count > 0 {
|
||||
let estimated_size = sampled_size * file_count as u64 / sampled_count as u64;
|
||||
let overflow_count = file_count - max_files_threshold;
|
||||
let estimated_overflow = overflow_sampled_bytes.saturating_mul(overflow_count as u64) / sampled_count as u64;
|
||||
let estimated_size = exact_prefix_bytes.saturating_add(estimated_overflow);
|
||||
info!(
|
||||
"Large directory detected: {} files, estimated size: {} bytes (sampled {}/{} files)",
|
||||
file_count, estimated_size, sampled_count, file_count
|
||||
"Large directory detected: {} files, estimated size: {} bytes (exact prefix: {}, sampled overflow {}/{})",
|
||||
file_count, estimated_size, exact_prefix_bytes, sampled_count, overflow_count
|
||||
);
|
||||
Ok(estimated_size)
|
||||
record_capacity_scan_sampling(sampled_count, true);
|
||||
Ok(CapacityScanResult {
|
||||
used_bytes: estimated_size,
|
||||
file_count,
|
||||
sampled_count,
|
||||
is_estimated: true,
|
||||
scan_duration: start_time.elapsed(),
|
||||
had_partial_errors,
|
||||
})
|
||||
} else if file_count > max_files_threshold {
|
||||
// sampled_count == 0: too few overflow files to reach the sample rate threshold.
|
||||
// Fall back to estimating the overflow using the average file size from the exact
|
||||
// prefix so that overflow files are not silently dropped from the total.
|
||||
let overflow_count = file_count - max_files_threshold;
|
||||
// Use the actual number of files counted in the exact prefix, not the threshold
|
||||
// value, to avoid a divide-by-zero or incorrect average when fewer files were
|
||||
// processed than max_files_threshold.
|
||||
let exact_prefix_count = file_count.min(max_files_threshold) as u64;
|
||||
let avg_prefix_size = if exact_prefix_count > 0 {
|
||||
exact_prefix_bytes / exact_prefix_count
|
||||
} else {
|
||||
0
|
||||
};
|
||||
let estimated_overflow = avg_prefix_size.saturating_mul(overflow_count as u64);
|
||||
let estimated_size = exact_prefix_bytes.saturating_add(estimated_overflow);
|
||||
info!(
|
||||
"Large directory detected: {} files, estimated size: {} bytes (no overflow samples, used prefix average {} bytes/file)",
|
||||
file_count, estimated_size, avg_prefix_size
|
||||
);
|
||||
record_capacity_scan_sampling(0, true);
|
||||
Ok(CapacityScanResult {
|
||||
used_bytes: estimated_size,
|
||||
file_count,
|
||||
sampled_count: 0,
|
||||
is_estimated: true,
|
||||
scan_duration: start_time.elapsed(),
|
||||
had_partial_errors,
|
||||
})
|
||||
} else {
|
||||
record_capacity_scan_sampling(0, false);
|
||||
debug!(
|
||||
"Directory size calculation completed: {} files, {} bytes, took {:?}",
|
||||
file_count,
|
||||
total_size,
|
||||
exact_prefix_bytes,
|
||||
start_time.elapsed()
|
||||
);
|
||||
Ok(total_size)
|
||||
Ok(CapacityScanResult {
|
||||
used_bytes: exact_prefix_bytes,
|
||||
file_count,
|
||||
sampled_count,
|
||||
is_estimated: false,
|
||||
scan_duration: start_time.elapsed(),
|
||||
had_partial_errors,
|
||||
})
|
||||
}
|
||||
})
|
||||
.await
|
||||
@@ -616,47 +700,65 @@ impl DefaultAdminUsecase {
|
||||
if cache_age < fast_update_threshold {
|
||||
info.total_used_capacity = cached.total_used;
|
||||
debug!(
|
||||
"Using cached capacity: {} bytes (age: {:?}, source: {:?})",
|
||||
cached.total_used, cache_age, cached.source
|
||||
"Using cached capacity: {} bytes (age: {:?}, source: {:?}, files={}, estimated={})",
|
||||
cached.total_used, cache_age, cached.source, cached.file_count, cached.is_estimated
|
||||
);
|
||||
} else {
|
||||
// Cache is stale, check if we need fast update
|
||||
let needs_update = capacity_manager.needs_fast_update().await;
|
||||
let should_block = capacity_manager.should_block_on_refresh(cache_age);
|
||||
|
||||
if needs_update {
|
||||
// Fast update needed (recent writes or high frequency)
|
||||
if needs_update && should_block {
|
||||
let start = Instant::now();
|
||||
match calculate_data_dir_used_capacity(&storage_info.disks).await {
|
||||
Ok(used_capacity) => {
|
||||
info.total_used_capacity = used_capacity;
|
||||
capacity_manager
|
||||
.update_capacity(used_capacity, DataSource::WriteTriggered)
|
||||
.await;
|
||||
match capacity_manager
|
||||
.refresh_or_join(DataSource::WriteTriggered, || async {
|
||||
calculate_data_dir_used_capacity(&storage_info.disks)
|
||||
.await
|
||||
.map(|scan| scan.to_capacity_update())
|
||||
.map_err(|e| e.to_string())
|
||||
})
|
||||
.await
|
||||
{
|
||||
Ok(update) => {
|
||||
info.total_used_capacity = update.total_used;
|
||||
|
||||
let elapsed = start.elapsed();
|
||||
debug!("Fast capacity update completed in {:?}", elapsed);
|
||||
debug!(
|
||||
"Foreground capacity refresh completed in {:?} (files={}, estimated={})",
|
||||
elapsed, update.file_count, update.is_estimated
|
||||
);
|
||||
}
|
||||
Err(e) => {
|
||||
warn!("Fast capacity update failed: {:?}, using cached value", e);
|
||||
warn!("Foreground capacity refresh failed: {}, using cached value", e);
|
||||
info.total_used_capacity = cached.total_used;
|
||||
}
|
||||
}
|
||||
} else {
|
||||
// Use stale cache and trigger background update (if not already in progress)
|
||||
info.total_used_capacity = cached.total_used;
|
||||
debug!("Using stale cache, background update will be triggered if not already in progress");
|
||||
debug!(
|
||||
"Using stale cached capacity: {} bytes (age: {:?}, source: {:?}, files={}, estimated={}, needs_update={}, blocking={})",
|
||||
cached.total_used,
|
||||
cache_age,
|
||||
cached.source,
|
||||
cached.file_count,
|
||||
cached.is_estimated,
|
||||
needs_update,
|
||||
should_block
|
||||
);
|
||||
|
||||
// Trigger background update only if not already in progress (prevent thundering herd)
|
||||
if capacity_manager.try_start_background_update() {
|
||||
let disks = storage_info.disks.clone();
|
||||
let manager = capacity_manager.clone();
|
||||
tokio::spawn(async move {
|
||||
if let Ok(new_capacity) = calculate_data_dir_used_capacity(&disks).await {
|
||||
manager.update_capacity(new_capacity, DataSource::Scheduled).await;
|
||||
debug!("Background capacity update completed: {} bytes", new_capacity);
|
||||
}
|
||||
manager.complete_background_update();
|
||||
});
|
||||
let disks = storage_info.disks.clone();
|
||||
let manager = capacity_manager.clone();
|
||||
if manager
|
||||
.clone()
|
||||
.spawn_refresh_if_needed(DataSource::Scheduled, move || async move {
|
||||
calculate_data_dir_used_capacity(&disks)
|
||||
.await
|
||||
.map(|scan| scan.to_capacity_update())
|
||||
.map_err(|e| e.to_string())
|
||||
})
|
||||
.await
|
||||
{
|
||||
debug!("Background capacity update started");
|
||||
} else {
|
||||
debug!("Background update already in progress, skipping spawn");
|
||||
}
|
||||
@@ -665,21 +767,33 @@ impl DefaultAdminUsecase {
|
||||
} else {
|
||||
// No cache, perform initial calculation
|
||||
let start = Instant::now();
|
||||
match calculate_data_dir_used_capacity(&storage_info.disks).await {
|
||||
Ok(used_capacity) => {
|
||||
info.total_used_capacity = used_capacity;
|
||||
capacity_manager.update_capacity(used_capacity, DataSource::RealTime).await;
|
||||
match capacity_manager
|
||||
.refresh_or_join(DataSource::RealTime, || async {
|
||||
calculate_data_dir_used_capacity(&storage_info.disks)
|
||||
.await
|
||||
.map(|scan| scan.to_capacity_update())
|
||||
.map_err(|e| e.to_string())
|
||||
})
|
||||
.await
|
||||
{
|
||||
Ok(update) => {
|
||||
info.total_used_capacity = update.total_used;
|
||||
|
||||
let elapsed = start.elapsed();
|
||||
info!("Initial capacity calculation completed: {} bytes in {:?}", used_capacity, elapsed);
|
||||
info!(
|
||||
"Initial capacity calculation completed: {} bytes in {:?} (files={}, estimated={})",
|
||||
update.total_used, elapsed, update.file_count, update.is_estimated
|
||||
);
|
||||
}
|
||||
Err(e) => {
|
||||
warn!(
|
||||
"Failed to calculate data directory used capacity: {:?}, falling back to disk used capacity",
|
||||
"Failed to calculate data directory used capacity: {}, falling back to disk used capacity",
|
||||
e
|
||||
);
|
||||
// Fallback: use disk used capacity
|
||||
info.total_used_capacity = info.total_capacity.saturating_sub(info.total_free_capacity);
|
||||
capacity_manager
|
||||
.update_capacity(CapacityUpdate::fallback(info.total_used_capacity), DataSource::Fallback)
|
||||
.await;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -805,7 +919,8 @@ mod tests {
|
||||
|
||||
let temp_dir = TempDir::new().unwrap();
|
||||
let size = get_dir_size_async(temp_dir.path()).await.unwrap();
|
||||
assert_eq!(size, 0);
|
||||
assert_eq!(size.used_bytes, 0);
|
||||
assert_eq!(size.file_count, 0);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -820,7 +935,8 @@ mod tests {
|
||||
file.write_all(b"Hello, World!").unwrap();
|
||||
|
||||
let size = get_dir_size_async(temp_dir.path()).await.unwrap();
|
||||
assert_eq!(size, 13);
|
||||
assert_eq!(size.used_bytes, 13);
|
||||
assert_eq!(size.file_count, 1);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -839,7 +955,8 @@ mod tests {
|
||||
}
|
||||
|
||||
let size = get_dir_size_async(temp_dir.path()).await.unwrap();
|
||||
assert_eq!(size, 40); // 10 files * 4 bytes
|
||||
assert_eq!(size.used_bytes, 40); // 10 files * 4 bytes
|
||||
assert_eq!(size.file_count, 10);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
@@ -863,7 +980,8 @@ mod tests {
|
||||
f2.write_all(b"content2").unwrap();
|
||||
|
||||
let size = get_dir_size_async(temp_dir.path()).await.unwrap();
|
||||
assert_eq!(size, 16); // "content1" (8) + "content2" (8)
|
||||
assert_eq!(size.used_bytes, 16); // "content1" (8) + "content2" (8)
|
||||
assert_eq!(size.file_count, 2);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
|
||||
@@ -15,9 +15,8 @@
|
||||
//! Capacity management integration for application startup
|
||||
|
||||
use crate::capacity::capacity_manager::{DataSource, get_capacity_manager, start_background_task};
|
||||
use crate::capacity::capacity_metrics::{get_capacity_metrics, start_metrics_logging};
|
||||
use rustfs_ecstore::disk::DiskAPI;
|
||||
use std::time::Duration;
|
||||
use rustfs_io_metrics::{record_capacity_cache_hit, record_capacity_cache_miss};
|
||||
use tracing::{info, warn};
|
||||
|
||||
/// Initialize capacity management system
|
||||
@@ -50,11 +49,6 @@ pub async fn init_capacity_management() {
|
||||
info!("Starting background capacity update task...");
|
||||
start_background_task(disk_refs).await;
|
||||
|
||||
// Start metrics logging (log every 10 minutes)
|
||||
let metrics_interval = Duration::from_secs(600);
|
||||
info!("Starting metrics logging task (interval: {:?})...", metrics_interval);
|
||||
start_metrics_logging(metrics_interval).await;
|
||||
|
||||
info!("Capacity management system initialized successfully");
|
||||
}
|
||||
|
||||
@@ -62,11 +56,10 @@ pub async fn init_capacity_management() {
|
||||
#[allow(dead_code)]
|
||||
pub async fn get_capacity_with_metrics() -> Option<(u64, String)> {
|
||||
let manager = get_capacity_manager();
|
||||
let metrics = get_capacity_metrics();
|
||||
|
||||
// Check cache
|
||||
if let Some(cached) = manager.get_capacity().await {
|
||||
metrics.record_cache_hit();
|
||||
record_capacity_cache_hit();
|
||||
|
||||
let source = match cached.source {
|
||||
DataSource::RealTime => "real-time",
|
||||
@@ -78,19 +71,21 @@ pub async fn get_capacity_with_metrics() -> Option<(u64, String)> {
|
||||
return Some((cached.total_used, source.to_string()));
|
||||
}
|
||||
|
||||
metrics.record_cache_miss();
|
||||
record_capacity_cache_miss();
|
||||
None
|
||||
}
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
use crate::capacity::capacity_manager::{DataSource, get_capacity_manager};
|
||||
use crate::capacity::capacity_manager::{CapacityUpdate, DataSource, get_capacity_manager};
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_get_capacity_with_metrics() {
|
||||
let manager = get_capacity_manager();
|
||||
manager.update_capacity(1000, DataSource::RealTime).await;
|
||||
manager
|
||||
.update_capacity(CapacityUpdate::exact(1000, 0), DataSource::RealTime)
|
||||
.await;
|
||||
|
||||
let result = get_capacity_with_metrics().await;
|
||||
assert!(result.is_some());
|
||||
|
||||
@@ -15,7 +15,6 @@
|
||||
//! Hybrid Capacity Manager for efficient capacity statistics
|
||||
|
||||
use crate::app::admin_usecase::calculate_data_dir_used_capacity;
|
||||
use metrics::{counter, gauge};
|
||||
use rustfs_config::{
|
||||
DEFAULT_CAPACITY_ENABLE_DYNAMIC_TIMEOUT, DEFAULT_CAPACITY_FOLLOW_SYMLINKS, DEFAULT_CAPACITY_MAX_SYMLINK_DEPTH,
|
||||
DEFAULT_CAPACITY_MAX_TIMEOUT_SECS, DEFAULT_CAPACITY_MIN_TIMEOUT_SECS, DEFAULT_CAPACITY_STALL_TIMEOUT_SECS,
|
||||
@@ -26,12 +25,13 @@ use rustfs_config::{
|
||||
ENV_CAPACITY_SAMPLE_RATE, ENV_CAPACITY_SCHEDULED_INTERVAL, ENV_CAPACITY_STALL_TIMEOUT, ENV_CAPACITY_STAT_TIMEOUT,
|
||||
ENV_CAPACITY_WRITE_FREQUENCY_THRESHOLD, ENV_CAPACITY_WRITE_TRIGGER_DELAY,
|
||||
};
|
||||
use rustfs_io_metrics::{record_capacity_current_bytes, record_capacity_update_completed, record_capacity_write_operation};
|
||||
use rustfs_utils::{get_env_bool, get_env_u64, get_env_usize};
|
||||
use std::future::Future;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicBool, Ordering};
|
||||
use std::time::{Duration, Instant};
|
||||
use tokio::sync::RwLock;
|
||||
use tracing::{debug, error, info, warn};
|
||||
use tokio::sync::{Mutex, RwLock, watch};
|
||||
use tracing::{debug, info, warn};
|
||||
|
||||
// ============================================================================
|
||||
// Configuration Functions
|
||||
@@ -279,15 +279,53 @@ pub struct CachedCapacity {
|
||||
/// Last update time
|
||||
pub last_update: Instant,
|
||||
/// File count (optional)
|
||||
#[allow(dead_code)]
|
||||
pub file_count: usize,
|
||||
/// Whether it's an estimated value
|
||||
#[allow(dead_code)]
|
||||
pub is_estimated: bool,
|
||||
/// Data source
|
||||
pub source: DataSource,
|
||||
}
|
||||
|
||||
/// Structured capacity update payload.
|
||||
#[derive(Clone, Debug)]
|
||||
pub struct CapacityUpdate {
|
||||
/// Total used capacity in bytes.
|
||||
pub total_used: u64,
|
||||
/// Number of files observed during scan.
|
||||
pub file_count: usize,
|
||||
/// Whether the value is estimated instead of exact.
|
||||
pub is_estimated: bool,
|
||||
}
|
||||
|
||||
impl CapacityUpdate {
|
||||
/// Create an exact capacity update.
|
||||
pub fn exact(total_used: u64, file_count: usize) -> Self {
|
||||
Self {
|
||||
total_used,
|
||||
file_count,
|
||||
is_estimated: false,
|
||||
}
|
||||
}
|
||||
|
||||
/// Create an estimated capacity update.
|
||||
pub fn estimated(total_used: u64, file_count: usize) -> Self {
|
||||
Self {
|
||||
total_used,
|
||||
file_count,
|
||||
is_estimated: true,
|
||||
}
|
||||
}
|
||||
|
||||
/// Create a fallback capacity update.
|
||||
pub fn fallback(total_used: u64) -> Self {
|
||||
Self {
|
||||
total_used,
|
||||
file_count: 0,
|
||||
is_estimated: true,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
#[derive(Clone, Debug, PartialEq, Copy, Eq)]
|
||||
pub enum DataSource {
|
||||
/// Real-time statistics
|
||||
@@ -301,6 +339,17 @@ pub enum DataSource {
|
||||
Fallback,
|
||||
}
|
||||
|
||||
impl DataSource {
|
||||
fn as_metric_label(self) -> &'static str {
|
||||
match self {
|
||||
Self::RealTime => "realtime",
|
||||
Self::Scheduled => "scheduled",
|
||||
Self::WriteTriggered => "write_triggered",
|
||||
Self::Fallback => "fallback",
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Write record for tracking write operations
|
||||
#[derive(Debug)]
|
||||
pub struct WriteRecord {
|
||||
@@ -354,6 +403,25 @@ impl HybridStrategyConfig {
|
||||
// Hybrid Capacity Manager
|
||||
// ============================================================================
|
||||
|
||||
struct RefreshState {
|
||||
running: bool,
|
||||
/// Sender for the current refresh cycle. Joiners subscribe to this before releasing the
|
||||
/// mutex so they cannot miss the completion notification. A new channel is created at the
|
||||
/// start of every refresh cycle so stale subscribers from previous cycles are not confused
|
||||
/// by results that were already published.
|
||||
result_tx: watch::Sender<Option<Result<CapacityUpdate, String>>>,
|
||||
}
|
||||
|
||||
impl Default for RefreshState {
|
||||
fn default() -> Self {
|
||||
let (tx, _) = watch::channel(None);
|
||||
Self {
|
||||
running: false,
|
||||
result_tx: tx,
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// Hybrid capacity manager
|
||||
pub struct HybridCapacityManager {
|
||||
/// Capacity cache
|
||||
@@ -362,11 +430,17 @@ pub struct HybridCapacityManager {
|
||||
write_record: Arc<RwLock<WriteRecord>>,
|
||||
/// Configuration
|
||||
config: HybridStrategyConfig,
|
||||
/// Background update in progress flag
|
||||
update_in_progress: Arc<AtomicBool>,
|
||||
/// Shared singleflight refresh state
|
||||
refresh_state: Arc<Mutex<RefreshState>>,
|
||||
}
|
||||
|
||||
impl HybridCapacityManager {
|
||||
fn max_stale_age(&self) -> Duration {
|
||||
self.config
|
||||
.scheduled_update_interval
|
||||
.max(self.config.fast_update_threshold.checked_mul(3).unwrap_or(Duration::MAX))
|
||||
}
|
||||
|
||||
/// Create a new hybrid capacity manager
|
||||
pub fn new(config: HybridStrategyConfig) -> Self {
|
||||
Self {
|
||||
@@ -377,7 +451,7 @@ impl HybridCapacityManager {
|
||||
write_window: Vec::new(),
|
||||
})),
|
||||
config,
|
||||
update_in_progress: Arc::new(AtomicBool::new(false)),
|
||||
refresh_state: Arc::new(Mutex::new(RefreshState::default())),
|
||||
}
|
||||
}
|
||||
|
||||
@@ -393,25 +467,23 @@ impl HybridCapacityManager {
|
||||
}
|
||||
|
||||
/// Update capacity
|
||||
pub async fn update_capacity(&self, capacity: u64, source: DataSource) {
|
||||
pub async fn update_capacity(&self, update: CapacityUpdate, source: DataSource) {
|
||||
let start = Instant::now();
|
||||
let mut cache = self.cache.write().await;
|
||||
*cache = Some(CachedCapacity {
|
||||
total_used: capacity,
|
||||
total_used: update.total_used,
|
||||
last_update: Instant::now(),
|
||||
file_count: 0,
|
||||
is_estimated: false,
|
||||
file_count: update.file_count,
|
||||
is_estimated: update.is_estimated,
|
||||
source,
|
||||
});
|
||||
|
||||
debug!("Capacity updated: {} bytes, source: {:?}", capacity, source);
|
||||
// Update metrics
|
||||
gauge!("rustfs.capacity.current").set(capacity as f64);
|
||||
match source {
|
||||
DataSource::RealTime => counter!("rustfs.capacity.update.realtime").increment(1),
|
||||
DataSource::Scheduled => counter!("rustfs.capacity.update.scheduled").increment(1),
|
||||
DataSource::WriteTriggered => counter!("rustfs.capacity.update.write_triggered").increment(1),
|
||||
DataSource::Fallback => counter!("rustfs.capacity.update.fallback").increment(1),
|
||||
}
|
||||
debug!(
|
||||
"Capacity updated: {} bytes, files={}, estimated={}, source: {:?}",
|
||||
update.total_used, update.file_count, update.is_estimated, source
|
||||
);
|
||||
record_capacity_current_bytes(update.total_used);
|
||||
record_capacity_update_completed(source.as_metric_label(), start.elapsed(), update.total_used, update.is_estimated);
|
||||
}
|
||||
|
||||
/// Record write operation
|
||||
@@ -432,8 +504,7 @@ impl HybridCapacityManager {
|
||||
record.write_window.push(now);
|
||||
}
|
||||
|
||||
counter!("rustfs.capacity.write.operations").increment(1);
|
||||
gauge!("rustfs.capacity.write.frequency").set(record.write_window.len() as f64);
|
||||
record_capacity_write_operation(record.write_window.len());
|
||||
debug!(
|
||||
"Write operation recorded: total writes = {}, recent writes = {}",
|
||||
record.write_count,
|
||||
@@ -490,21 +561,110 @@ impl HybridCapacityManager {
|
||||
record.write_window.len()
|
||||
}
|
||||
|
||||
/// Run a singleflight refresh. Callers either join an existing in-flight refresh or become the leader.
|
||||
///
|
||||
/// Joiners subscribe to the watch channel *before* releasing the mutex, which guarantees
|
||||
/// they cannot miss the completion notification even if the leader finishes very quickly.
|
||||
pub async fn refresh_or_join<F, Fut>(&self, source: DataSource, refresh_fn: F) -> Result<CapacityUpdate, String>
|
||||
where
|
||||
F: FnOnce() -> Fut,
|
||||
Fut: Future<Output = Result<CapacityUpdate, String>>,
|
||||
{
|
||||
let maybe_rx = {
|
||||
let mut state = self.refresh_state.lock().await;
|
||||
if state.running {
|
||||
// Subscribe while holding the lock so the send that completes the current
|
||||
// refresh cycle cannot happen before we are subscribed.
|
||||
Some(state.result_tx.subscribe())
|
||||
} else {
|
||||
// Become the leader. Create a fresh channel so that joiners from a previous
|
||||
// cycle cannot observe the result that was published for the new cycle.
|
||||
let (tx, _) = watch::channel(None);
|
||||
state.result_tx = tx;
|
||||
state.running = true;
|
||||
None
|
||||
}
|
||||
};
|
||||
|
||||
if let Some(mut result_rx) = maybe_rx {
|
||||
// Wait until the leader publishes Some(result). Because we subscribed before
|
||||
// releasing the mutex, we cannot miss the notification.
|
||||
if result_rx.wait_for(|v| v.is_some()).await.is_err() {
|
||||
// The leader's sender was dropped (e.g. due to a panic) without publishing
|
||||
// a result. Surface a clear error rather than silently returning the default.
|
||||
return Err("capacity refresh leader exited without publishing a result".to_string());
|
||||
}
|
||||
return result_rx
|
||||
.borrow()
|
||||
.as_ref()
|
||||
.cloned()
|
||||
.unwrap_or_else(|| Err("capacity refresh completed without a result".to_string()));
|
||||
}
|
||||
|
||||
let result = refresh_fn().await;
|
||||
if let Ok(update) = &result {
|
||||
self.update_capacity(update.clone(), source).await;
|
||||
}
|
||||
|
||||
{
|
||||
let mut state = self.refresh_state.lock().await;
|
||||
state.running = false;
|
||||
let _ = state.result_tx.send(Some(result.clone()));
|
||||
}
|
||||
|
||||
result
|
||||
}
|
||||
|
||||
/// Start a background refresh if one is not already in flight.
|
||||
pub async fn spawn_refresh_if_needed<F, Fut>(self: Arc<Self>, source: DataSource, refresh_fn: F) -> bool
|
||||
where
|
||||
F: FnOnce() -> Fut + Send + 'static,
|
||||
Fut: Future<Output = Result<CapacityUpdate, String>> + Send + 'static,
|
||||
{
|
||||
let should_spawn = {
|
||||
let mut state = self.refresh_state.lock().await;
|
||||
if state.running {
|
||||
false
|
||||
} else {
|
||||
let (tx, _) = watch::channel(None);
|
||||
state.result_tx = tx;
|
||||
state.running = true;
|
||||
true
|
||||
}
|
||||
};
|
||||
|
||||
if !should_spawn {
|
||||
return false;
|
||||
}
|
||||
|
||||
tokio::spawn(async move {
|
||||
let result = refresh_fn().await;
|
||||
if let Ok(update) = &result {
|
||||
self.update_capacity(update.clone(), source).await;
|
||||
}
|
||||
|
||||
let mut state = self.refresh_state.lock().await;
|
||||
state.running = false;
|
||||
let _ = state.result_tx.send(Some(result));
|
||||
});
|
||||
|
||||
true
|
||||
}
|
||||
|
||||
/// Get config
|
||||
pub fn get_config(&self) -> &HybridStrategyConfig {
|
||||
&self.config
|
||||
}
|
||||
|
||||
/// Try to start a background update, returns true if update was started (false if already in progress)
|
||||
pub fn try_start_background_update(&self) -> bool {
|
||||
self.update_in_progress
|
||||
.compare_exchange(false, true, Ordering::AcqRel, Ordering::Acquire)
|
||||
.is_ok()
|
||||
/// Check if the cache is too stale to keep serving without a foreground refresh.
|
||||
pub fn should_block_on_refresh(&self, cache_age: Duration) -> bool {
|
||||
cache_age >= self.max_stale_age()
|
||||
}
|
||||
|
||||
/// Mark background update as complete
|
||||
pub fn complete_background_update(&self) {
|
||||
self.update_in_progress.store(false, Ordering::Release);
|
||||
/// Return whether a refresh is currently in flight.
|
||||
#[cfg(test)]
|
||||
pub async fn refresh_in_progress(&self) -> bool {
|
||||
self.refresh_state.lock().await.running
|
||||
}
|
||||
}
|
||||
|
||||
@@ -526,7 +686,9 @@ pub fn get_capacity_manager() -> Arc<HybridCapacityManager> {
|
||||
/// # Example
|
||||
/// ```no_run
|
||||
/// let manager = create_isolated_manager(HybridStrategyConfig::default());
|
||||
/// manager.update_capacity(1000, DataSource::RealTime).await;
|
||||
/// manager
|
||||
/// .update_capacity(CapacityUpdate::exact(1000, 0), DataSource::RealTime)
|
||||
/// .await;
|
||||
/// ```
|
||||
#[cfg(test)]
|
||||
#[allow(dead_code)]
|
||||
@@ -553,17 +715,22 @@ pub async fn start_background_task(disks: Vec<rustfs_madmin::Disk>) {
|
||||
|
||||
info!("Starting scheduled capacity update");
|
||||
let start = Instant::now();
|
||||
let manager = manager.clone();
|
||||
let disks = disks.clone();
|
||||
let started = manager
|
||||
.clone()
|
||||
.spawn_refresh_if_needed(DataSource::Scheduled, move || async move {
|
||||
calculate_data_dir_used_capacity(&disks)
|
||||
.await
|
||||
.map(|scan| scan.to_capacity_update())
|
||||
.map_err(|e| e.to_string())
|
||||
})
|
||||
.await;
|
||||
|
||||
// Import the calculate function
|
||||
match calculate_data_dir_used_capacity(&disks).await {
|
||||
Ok(new_capacity) => {
|
||||
let elapsed = start.elapsed();
|
||||
info!("Scheduled update completed: {} bytes in {:?}", new_capacity, elapsed);
|
||||
manager.update_capacity(new_capacity, DataSource::Scheduled).await;
|
||||
}
|
||||
Err(e) => {
|
||||
error!("Scheduled update failed: {:?}", e);
|
||||
}
|
||||
if started {
|
||||
debug!("Scheduled capacity refresh started in {:?}", start.elapsed());
|
||||
} else {
|
||||
debug!("Scheduled capacity refresh skipped because another refresh is already in progress");
|
||||
}
|
||||
}
|
||||
});
|
||||
@@ -586,49 +753,49 @@ mod tests {
|
||||
#[serial]
|
||||
fn test_get_scheduled_update_interval() {
|
||||
let interval = get_scheduled_update_interval();
|
||||
assert_eq!(interval, Duration::from_secs(300));
|
||||
assert_eq!(interval, Duration::from_secs(120));
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[serial]
|
||||
fn test_get_write_trigger_delay() {
|
||||
let delay = get_write_trigger_delay();
|
||||
assert_eq!(delay, Duration::from_secs(10));
|
||||
assert_eq!(delay, Duration::from_secs(5));
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[serial]
|
||||
fn test_get_write_frequency_threshold() {
|
||||
let threshold = get_write_frequency_threshold();
|
||||
assert_eq!(threshold, 10);
|
||||
assert_eq!(threshold, 5);
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[serial]
|
||||
fn test_get_fast_update_threshold() {
|
||||
let threshold = get_fast_update_threshold();
|
||||
assert_eq!(threshold, Duration::from_secs(60));
|
||||
assert_eq!(threshold, Duration::from_secs(30));
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[serial]
|
||||
fn test_get_max_files_threshold() {
|
||||
let threshold = get_max_files_threshold();
|
||||
assert_eq!(threshold, 1_000_000);
|
||||
assert_eq!(threshold, 200_000);
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[serial]
|
||||
fn test_get_stat_timeout() {
|
||||
let timeout = get_stat_timeout();
|
||||
assert_eq!(timeout, Duration::from_secs(5));
|
||||
assert_eq!(timeout, Duration::from_secs(3));
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[serial]
|
||||
fn test_get_sample_rate() {
|
||||
let rate = get_sample_rate();
|
||||
assert_eq!(rate, 100);
|
||||
assert_eq!(rate, 200);
|
||||
}
|
||||
|
||||
#[test]
|
||||
@@ -708,7 +875,9 @@ mod tests {
|
||||
async fn test_update_capacity() {
|
||||
let manager = HybridCapacityManager::from_env();
|
||||
|
||||
manager.update_capacity(1000, DataSource::RealTime).await;
|
||||
manager
|
||||
.update_capacity(CapacityUpdate::exact(1000, 0), DataSource::RealTime)
|
||||
.await;
|
||||
|
||||
let cached = manager.get_capacity().await;
|
||||
assert!(cached.is_some());
|
||||
@@ -735,7 +904,9 @@ mod tests {
|
||||
assert!(!manager.needs_fast_update().await);
|
||||
|
||||
// Update cache
|
||||
manager.update_capacity(1000, DataSource::RealTime).await;
|
||||
manager
|
||||
.update_capacity(CapacityUpdate::exact(1000, 0), DataSource::RealTime)
|
||||
.await;
|
||||
|
||||
// Fresh cache, should not need update
|
||||
assert!(!manager.needs_fast_update().await);
|
||||
@@ -747,10 +918,10 @@ mod tests {
|
||||
let config = HybridStrategyConfig::from_env();
|
||||
|
||||
// Check default values
|
||||
assert_eq!(config.scheduled_update_interval, Duration::from_secs(300));
|
||||
assert_eq!(config.write_trigger_delay, Duration::from_secs(10));
|
||||
assert_eq!(config.write_frequency_threshold, 10);
|
||||
assert_eq!(config.fast_update_threshold, Duration::from_secs(60));
|
||||
assert_eq!(config.scheduled_update_interval, Duration::from_secs(120));
|
||||
assert_eq!(config.write_trigger_delay, Duration::from_secs(5));
|
||||
assert_eq!(config.write_frequency_threshold, 5);
|
||||
assert_eq!(config.fast_update_threshold, Duration::from_secs(30));
|
||||
assert!(config.enable_smart_update);
|
||||
assert!(config.enable_write_trigger);
|
||||
}
|
||||
|
||||
@@ -16,9 +16,10 @@
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use crate::capacity::capacity_manager::{DataSource, HybridCapacityManager, HybridStrategyConfig};
|
||||
use crate::capacity::capacity_manager::{CapacityUpdate, DataSource, HybridCapacityManager, HybridStrategyConfig};
|
||||
use serial_test::serial;
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicUsize, Ordering};
|
||||
use std::time::Duration;
|
||||
use tokio::time::sleep;
|
||||
|
||||
@@ -33,17 +34,17 @@ mod tests {
|
||||
async fn test_capacity_update_and_retrieval() {
|
||||
let manager = HybridCapacityManager::from_env();
|
||||
|
||||
// Initially no cache
|
||||
assert!(manager.get_capacity().await.is_none());
|
||||
|
||||
// Update capacity
|
||||
manager.update_capacity(1000, DataSource::RealTime).await;
|
||||
manager
|
||||
.update_capacity(CapacityUpdate::exact(1000, 10), DataSource::RealTime)
|
||||
.await;
|
||||
|
||||
// Retrieve cached value
|
||||
let cached = manager.get_capacity().await;
|
||||
assert!(cached.is_some());
|
||||
let cached = cached.unwrap();
|
||||
assert_eq!(cached.total_used, 1000);
|
||||
assert_eq!(cached.file_count, 10);
|
||||
assert_eq!(cached.source, DataSource::RealTime);
|
||||
assert!(!cached.is_estimated);
|
||||
}
|
||||
@@ -52,7 +53,6 @@ mod tests {
|
||||
async fn test_write_operation_recording() {
|
||||
let manager = HybridCapacityManager::from_env();
|
||||
|
||||
// Record multiple write operations
|
||||
manager.record_write_operation().await;
|
||||
manager.record_write_operation().await;
|
||||
manager.record_write_operation().await;
|
||||
@@ -65,23 +65,17 @@ mod tests {
|
||||
async fn test_fast_update_detection() {
|
||||
let manager = HybridCapacityManager::from_env();
|
||||
|
||||
// No cache, should not need fast update
|
||||
assert!(!manager.needs_fast_update().await);
|
||||
|
||||
// Update cache
|
||||
manager.update_capacity(1000, DataSource::RealTime).await;
|
||||
manager
|
||||
.update_capacity(CapacityUpdate::exact(1000, 1), DataSource::RealTime)
|
||||
.await;
|
||||
|
||||
// Fresh cache, should not need fast update
|
||||
assert!(!manager.needs_fast_update().await);
|
||||
|
||||
// Record write operation
|
||||
manager.record_write_operation().await;
|
||||
|
||||
// Wait for cache to become stale
|
||||
sleep(Duration::from_millis(100)).await;
|
||||
|
||||
// Now cache is stale and there's recent write
|
||||
// Note: This might not trigger due to timing, so we just check it doesn't panic
|
||||
let _needs_update = manager.needs_fast_update().await;
|
||||
}
|
||||
|
||||
@@ -89,22 +83,19 @@ mod tests {
|
||||
async fn test_cache_age_tracking() {
|
||||
let manager = HybridCapacityManager::from_env();
|
||||
|
||||
// No cache, age should be None
|
||||
assert!(manager.get_cache_age().await.is_none());
|
||||
|
||||
// Update cache
|
||||
manager.update_capacity(1000, DataSource::RealTime).await;
|
||||
manager
|
||||
.update_capacity(CapacityUpdate::exact(1000, 1), DataSource::RealTime)
|
||||
.await;
|
||||
|
||||
// Check cache age
|
||||
let age = manager.get_cache_age().await;
|
||||
assert!(age.is_some());
|
||||
let age = age.unwrap();
|
||||
assert!(age < Duration::from_secs(1));
|
||||
|
||||
// Wait a bit
|
||||
sleep(Duration::from_millis(100)).await;
|
||||
|
||||
// Check age again
|
||||
let age = manager.get_cache_age().await.unwrap();
|
||||
assert!(age >= Duration::from_millis(100));
|
||||
}
|
||||
@@ -113,7 +104,6 @@ mod tests {
|
||||
async fn test_data_source_tracking() {
|
||||
let manager = HybridCapacityManager::from_env();
|
||||
|
||||
// Test different data sources
|
||||
let sources = vec![
|
||||
DataSource::RealTime,
|
||||
DataSource::Scheduled,
|
||||
@@ -122,7 +112,7 @@ mod tests {
|
||||
];
|
||||
|
||||
for source in sources {
|
||||
manager.update_capacity(1000, source).await;
|
||||
manager.update_capacity(CapacityUpdate::exact(1000, 1), source).await;
|
||||
let cached = manager.get_capacity().await.unwrap();
|
||||
assert_eq!(cached.source, source);
|
||||
}
|
||||
@@ -132,11 +122,10 @@ mod tests {
|
||||
async fn test_config_from_env() {
|
||||
let config = HybridStrategyConfig::from_env();
|
||||
|
||||
// Check default values
|
||||
assert_eq!(config.scheduled_update_interval, Duration::from_secs(300));
|
||||
assert_eq!(config.write_trigger_delay, Duration::from_secs(10));
|
||||
assert_eq!(config.write_frequency_threshold, 10);
|
||||
assert_eq!(config.fast_update_threshold, Duration::from_secs(60));
|
||||
assert_eq!(config.scheduled_update_interval, Duration::from_secs(120));
|
||||
assert_eq!(config.write_trigger_delay, Duration::from_secs(5));
|
||||
assert_eq!(config.write_frequency_threshold, 5);
|
||||
assert_eq!(config.fast_update_threshold, Duration::from_secs(30));
|
||||
assert!(config.enable_smart_update);
|
||||
assert!(config.enable_write_trigger);
|
||||
}
|
||||
@@ -145,42 +134,34 @@ mod tests {
|
||||
async fn test_write_frequency_window() {
|
||||
let manager = HybridCapacityManager::from_env();
|
||||
|
||||
// Record many write operations
|
||||
for _ in 0..20 {
|
||||
manager.record_write_operation().await;
|
||||
}
|
||||
|
||||
// Check frequency (should be 20 since all are within 1 minute)
|
||||
let frequency = manager.get_write_frequency().await;
|
||||
assert_eq!(frequency, 20);
|
||||
|
||||
// Note: In a real test, we would wait for the window to expire
|
||||
// and verify that old writes are removed
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn test_concurrent_access() {
|
||||
let manager = Arc::new(HybridCapacityManager::from_env());
|
||||
|
||||
// Simulate concurrent updates
|
||||
let mut handles = vec![];
|
||||
|
||||
for i in 0..10 {
|
||||
let mgr = manager.clone();
|
||||
let handle = tokio::spawn(async move {
|
||||
mgr.update_capacity(i as u64 * 100, DataSource::RealTime).await;
|
||||
mgr.update_capacity(CapacityUpdate::exact(i as u64 * 100, i), DataSource::RealTime)
|
||||
.await;
|
||||
mgr.record_write_operation().await;
|
||||
});
|
||||
handles.push(handle);
|
||||
}
|
||||
|
||||
// Wait for all tasks to complete
|
||||
for handle in handles {
|
||||
handle.await.unwrap();
|
||||
}
|
||||
|
||||
// Verify final state
|
||||
let cached = manager.get_capacity().await;
|
||||
assert!(cached.is_some());
|
||||
|
||||
@@ -192,21 +173,95 @@ mod tests {
|
||||
#[serial]
|
||||
async fn test_performance_overhead() {
|
||||
let manager = Arc::new(HybridCapacityManager::from_env());
|
||||
|
||||
// Measure time for 1000 operations
|
||||
let start = std::time::Instant::now();
|
||||
|
||||
for i in 0..1000 {
|
||||
manager.update_capacity(i as u64, DataSource::RealTime).await;
|
||||
manager
|
||||
.update_capacity(CapacityUpdate::exact(i as u64, i), DataSource::RealTime)
|
||||
.await;
|
||||
manager.record_write_operation().await;
|
||||
let _ = manager.get_capacity().await;
|
||||
}
|
||||
|
||||
let elapsed = start.elapsed();
|
||||
|
||||
// Should complete in less than 1 second
|
||||
assert!(elapsed < Duration::from_secs(1));
|
||||
|
||||
println!("1000 operations completed in {:?}", elapsed);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_refresh_or_join_singleflight() {
|
||||
let manager = Arc::new(HybridCapacityManager::from_env());
|
||||
let calls = Arc::new(AtomicUsize::new(0));
|
||||
|
||||
let mgr1 = manager.clone();
|
||||
let calls1 = calls.clone();
|
||||
let first = tokio::spawn(async move {
|
||||
mgr1.refresh_or_join(DataSource::Scheduled, move || async move {
|
||||
calls1.fetch_add(1, Ordering::SeqCst);
|
||||
sleep(Duration::from_millis(50)).await;
|
||||
Ok(CapacityUpdate::exact(2048, 8))
|
||||
})
|
||||
.await
|
||||
});
|
||||
|
||||
sleep(Duration::from_millis(10)).await;
|
||||
|
||||
let mgr2 = manager.clone();
|
||||
let calls2 = calls.clone();
|
||||
let second = tokio::spawn(async move {
|
||||
mgr2.refresh_or_join(DataSource::WriteTriggered, move || async move {
|
||||
calls2.fetch_add(1, Ordering::SeqCst);
|
||||
Ok(CapacityUpdate::exact(4096, 16))
|
||||
})
|
||||
.await
|
||||
});
|
||||
|
||||
let first = first.await.unwrap().unwrap();
|
||||
let second = second.await.unwrap().unwrap();
|
||||
|
||||
assert_eq!(calls.load(Ordering::SeqCst), 1);
|
||||
assert_eq!(first.total_used, 2048);
|
||||
assert_eq!(second.total_used, 2048);
|
||||
let cached = manager.get_capacity().await.unwrap();
|
||||
assert_eq!(cached.total_used, 2048);
|
||||
assert_eq!(cached.file_count, 8);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_spawn_refresh_if_needed_deduplicates_background_refresh() {
|
||||
let manager = Arc::new(HybridCapacityManager::from_env());
|
||||
let calls = Arc::new(AtomicUsize::new(0));
|
||||
|
||||
let first_manager = manager.clone();
|
||||
let first_calls = calls.clone();
|
||||
let started = first_manager
|
||||
.clone()
|
||||
.spawn_refresh_if_needed(DataSource::Scheduled, move || async move {
|
||||
first_calls.fetch_add(1, Ordering::SeqCst);
|
||||
sleep(Duration::from_millis(50)).await;
|
||||
Ok(CapacityUpdate::estimated(8192, 32))
|
||||
})
|
||||
.await;
|
||||
assert!(started);
|
||||
|
||||
let second_manager = manager.clone();
|
||||
let second_calls = calls.clone();
|
||||
let started = second_manager
|
||||
.clone()
|
||||
.spawn_refresh_if_needed(DataSource::Scheduled, move || async move {
|
||||
second_calls.fetch_add(1, Ordering::SeqCst);
|
||||
Ok(CapacityUpdate::exact(1, 1))
|
||||
})
|
||||
.await;
|
||||
assert!(!started);
|
||||
|
||||
sleep(Duration::from_millis(100)).await;
|
||||
|
||||
assert_eq!(calls.load(Ordering::SeqCst), 1);
|
||||
assert!(!manager.refresh_in_progress().await);
|
||||
let cached = manager.get_capacity().await.unwrap();
|
||||
assert_eq!(cached.total_used, 8192);
|
||||
assert!(cached.is_estimated);
|
||||
}
|
||||
}
|
||||
|
||||
@@ -1,465 +0,0 @@
|
||||
// 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.
|
||||
|
||||
//! Capacity Metrics for monitoring
|
||||
|
||||
use metrics::{counter, gauge, histogram};
|
||||
use std::sync::Arc;
|
||||
use std::sync::atomic::{AtomicU64, Ordering};
|
||||
use std::time::Duration;
|
||||
use tracing::info;
|
||||
|
||||
// ============================================================================
|
||||
// Metric Name Constants (following existing naming convention)
|
||||
// ============================================================================
|
||||
|
||||
/// Cache hit counter
|
||||
const CAPACITY_CACHE_HIT: &str = "rustfs.capacity.cache.hits";
|
||||
|
||||
/// Cache miss counter
|
||||
const CAPACITY_CACHE_MISS: &str = "rustfs.capacity.cache.misses";
|
||||
|
||||
/// Cache hit rate gauge
|
||||
const CAPACITY_CACHE_HIT_RATE: &str = "rustfs.capacity.cache.hit_rate";
|
||||
|
||||
/// Cache hits total gauge
|
||||
const CAPACITY_CACHE_HITS_TOTAL: &str = "rustfs.capacity.cache.hits_total";
|
||||
|
||||
/// Cache misses total gauge
|
||||
const CAPACITY_CACHE_MISSES_TOTAL: &str = "rustfs.capacity.cache.misses_total";
|
||||
|
||||
/// Scheduled update counter
|
||||
const CAPACITY_UPDATE_SCHEDULED: &str = "rustfs.capacity.update.scheduled";
|
||||
|
||||
/// Write-triggered update counter
|
||||
const CAPACITY_UPDATE_WRITE_TRIGGERED: &str = "rustfs.capacity.update.write_triggered";
|
||||
|
||||
/// Update failure counter
|
||||
const CAPACITY_UPDATE_FAILURES: &str = "rustfs.capacity.update.failures";
|
||||
|
||||
/// Current capacity in bytes gauge
|
||||
#[allow(dead_code)]
|
||||
const CAPACITY_CURRENT_BYTES: &str = "rustfs.capacity.current";
|
||||
|
||||
/// Write operations counter
|
||||
const CAPACITY_WRITE_OPERATIONS: &str = "rustfs.capacity.write.operations";
|
||||
|
||||
/// Write frequency gauge
|
||||
#[allow(dead_code)]
|
||||
const CAPACITY_WRITE_FREQUENCY: &str = "rustfs.capacity.write.frequency";
|
||||
|
||||
/// Update duration in microseconds histogram
|
||||
const CAPACITY_UPDATE_DURATION_US: &str = "rustfs.capacity.update.duration_us";
|
||||
|
||||
/// Scheduled updates total gauge
|
||||
const CAPACITY_UPDATE_SCHEDULED_TOTAL: &str = "rustfs.capacity.update.scheduled_total";
|
||||
|
||||
/// Write-triggered updates total gauge
|
||||
const CAPACITY_UPDATE_WRITE_TRIGGERED_TOTAL: &str = "rustfs.capacity.update.write_triggered_total";
|
||||
|
||||
/// Update failures total gauge
|
||||
const CAPACITY_UPDATE_FAILURES_TOTAL: &str = "rustfs.capacity.update.failures_total";
|
||||
|
||||
/// Symlinks encountered counter
|
||||
const CAPACITY_SYMLINKS_ENCOUNTERED: &str = "rustfs.capacity.symlinks.encountered";
|
||||
|
||||
/// Symlinks total size gauge
|
||||
const CAPACITY_SYMLINKS_SIZE: &str = "rustfs.capacity.symlinks.total_size";
|
||||
|
||||
/// Symlinks count gauge
|
||||
const CAPACITY_SYMLINKS_COUNT: &str = "rustfs.capacity.symlinks.count";
|
||||
|
||||
/// Dynamic timeout counter
|
||||
const CAPACITY_TIMEOUT_DYNAMIC: &str = "rustfs.capacity.timeout.dynamic";
|
||||
|
||||
/// Timeout fallback counter
|
||||
const CAPACITY_TIMEOUT_FALLBACK: &str = "rustfs.capacity.timeout.fallback";
|
||||
|
||||
/// Stall detected counter
|
||||
const CAPACITY_TIMEOUT_STALL: &str = "rustfs.capacity.timeout.stall";
|
||||
|
||||
/// Dynamic timeout total gauge
|
||||
const CAPACITY_TIMEOUT_DYNAMIC_TOTAL: &str = "rustfs.capacity.timeout.dynamic_total";
|
||||
|
||||
/// Timeout fallback total gauge
|
||||
const CAPACITY_TIMEOUT_FALLBACK_TOTAL: &str = "rustfs.capacity.timeout.fallback_total";
|
||||
|
||||
/// Stall detected total gauge
|
||||
const CAPACITY_TIMEOUT_STALL_TOTAL: &str = "rustfs.capacity.timeout.stall_total";
|
||||
|
||||
// ============================================================================
|
||||
// Capacity Metrics
|
||||
// ============================================================================
|
||||
|
||||
/// Capacity metrics for monitoring
|
||||
#[derive(Debug, Default)]
|
||||
pub struct CapacityMetrics {
|
||||
/// Cache hit count
|
||||
pub cache_hits: AtomicU64,
|
||||
/// Cache miss count
|
||||
pub cache_misses: AtomicU64,
|
||||
/// Scheduled update count
|
||||
pub scheduled_updates: AtomicU64,
|
||||
/// Write triggered update count
|
||||
pub write_triggered_updates: AtomicU64,
|
||||
/// Update failure count
|
||||
pub update_failures: AtomicU64,
|
||||
/// Total update duration in microseconds
|
||||
pub total_update_duration_us: AtomicU64,
|
||||
/// Update count for average calculation
|
||||
pub update_count: AtomicU64,
|
||||
/// Symlink count encountered during capacity calculation
|
||||
pub symlink_count: AtomicU64,
|
||||
/// Total size of symlink targets
|
||||
pub symlink_size: AtomicU64,
|
||||
/// Dynamic timeout usage count
|
||||
pub dynamic_timeout_count: AtomicU64,
|
||||
/// Timeout fallback to sampling count
|
||||
pub timeout_fallback_count: AtomicU64,
|
||||
/// Stall detection count
|
||||
pub stall_detected_count: AtomicU64,
|
||||
}
|
||||
|
||||
impl CapacityMetrics {
|
||||
/// Create new metrics
|
||||
pub fn new() -> Self {
|
||||
Self::default()
|
||||
}
|
||||
|
||||
/// Record cache hit
|
||||
pub fn record_cache_hit(&self) {
|
||||
self.cache_hits.fetch_add(1, Ordering::Relaxed);
|
||||
counter!(CAPACITY_CACHE_HIT).increment(1);
|
||||
}
|
||||
|
||||
/// Record cache miss
|
||||
pub fn record_cache_miss(&self) {
|
||||
self.cache_misses.fetch_add(1, Ordering::Relaxed);
|
||||
counter!(CAPACITY_CACHE_MISS).increment(1);
|
||||
}
|
||||
|
||||
/// Record scheduled update
|
||||
#[allow(dead_code)]
|
||||
pub fn record_scheduled_update(&self) {
|
||||
self.scheduled_updates.fetch_add(1, Ordering::Relaxed);
|
||||
counter!(CAPACITY_UPDATE_SCHEDULED).increment(1);
|
||||
}
|
||||
|
||||
/// Record write triggered update
|
||||
#[allow(dead_code)]
|
||||
pub fn record_write_triggered_update(&self) {
|
||||
self.write_triggered_updates.fetch_add(1, Ordering::Relaxed);
|
||||
counter!(CAPACITY_UPDATE_WRITE_TRIGGERED).increment(1);
|
||||
}
|
||||
|
||||
/// Record update failure
|
||||
#[allow(dead_code)]
|
||||
pub fn record_update_failure(&self) {
|
||||
self.update_failures.fetch_add(1, Ordering::Relaxed);
|
||||
counter!(CAPACITY_UPDATE_FAILURES).increment(1);
|
||||
}
|
||||
|
||||
/// Record write operation
|
||||
#[allow(dead_code)]
|
||||
pub fn record_write_operation(&self) {
|
||||
counter!(CAPACITY_WRITE_OPERATIONS).increment(1);
|
||||
}
|
||||
|
||||
/// Record symlink encountered
|
||||
pub fn record_symlink(&self, size: u64) {
|
||||
self.symlink_count.fetch_add(1, Ordering::Relaxed);
|
||||
let total_size = self.symlink_size.fetch_add(size, Ordering::Relaxed) + size;
|
||||
counter!(CAPACITY_SYMLINKS_ENCOUNTERED).increment(1);
|
||||
gauge!(CAPACITY_SYMLINKS_SIZE).set(total_size as f64);
|
||||
}
|
||||
|
||||
/// Record dynamic timeout usage
|
||||
pub fn record_dynamic_timeout(&self) {
|
||||
self.dynamic_timeout_count.fetch_add(1, Ordering::Relaxed);
|
||||
counter!(CAPACITY_TIMEOUT_DYNAMIC).increment(1);
|
||||
}
|
||||
|
||||
/// Record timeout fallback to sampling
|
||||
pub fn record_timeout_fallback(&self) {
|
||||
self.timeout_fallback_count.fetch_add(1, Ordering::Relaxed);
|
||||
counter!(CAPACITY_TIMEOUT_FALLBACK).increment(1);
|
||||
}
|
||||
|
||||
/// Record stall detection
|
||||
pub fn record_stall_detected(&self) {
|
||||
self.stall_detected_count.fetch_add(1, Ordering::Relaxed);
|
||||
counter!(CAPACITY_TIMEOUT_STALL).increment(1);
|
||||
}
|
||||
|
||||
/// Get symlink statistics
|
||||
#[allow(dead_code)]
|
||||
pub fn get_symlink_stats(&self) -> (u64, u64) {
|
||||
(self.symlink_count.load(Ordering::Relaxed), self.symlink_size.load(Ordering::Relaxed))
|
||||
}
|
||||
|
||||
/// Get timeout statistics
|
||||
#[allow(dead_code)]
|
||||
pub fn get_timeout_stats(&self) -> (u64, u64, u64) {
|
||||
(
|
||||
self.dynamic_timeout_count.load(Ordering::Relaxed),
|
||||
self.timeout_fallback_count.load(Ordering::Relaxed),
|
||||
self.stall_detected_count.load(Ordering::Relaxed),
|
||||
)
|
||||
}
|
||||
|
||||
/// Record update duration
|
||||
#[allow(dead_code)]
|
||||
pub fn record_update_duration(&self, duration: Duration) {
|
||||
let duration_us = duration.as_micros() as u64;
|
||||
self.total_update_duration_us.fetch_add(duration_us, Ordering::Relaxed);
|
||||
self.update_count.fetch_add(1, Ordering::Relaxed);
|
||||
|
||||
histogram!(CAPACITY_UPDATE_DURATION_US).record(duration_us as f64);
|
||||
}
|
||||
|
||||
/// Get cache hit rate
|
||||
pub fn get_cache_hit_rate(&self) -> f64 {
|
||||
let hits = self.cache_hits.load(Ordering::Relaxed);
|
||||
let misses = self.cache_misses.load(Ordering::Relaxed);
|
||||
let total = hits + misses;
|
||||
if total == 0 { 0.0 } else { hits as f64 / total as f64 }
|
||||
}
|
||||
|
||||
/// Get average update duration
|
||||
pub fn get_avg_update_duration(&self) -> Duration {
|
||||
let total_us = self.total_update_duration_us.load(Ordering::Relaxed);
|
||||
let count = self.update_count.load(Ordering::Relaxed);
|
||||
if count == 0 {
|
||||
Duration::from_secs(0)
|
||||
} else {
|
||||
Duration::from_micros(total_us / count)
|
||||
}
|
||||
}
|
||||
|
||||
/// Get metrics summary
|
||||
pub fn get_summary(&self) -> MetricsSummary {
|
||||
MetricsSummary {
|
||||
cache_hits: self.cache_hits.load(Ordering::Relaxed),
|
||||
cache_misses: self.cache_misses.load(Ordering::Relaxed),
|
||||
cache_hit_rate: self.get_cache_hit_rate(),
|
||||
scheduled_updates: self.scheduled_updates.load(Ordering::Relaxed),
|
||||
write_triggered_updates: self.write_triggered_updates.load(Ordering::Relaxed),
|
||||
update_failures: self.update_failures.load(Ordering::Relaxed),
|
||||
avg_update_duration: self.get_avg_update_duration(),
|
||||
symlink_count: self.symlink_count.load(Ordering::Relaxed),
|
||||
symlink_size: self.symlink_size.load(Ordering::Relaxed),
|
||||
dynamic_timeout_count: self.dynamic_timeout_count.load(Ordering::Relaxed),
|
||||
timeout_fallback_count: self.timeout_fallback_count.load(Ordering::Relaxed),
|
||||
stall_detected_count: self.stall_detected_count.load(Ordering::Relaxed),
|
||||
}
|
||||
}
|
||||
|
||||
/// Log metrics summary
|
||||
pub fn log_summary(&self) {
|
||||
let summary = self.get_summary();
|
||||
|
||||
// Update gauges for current values using constant names
|
||||
gauge!(CAPACITY_CACHE_HIT_RATE).set(summary.cache_hit_rate);
|
||||
gauge!(CAPACITY_CACHE_HITS_TOTAL).set(summary.cache_hits as f64);
|
||||
gauge!(CAPACITY_CACHE_MISSES_TOTAL).set(summary.cache_misses as f64);
|
||||
gauge!(CAPACITY_UPDATE_SCHEDULED_TOTAL).set(summary.scheduled_updates as f64);
|
||||
gauge!(CAPACITY_UPDATE_WRITE_TRIGGERED_TOTAL).set(summary.write_triggered_updates as f64);
|
||||
gauge!(CAPACITY_UPDATE_FAILURES_TOTAL).set(summary.update_failures as f64);
|
||||
gauge!(CAPACITY_SYMLINKS_COUNT).set(summary.symlink_count as f64);
|
||||
gauge!(CAPACITY_SYMLINKS_SIZE).set(summary.symlink_size as f64);
|
||||
gauge!(CAPACITY_TIMEOUT_DYNAMIC_TOTAL).set(summary.dynamic_timeout_count as f64);
|
||||
gauge!(CAPACITY_TIMEOUT_FALLBACK_TOTAL).set(summary.timeout_fallback_count as f64);
|
||||
gauge!(CAPACITY_TIMEOUT_STALL_TOTAL).set(summary.stall_detected_count as f64);
|
||||
|
||||
info!(
|
||||
"Capacity Metrics: cache_hit_rate={:.2}%, cache_hits={}, cache_misses={}, scheduled_updates={}, write_triggered_updates={}, update_failures={}, avg_update_duration={:?}, symlinks={}, symlink_size={}, dynamic_timeouts={}, timeout_fallbacks={}, stalls={}",
|
||||
summary.cache_hit_rate * 100.0,
|
||||
summary.cache_hits,
|
||||
summary.cache_misses,
|
||||
summary.scheduled_updates,
|
||||
summary.write_triggered_updates,
|
||||
summary.update_failures,
|
||||
summary.avg_update_duration,
|
||||
summary.symlink_count,
|
||||
summary.symlink_size,
|
||||
summary.dynamic_timeout_count,
|
||||
summary.timeout_fallback_count,
|
||||
summary.stall_detected_count
|
||||
);
|
||||
}
|
||||
}
|
||||
|
||||
/// Metrics summary
|
||||
#[derive(Debug, Clone)]
|
||||
pub struct MetricsSummary {
|
||||
pub cache_hits: u64,
|
||||
pub cache_misses: u64,
|
||||
pub cache_hit_rate: f64,
|
||||
pub scheduled_updates: u64,
|
||||
pub write_triggered_updates: u64,
|
||||
pub update_failures: u64,
|
||||
pub avg_update_duration: Duration,
|
||||
pub symlink_count: u64,
|
||||
pub symlink_size: u64,
|
||||
pub dynamic_timeout_count: u64,
|
||||
pub timeout_fallback_count: u64,
|
||||
pub stall_detected_count: u64,
|
||||
}
|
||||
|
||||
/// Global metrics instance
|
||||
static CAPACITY_METRICS: std::sync::OnceLock<Arc<CapacityMetrics>> = std::sync::OnceLock::new();
|
||||
|
||||
/// Get global metrics
|
||||
pub fn get_capacity_metrics() -> Arc<CapacityMetrics> {
|
||||
CAPACITY_METRICS.get_or_init(|| Arc::new(CapacityMetrics::new())).clone()
|
||||
}
|
||||
|
||||
/// Start metrics logging task
|
||||
pub async fn start_metrics_logging(interval: Duration) {
|
||||
let metrics = get_capacity_metrics();
|
||||
|
||||
tokio::spawn(async move {
|
||||
let mut timer = tokio::time::interval(interval);
|
||||
|
||||
loop {
|
||||
timer.tick().await;
|
||||
metrics.log_summary();
|
||||
}
|
||||
});
|
||||
}
|
||||
|
||||
/// Record a write operation globally
|
||||
#[allow(dead_code)]
|
||||
pub fn record_global_write_operation() {
|
||||
let metrics = get_capacity_metrics();
|
||||
metrics.record_write_operation();
|
||||
}
|
||||
|
||||
/// Record cache hit globally
|
||||
#[allow(dead_code)]
|
||||
pub fn record_global_cache_hit() {
|
||||
let metrics = get_capacity_metrics();
|
||||
metrics.record_cache_hit();
|
||||
}
|
||||
|
||||
/// Record cache miss globally
|
||||
#[allow(dead_code)]
|
||||
pub fn record_global_cache_miss() {
|
||||
let metrics = get_capacity_metrics();
|
||||
metrics.record_cache_miss();
|
||||
}
|
||||
|
||||
// ============================================================================
|
||||
// Tests
|
||||
// ============================================================================
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use super::*;
|
||||
|
||||
#[test]
|
||||
fn test_metrics_creation() {
|
||||
let metrics = CapacityMetrics::new();
|
||||
assert_eq!(metrics.cache_hits.load(Ordering::Relaxed), 0);
|
||||
assert_eq!(metrics.cache_misses.load(Ordering::Relaxed), 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_record_cache_hit() {
|
||||
let metrics = CapacityMetrics::new();
|
||||
metrics.record_cache_hit();
|
||||
metrics.record_cache_hit();
|
||||
assert_eq!(metrics.cache_hits.load(Ordering::Relaxed), 2);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_cache_hit_rate() {
|
||||
let metrics = CapacityMetrics::new();
|
||||
metrics.record_cache_hit();
|
||||
metrics.record_cache_hit();
|
||||
metrics.record_cache_miss();
|
||||
|
||||
let rate = metrics.get_cache_hit_rate();
|
||||
assert!((rate - 0.6666666666666666).abs() < 0.0001);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_avg_update_duration() {
|
||||
let metrics = CapacityMetrics::new();
|
||||
metrics.record_update_duration(Duration::from_millis(100));
|
||||
metrics.record_update_duration(Duration::from_millis(200));
|
||||
|
||||
let avg = metrics.get_avg_update_duration();
|
||||
assert_eq!(avg, Duration::from_millis(150));
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_get_summary() {
|
||||
let metrics = CapacityMetrics::new();
|
||||
metrics.record_cache_hit();
|
||||
metrics.record_scheduled_update();
|
||||
metrics.record_update_duration(Duration::from_millis(100));
|
||||
|
||||
let summary = metrics.get_summary();
|
||||
assert_eq!(summary.cache_hits, 1);
|
||||
assert_eq!(summary.scheduled_updates, 1);
|
||||
assert_eq!(summary.avg_update_duration, Duration::from_millis(100));
|
||||
assert_eq!(summary.symlink_count, 0);
|
||||
assert_eq!(summary.dynamic_timeout_count, 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_record_write_operation() {
|
||||
let metrics = CapacityMetrics::new();
|
||||
metrics.record_write_operation();
|
||||
metrics.record_write_operation();
|
||||
// This test just ensures the method doesn't panic
|
||||
assert_eq!(metrics.write_triggered_updates.load(Ordering::Relaxed), 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_record_symlink() {
|
||||
let metrics = CapacityMetrics::new();
|
||||
metrics.record_symlink(1024);
|
||||
metrics.record_symlink(2048);
|
||||
|
||||
let (count, size) = metrics.get_symlink_stats();
|
||||
assert_eq!(count, 2);
|
||||
assert_eq!(size, 3072);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_record_dynamic_timeout() {
|
||||
let metrics = CapacityMetrics::new();
|
||||
metrics.record_dynamic_timeout();
|
||||
metrics.record_dynamic_timeout();
|
||||
|
||||
let (dynamic, fallback, stalls) = metrics.get_timeout_stats();
|
||||
assert_eq!(dynamic, 2);
|
||||
assert_eq!(fallback, 0);
|
||||
assert_eq!(stalls, 0);
|
||||
}
|
||||
|
||||
#[test]
|
||||
fn test_record_timeout_fallback() {
|
||||
let metrics = CapacityMetrics::new();
|
||||
metrics.record_timeout_fallback();
|
||||
metrics.record_stall_detected();
|
||||
|
||||
let (dynamic, fallback, stalls) = metrics.get_timeout_stats();
|
||||
assert_eq!(dynamic, 0);
|
||||
assert_eq!(fallback, 1);
|
||||
assert_eq!(stalls, 1);
|
||||
}
|
||||
}
|
||||
+15
-24
@@ -18,24 +18,24 @@
|
||||
//! - Scheduled background updates (configurable interval)
|
||||
//! - Write-triggered updates for high-frequency write scenarios
|
||||
//! - Configurable caching thresholds and smart update strategies
|
||||
//! - Comprehensive metrics collection for monitoring
|
||||
//! - Capacity metrics emitted through `rustfs-io-metrics`
|
||||
//!
|
||||
//! ## Configuration
|
||||
//!
|
||||
//! All configuration is via environment variables (see `rustfs_config`):
|
||||
//! - `RUSTFS_CAPACITY_SCHEDULED_INTERVAL` - Update interval in seconds (default: 300)
|
||||
//! - `RUSTFS_CAPACITY_WRITE_TRIGGER_DELAY` - Write trigger delay (default: 10s)
|
||||
//! - `RUSTFS_CAPACITY_WRITE_FREQUENCY_THRESHOLD` - Write frequency threshold (default: 10 writes/min)
|
||||
//! - `RUSTFS_CAPACITY_FAST_UPDATE_THRESHOLD` - Fast update threshold (default: 60s)
|
||||
//! - `RUSTFS_CAPACITY_MAX_FILES_THRESHOLD` - Max files before sampling (default: 1,000,000)
|
||||
//! - `RUSTFS_CAPACITY_STAT_TIMEOUT` - Stat operation timeout (default: 5s)
|
||||
//! - `RUSTFS_CAPACITY_SAMPLE_RATE` - Sampling rate for metrics (default: 100)
|
||||
//! - `RUSTFS_CAPACITY_SCHEDULED_INTERVAL` - Update interval in seconds (default: 120)
|
||||
//! - `RUSTFS_CAPACITY_WRITE_TRIGGER_DELAY` - Write trigger delay (default: 5s)
|
||||
//! - `RUSTFS_CAPACITY_WRITE_FREQUENCY_THRESHOLD` - Write frequency threshold (default: 5 writes/min)
|
||||
//! - `RUSTFS_CAPACITY_FAST_UPDATE_THRESHOLD` - Fast update threshold (default: 30s)
|
||||
//! - `RUSTFS_CAPACITY_MAX_FILES_THRESHOLD` - Max files before sampling (default: 200,000)
|
||||
//! - `RUSTFS_CAPACITY_STAT_TIMEOUT` - Stat operation timeout (default: 3s)
|
||||
//! - `RUSTFS_CAPACITY_SAMPLE_RATE` - Sampling rate for metrics (default: 200)
|
||||
//! - `RUSTFS_CAPACITY_FOLLOW_SYMLINKS` - Follow symlinks during traversal (default: false)
|
||||
//! - `RUSTFS_CAPACITY_MAX_SYMLINK_DEPTH` - Max symlink depth (default: 8)
|
||||
//! - `RUSTFS_CAPACITY_ENABLE_DYNAMIC_TIMEOUT` - Enable dynamic timeout (default: false)
|
||||
//! - `RUSTFS_CAPACITY_MIN_TIMEOUT` - Minimum timeout (default: 1s)
|
||||
//! - `RUSTFS_CAPACITY_MAX_TIMEOUT` - Maximum timeout (default: 300s)
|
||||
//! - `RUSTFS_CAPACITY_STALL_TIMEOUT` - Stall detection timeout (default: 30s)
|
||||
//! - `RUSTFS_CAPACITY_MAX_SYMLINK_DEPTH` - Max symlink depth (default: 3)
|
||||
//! - `RUSTFS_CAPACITY_ENABLE_DYNAMIC_TIMEOUT` - Enable dynamic timeout (default: true)
|
||||
//! - `RUSTFS_CAPACITY_MIN_TIMEOUT` - Minimum timeout (default: 2s)
|
||||
//! - `RUSTFS_CAPACITY_MAX_TIMEOUT` - Maximum timeout (default: 15s)
|
||||
//! - `RUSTFS_CAPACITY_STALL_TIMEOUT` - Stall detection timeout (default: 20s)
|
||||
//!
|
||||
//! ## Architecture
|
||||
//!
|
||||
@@ -45,16 +45,8 @@
|
||||
//! 3. **Cached responses**: Returns cached data when fresh
|
||||
//! 4. **Timeout protection**: Dynamic timeouts prevent hangs on large directories
|
||||
//!
|
||||
//! ## Metrics
|
||||
//!
|
||||
//! Metrics are automatically recorded via the `metrics` crate and accessible
|
||||
//! through the `rustfs-metrics` collection system. Key metrics include:
|
||||
//! - `rustfs.capacity.cache.{hits,misses}` - Cache hit/miss tracking
|
||||
//! - `rustfs.capacity.current` - Current capacity in bytes
|
||||
//! - `rustfs.capacity.write.operations` - Write operation count
|
||||
//! - `rustfs.capacity.update.{scheduled,write_triggered,failures}` - Update statistics
|
||||
//! - `rustfs.capacity.symlinks.*` - Symlink tracking statistics
|
||||
//! - `rustfs.capacity.timeout.*` - Timeout and stall detection
|
||||
//! Capacity metrics flow through the existing observability pipeline via the `metrics`
|
||||
//! crate and `rustfs-io-metrics`; this module does not expose a Prometheus HTTP endpoint.
|
||||
//!
|
||||
//! ## Testing
|
||||
//!
|
||||
@@ -73,6 +65,5 @@ pub mod capacity_integration;
|
||||
pub mod capacity_manager;
|
||||
#[cfg(test)]
|
||||
mod capacity_manager_test;
|
||||
pub mod capacity_metrics;
|
||||
#[cfg(test)]
|
||||
mod write_trigger_test;
|
||||
|
||||
@@ -16,10 +16,7 @@
|
||||
|
||||
#[cfg(test)]
|
||||
mod tests {
|
||||
use crate::capacity::capacity_manager::{DataSource, HybridCapacityManager};
|
||||
use crate::capacity::capacity_metrics::{
|
||||
CapacityMetrics, get_capacity_metrics, record_global_cache_hit, record_global_cache_miss, record_global_write_operation,
|
||||
};
|
||||
use crate::capacity::capacity_manager::{CapacityUpdate, DataSource, HybridCapacityManager};
|
||||
use serial_test::serial;
|
||||
use std::time::Duration;
|
||||
|
||||
@@ -27,93 +24,45 @@ mod tests {
|
||||
#[serial]
|
||||
async fn test_write_trigger_integration() {
|
||||
let manager = HybridCapacityManager::from_env();
|
||||
let metrics = CapacityMetrics::new();
|
||||
|
||||
// Record write operations
|
||||
manager.record_write_operation().await;
|
||||
manager.record_write_operation().await;
|
||||
manager.record_write_operation().await;
|
||||
|
||||
// Check write frequency
|
||||
let frequency = manager.get_write_frequency().await;
|
||||
assert_eq!(frequency, 3);
|
||||
|
||||
// Check metrics
|
||||
let summary = metrics.get_summary();
|
||||
assert_eq!(summary.write_triggered_updates, 0); // Not triggered yet
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn test_write_trigger_with_capacity_update() {
|
||||
let manager = HybridCapacityManager::from_env();
|
||||
let metrics = CapacityMetrics::new();
|
||||
|
||||
// Simulate write-triggered update by calling metrics directly
|
||||
metrics.record_write_triggered_update();
|
||||
manager
|
||||
.update_capacity(CapacityUpdate::exact(1000, 4), DataSource::WriteTriggered)
|
||||
.await;
|
||||
|
||||
// Check metrics
|
||||
let summary = metrics.get_summary();
|
||||
assert_eq!(summary.write_triggered_updates, 1);
|
||||
|
||||
// Also test manager update
|
||||
manager.update_capacity(1000, DataSource::WriteTriggered).await;
|
||||
|
||||
// Check capacity
|
||||
let cached = manager.get_capacity().await;
|
||||
assert!(cached.is_some());
|
||||
assert_eq!(cached.unwrap().total_used, 1000);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
#[serial]
|
||||
async fn test_metrics_recording() {
|
||||
let metrics = CapacityMetrics::new();
|
||||
|
||||
// Record various operations
|
||||
metrics.record_cache_hit();
|
||||
metrics.record_cache_hit();
|
||||
metrics.record_cache_miss();
|
||||
|
||||
metrics.record_scheduled_update();
|
||||
metrics.record_write_triggered_update();
|
||||
|
||||
metrics.record_update_duration(Duration::from_millis(100));
|
||||
metrics.record_update_duration(Duration::from_millis(200));
|
||||
|
||||
// Check summary
|
||||
let summary = metrics.get_summary();
|
||||
assert_eq!(summary.cache_hits, 2);
|
||||
assert_eq!(summary.cache_misses, 1);
|
||||
assert_eq!(summary.scheduled_updates, 1);
|
||||
assert_eq!(summary.write_triggered_updates, 1);
|
||||
assert_eq!(summary.avg_update_duration, Duration::from_millis(150));
|
||||
|
||||
// Check hit rate
|
||||
let hit_rate = metrics.get_cache_hit_rate();
|
||||
assert!((hit_rate - 0.6666666666666666).abs() < 0.0001);
|
||||
let cached = cached.unwrap();
|
||||
assert_eq!(cached.total_used, 1000);
|
||||
assert_eq!(cached.file_count, 4);
|
||||
assert_eq!(cached.source, DataSource::WriteTriggered);
|
||||
}
|
||||
|
||||
#[tokio::test]
|
||||
async fn test_write_frequency_tracking() {
|
||||
let manager = HybridCapacityManager::from_env();
|
||||
|
||||
// Initial state
|
||||
assert_eq!(manager.get_write_frequency().await, 0);
|
||||
|
||||
// Record writes
|
||||
for _ in 0..5 {
|
||||
manager.record_write_operation().await;
|
||||
}
|
||||
|
||||
// Check frequency
|
||||
assert_eq!(manager.get_write_frequency().await, 5);
|
||||
|
||||
// Wait for window to expire (60 seconds)
|
||||
// In real tests, we'd use a shorter window
|
||||
tokio::time::sleep(Duration::from_millis(10)).await;
|
||||
|
||||
// Frequency should still be 5 (window not expired)
|
||||
assert_eq!(manager.get_write_frequency().await, 5);
|
||||
}
|
||||
|
||||
@@ -121,37 +70,18 @@ mod tests {
|
||||
async fn test_needs_fast_update() {
|
||||
let manager = HybridCapacityManager::from_env();
|
||||
|
||||
// No cache, should not need update
|
||||
assert!(!manager.needs_fast_update().await);
|
||||
|
||||
// Update cache
|
||||
manager.update_capacity(1000, DataSource::Scheduled).await;
|
||||
manager
|
||||
.update_capacity(CapacityUpdate::exact(1000, 1), DataSource::Scheduled)
|
||||
.await;
|
||||
|
||||
// Fresh cache, should not need update
|
||||
assert!(!manager.needs_fast_update().await);
|
||||
|
||||
// Record write operation
|
||||
manager.record_write_operation().await;
|
||||
|
||||
// With recent write, should need fast update
|
||||
// (depending on configuration, this may or may not trigger)
|
||||
let needs_update = manager.needs_fast_update().await;
|
||||
// Just ensure it doesn't panic
|
||||
#[allow(clippy::overly_complex_bool_expr)]
|
||||
let _ = needs_update || !needs_update;
|
||||
}
|
||||
|
||||
#[test]
|
||||
#[serial]
|
||||
fn test_global_metrics_functions() {
|
||||
// Test global functions don't panic
|
||||
let before = get_capacity_metrics().cache_hits.load(std::sync::atomic::Ordering::Relaxed);
|
||||
|
||||
record_global_write_operation();
|
||||
record_global_cache_hit();
|
||||
record_global_cache_miss();
|
||||
|
||||
let metrics = get_capacity_metrics();
|
||||
assert!(metrics.cache_hits.load(std::sync::atomic::Ordering::Relaxed) > before);
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user