From 7820a55fdc0f3cc1b9926d0e0c06194673361d8e Mon Sep 17 00:00:00 2001 From: cxymds Date: Fri, 26 Jun 2026 21:55:18 +0800 Subject: [PATCH] fix: decouple readiness from cluster health (#3912) * fix: include foreground write pressure * fix: split readiness and cluster health probes * fix: publish ready on node readiness * fix: bound cluster health collection * test: pin health probe collector routing * fix: qualify app storage ECStore type * test(admin): narrow runtime capabilities topology assertion --------- Co-authored-by: houseme --- crates/config/src/constants/heal.rs | 7 + crates/config/src/constants/health.rs | 5 + .../ecstore/src/data_movement_backpressure.rs | 152 ++++++-- crates/heal/src/heal/manager.rs | 157 ++++++-- rustfs/src/admin/console.rs | 8 +- rustfs/src/admin/handlers/health.rs | 87 ++++- rustfs/src/admin/handlers/system.rs | 16 +- rustfs/src/server/compress.rs | 2 + rustfs/src/server/layer.rs | 42 ++- rustfs/src/server/mod.rs | 9 +- rustfs/src/server/prefix.rs | 3 + rustfs/src/server/readiness.rs | 357 ++++++++++++++++-- rustfs/src/storage/concurrency/manager.rs | 49 +++ rustfs/src/workload_admission.rs | 40 +- 14 files changed, 769 insertions(+), 165 deletions(-) diff --git a/crates/config/src/constants/heal.rs b/crates/config/src/constants/heal.rs index 8f067f509..647ca8533 100644 --- a/crates/config/src/constants/heal.rs +++ b/crates/config/src/constants/heal.rs @@ -140,6 +140,10 @@ pub const ENV_HEAL_MAINLINE_THROTTLE_ENABLE: &str = "RUSTFS_HEAL_MAINLINE_THROTT /// at which background heal work pauses starting new tasks. pub const ENV_HEAL_MAINLINE_READ_UTILIZATION_HIGH_PERCENT: &str = "RUSTFS_HEAL_MAINLINE_READ_UTILIZATION_HIGH_PERCENT"; +/// Environment variable that controls the foreground write utilization percentage +/// at which background heal work pauses starting new tasks. +pub const ENV_HEAL_MAINLINE_WRITE_UTILIZATION_HIGH_PERCENT: &str = "RUSTFS_HEAL_MAINLINE_WRITE_UTILIZATION_HIGH_PERCENT"; + /// Environment variable that controls how soon the heal scheduler rechecks foreground /// pressure after delaying background work. pub const ENV_HEAL_MAINLINE_MAX_SLEEP_MS: &str = "RUSTFS_HEAL_MAINLINE_MAX_SLEEP_MS"; @@ -168,5 +172,8 @@ pub const DEFAULT_HEAL_MAINLINE_THROTTLE_ENABLE: bool = true; /// Default foreground read permit utilization threshold for pausing best-effort heal task starts. pub const DEFAULT_HEAL_MAINLINE_READ_UTILIZATION_HIGH_PERCENT: usize = 80; +/// Default foreground write utilization threshold for pausing best-effort heal task starts. +pub const DEFAULT_HEAL_MAINLINE_WRITE_UTILIZATION_HIGH_PERCENT: usize = 80; + /// Default foreground pressure recheck delay for heal scheduler, in milliseconds. pub const DEFAULT_HEAL_MAINLINE_MAX_SLEEP_MS: u64 = 250; diff --git a/crates/config/src/constants/health.rs b/crates/config/src/constants/health.rs index 04b528dba..628d0f443 100644 --- a/crates/config/src/constants/health.rs +++ b/crates/config/src/constants/health.rs @@ -22,6 +22,11 @@ pub const DEFAULT_HEALTH_ENDPOINT_ENABLE: bool = true; pub const ENV_HEALTH_READINESS_CACHE_TTL_MS: &str = "RUSTFS_HEALTH_READINESS_CACHE_TTL_MS"; pub const DEFAULT_HEALTH_READINESS_CACHE_TTL_MS: u64 = 1000; +/// Timeout for cluster health readiness collectors (milliseconds). +/// This bounds expensive storage and lock quorum checks used by cluster probes. +pub const ENV_HEALTH_CLUSTER_TIMEOUT_MS: &str = "RUSTFS_HEALTH_CLUSTER_TIMEOUT_MS"; +pub const DEFAULT_HEALTH_CLUSTER_TIMEOUT_MS: u64 = 2000; + /// Enable minimal health payload mode for GET `/health*` responses. /// When enabled, only `status` and `ready` fields are returned. pub const ENV_HEALTH_MINIMAL_RESPONSE_ENABLE: &str = "RUSTFS_HEALTH_MINIMAL_RESPONSE_ENABLE"; diff --git a/crates/ecstore/src/data_movement_backpressure.rs b/crates/ecstore/src/data_movement_backpressure.rs index 84664c1b7..7d91a7204 100644 --- a/crates/ecstore/src/data_movement_backpressure.rs +++ b/crates/ecstore/src/data_movement_backpressure.rs @@ -26,9 +26,11 @@ const LOG_SUBSYSTEM_DATA_MOVEMENT: &str = "data_movement"; const EVENT_DATA_MOVEMENT_BACKPRESSURE: &str = "data_movement_backpressure"; const DATA_MOVEMENT_BACKPRESSURE_ENABLE_ENV: &str = "RUSTFS_DATA_MOVEMENT_BACKPRESSURE_ENABLE"; const DATA_MOVEMENT_FOREGROUND_READ_HIGH_PERCENT_ENV: &str = "RUSTFS_DATA_MOVEMENT_FOREGROUND_READ_HIGH_PERCENT"; +const DATA_MOVEMENT_FOREGROUND_WRITE_HIGH_PERCENT_ENV: &str = "RUSTFS_DATA_MOVEMENT_FOREGROUND_WRITE_HIGH_PERCENT"; const DATA_MOVEMENT_RECHECK_MS_ENV: &str = "RUSTFS_DATA_MOVEMENT_RECHECK_MS"; const DEFAULT_DATA_MOVEMENT_BACKPRESSURE_ENABLE: bool = true; const DEFAULT_DATA_MOVEMENT_FOREGROUND_READ_HIGH_PERCENT: usize = 80; +const DEFAULT_DATA_MOVEMENT_FOREGROUND_WRITE_HIGH_PERCENT: usize = 80; const DEFAULT_DATA_MOVEMENT_RECHECK_MS: u64 = 250; const MIN_DATA_MOVEMENT_RECHECK_MS: u64 = 1; @@ -51,6 +53,7 @@ impl DataMovementOperation { struct DataMovementBackpressureConfig { enabled: bool, foreground_read_high_percent: usize, + foreground_write_high_percent: usize, recheck_delay: Duration, } @@ -62,6 +65,10 @@ impl DataMovementBackpressureConfig { DATA_MOVEMENT_FOREGROUND_READ_HIGH_PERCENT_ENV, DEFAULT_DATA_MOVEMENT_FOREGROUND_READ_HIGH_PERCENT, ), + foreground_write_high_percent: rustfs_utils::get_env_usize( + DATA_MOVEMENT_FOREGROUND_WRITE_HIGH_PERCENT_ENV, + DEFAULT_DATA_MOVEMENT_FOREGROUND_WRITE_HIGH_PERCENT, + ), recheck_delay: Duration::from_millis( rustfs_utils::get_env_u64(DATA_MOVEMENT_RECHECK_MS_ENV, DEFAULT_DATA_MOVEMENT_RECHECK_MS) .max(MIN_DATA_MOVEMENT_RECHECK_MS), @@ -95,24 +102,24 @@ async fn wait_for_data_movement_admission_with_provider( if cancel_token.is_cancelled() { return Err(Error::OperationCanceled); } - if !config.enabled || config.foreground_read_high_percent == 0 { + if !config.enabled || (config.foreground_read_high_percent == 0 && config.foreground_write_high_percent == 0) { return Ok(()); } let Some(provider) = provider else { return Ok(()); }; - let mut delayed_since: Option = None; + let mut delayed_since: Option<(Instant, ForegroundPressure)> = None; loop { if cancel_token.is_cancelled() { record_delay_completion(operation, pool_index, delayed_since, "cancelled"); return Err(Error::OperationCanceled); } - if let Some(usage_pct) = foreground_read_pressure_pct(&config, Some(provider.as_ref())) { + if let Some(pressure) = foreground_pressure(&config, Some(provider.as_ref())) { if delayed_since.is_none() { - delayed_since = Some(Instant::now()); - record_delay_start(operation, pool_index, usage_pct, &config); + delayed_since = Some((Instant::now(), pressure)); + record_delay_start(operation, pool_index, pressure, &config); } tokio::select! { @@ -130,44 +137,77 @@ async fn wait_for_data_movement_admission_with_provider( } } -fn foreground_read_pressure_pct( +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct ForegroundPressure { + class: WorkloadClass, + usage_pct: usize, + threshold_pct: usize, +} + +impl ForegroundPressure { + const fn reason(self) -> &'static str { + match self.class { + WorkloadClass::ForegroundRead => "foreground_read_pressure", + WorkloadClass::ForegroundWrite => "foreground_write_pressure", + _ => "foreground_pressure", + } + } +} + +fn foreground_pressure( config: &DataMovementBackpressureConfig, provider: Option<&(dyn WorkloadAdmissionSnapshotProvider + Send + Sync)>, -) -> Option { - if !config.enabled || config.foreground_read_high_percent == 0 { +) -> Option { + if !config.enabled { return None; } let snapshot = provider?.workload_admission_snapshot(); - let foreground_read = snapshot.get(WorkloadClass::ForegroundRead)?; - if matches!(foreground_read.state, AdmissionState::Saturated) { - return Some(100); - } + [ + (WorkloadClass::ForegroundRead, config.foreground_read_high_percent), + (WorkloadClass::ForegroundWrite, config.foreground_write_high_percent), + ] + .into_iter() + .filter_map(|(class, threshold_pct)| { + if threshold_pct == 0 { + return None; + } - let limit = foreground_read.limit?; - if limit == 0 { - return None; - } + let entry = snapshot.get(class)?; + let usage_pct = if matches!(entry.state, AdmissionState::Saturated) { + 100 + } else { + let limit = entry.limit?; + if limit == 0 { + return None; + } + entry + .active + .unwrap_or(0) + .saturating_mul(100) + .checked_div(limit) + .unwrap_or(100) + }; - let usage_pct = foreground_read - .active - .unwrap_or(0) - .saturating_mul(100) - .checked_div(limit) - .unwrap_or(100); - (usage_pct >= config.foreground_read_high_percent).then_some(usage_pct) + (usage_pct >= threshold_pct).then_some(ForegroundPressure { + class, + usage_pct, + threshold_pct, + }) + }) + .max_by_key(|pressure| pressure.usage_pct) } fn record_delay_start( operation: DataMovementOperation, pool_index: usize, - foreground_read_usage_pct: usize, + pressure: ForegroundPressure, config: &DataMovementBackpressureConfig, ) { counter!( "rustfs_data_movement_backpressure_total", "operation" => operation.as_str().to_string(), - "reason" => "foreground_read_pressure".to_string(), + "reason" => pressure.reason().to_string(), "result" => "delayed".to_string(), "pool_index" => pool_index.to_string() ) @@ -181,21 +221,22 @@ fn record_delay_start( operation = operation.as_str(), pool_index, state = "delayed", - reason = "foreground_read_pressure", - foreground_read_usage_pct, - threshold_pct = config.foreground_read_high_percent, + reason = pressure.reason(), + workload_class = pressure.class.as_str(), + foreground_usage_pct = pressure.usage_pct, + threshold_pct = pressure.threshold_pct, recheck_delay_ms = config.recheck_delay.as_millis(), - "Data movement delayed under foreground read pressure" + "Data movement delayed under foreground pressure" ); } fn record_delay_completion( operation: DataMovementOperation, pool_index: usize, - delayed_since: Option, + delayed_since: Option<(Instant, ForegroundPressure)>, result: &'static str, ) { - let Some(delayed_since) = delayed_since else { + let Some((delayed_since, pressure)) = delayed_since else { return; }; @@ -203,7 +244,7 @@ fn record_delay_completion( counter!( "rustfs_data_movement_backpressure_total", "operation" => operation.as_str().to_string(), - "reason" => "foreground_read_pressure".to_string(), + "reason" => pressure.reason().to_string(), "result" => result.to_string(), "pool_index" => pool_index.to_string() ) @@ -211,7 +252,7 @@ fn record_delay_completion( histogram!( "rustfs_data_movement_backpressure_delay_seconds", "operation" => operation.as_str().to_string(), - "reason" => "foreground_read_pressure".to_string(), + "reason" => pressure.reason().to_string(), "result" => result.to_string(), "pool_index" => pool_index.to_string() ) @@ -225,6 +266,8 @@ fn record_delay_completion( operation = operation.as_str(), pool_index, state = result, + reason = pressure.reason(), + workload_class = pressure.class.as_str(), delay_secs = delay.as_secs_f64(), "Data movement backpressure wait completed" ); @@ -259,6 +302,7 @@ mod tests { DataMovementBackpressureConfig { enabled: true, foreground_read_high_percent: 80, + foreground_write_high_percent: 80, recheck_delay: Duration::from_millis(1), } } @@ -268,7 +312,14 @@ mod tests { let provider = StaticWorkloadProvider::new(WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundRead, AdmissionState::Saturated)); - assert_eq!(foreground_read_pressure_pct(&test_config(), Some(&provider)), Some(100)); + assert_eq!( + foreground_pressure(&test_config(), Some(&provider)), + Some(ForegroundPressure { + class: WorkloadClass::ForegroundRead, + usage_pct: 100, + threshold_pct: 80, + }) + ); } #[test] @@ -281,11 +332,38 @@ mod tests { ), ); - assert_eq!(foreground_read_pressure_pct(&test_config(), Some(&provider)), Some(80)); + assert_eq!( + foreground_pressure(&test_config(), Some(&provider)), + Some(ForegroundPressure { + class: WorkloadClass::ForegroundRead, + usage_pct: 80, + threshold_pct: 80, + }) + ); } #[test] - fn foreground_read_pressure_ignores_open_low_usage() { + fn foreground_write_pressure_uses_active_limit_threshold() { + let provider = StaticWorkloadProvider::new( + WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, AdmissionState::Open).with_counts( + Some(9), + None, + Some(10), + ), + ); + + assert_eq!( + foreground_pressure(&test_config(), Some(&provider)), + Some(ForegroundPressure { + class: WorkloadClass::ForegroundWrite, + usage_pct: 90, + threshold_pct: 80, + }) + ); + } + + #[test] + fn foreground_pressure_ignores_open_low_usage() { let provider = StaticWorkloadProvider::new( WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundRead, AdmissionState::Open).with_counts( Some(7), @@ -294,7 +372,7 @@ mod tests { ), ); - assert_eq!(foreground_read_pressure_pct(&test_config(), Some(&provider)), None); + assert_eq!(foreground_pressure(&test_config(), Some(&provider)), None); } #[tokio::test] diff --git a/crates/heal/src/heal/manager.rs b/crates/heal/src/heal/manager.rs index f1ba225d4..dd266de08 100644 --- a/crates/heal/src/heal/manager.rs +++ b/crates/heal/src/heal/manager.rs @@ -106,6 +106,23 @@ enum QueuePushOutcome { Merged, } +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct ForegroundPressure { + class: WorkloadClass, + usage_pct: usize, + threshold_pct: usize, +} + +impl ForegroundPressure { + const fn reason(self) -> &'static str { + match self.class { + WorkloadClass::ForegroundRead => "foreground_read_pressure", + WorkloadClass::ForegroundWrite => "foreground_write_pressure", + _ => "foreground_pressure", + } + } +} + #[derive(Debug, Clone)] struct CompletedHealStatus { heal_type: HealType, @@ -636,7 +653,9 @@ pub struct HealConfig { pub mainline_throttle_enable: bool, /// Foreground read permit utilization percentage that delays best-effort heal starts. pub mainline_read_utilization_high_percent: usize, - /// Delay before rechecking foreground read pressure after delaying heal starts. + /// Foreground write utilization percentage that delays best-effort heal starts. + pub mainline_write_utilization_high_percent: usize, + /// Delay before rechecking foreground pressure after delaying heal starts. pub mainline_max_sleep: Duration, } @@ -691,6 +710,11 @@ impl Default for HealConfig { rustfs_config::DEFAULT_HEAL_MAINLINE_READ_UTILIZATION_HIGH_PERCENT, ) .min(100); + let mainline_write_utilization_high_percent = rustfs_utils::get_env_usize( + rustfs_config::ENV_HEAL_MAINLINE_WRITE_UTILIZATION_HIGH_PERCENT, + rustfs_config::DEFAULT_HEAL_MAINLINE_WRITE_UTILIZATION_HIGH_PERCENT, + ) + .min(100); let mainline_max_sleep = Duration::from_millis(rustfs_utils::get_env_u64( rustfs_config::ENV_HEAL_MAINLINE_MAX_SLEEP_MS, rustfs_config::DEFAULT_HEAL_MAINLINE_MAX_SLEEP_MS, @@ -709,6 +733,7 @@ impl Default for HealConfig { page_parallel_enable, mainline_throttle_enable, mainline_read_utilization_high_percent, + mainline_write_utilization_high_percent, mainline_max_sleep, } } @@ -833,29 +858,51 @@ impl HealManager { || matches!(request.priority, HealPriority::High | HealPriority::Urgent) } - fn mainline_throttle_active(config: &HealConfig, provider: &Option) -> Option { - if !config.mainline_throttle_enable || config.mainline_read_utilization_high_percent == 0 { + fn mainline_throttle_active( + config: &HealConfig, + provider: &Option, + ) -> Option { + if !config.mainline_throttle_enable + || (config.mainline_read_utilization_high_percent == 0 && config.mainline_write_utilization_high_percent == 0) + { return None; } let provider = provider.as_ref()?; let snapshot = provider.workload_admission_snapshot(); - let foreground_read = snapshot.get(WorkloadClass::ForegroundRead)?; - if matches!(foreground_read.state, AdmissionState::Saturated) { - return Some(100); - } + [ + (WorkloadClass::ForegroundRead, config.mainline_read_utilization_high_percent), + (WorkloadClass::ForegroundWrite, config.mainline_write_utilization_high_percent), + ] + .into_iter() + .filter_map(|(class, threshold_pct)| { + if threshold_pct == 0 { + return None; + } - let limit = foreground_read.limit?; - if limit == 0 { - return None; - } - let usage_pct = foreground_read - .active - .unwrap_or(0) - .saturating_mul(100) - .checked_div(limit) - .unwrap_or(100); - (usage_pct >= config.mainline_read_utilization_high_percent).then_some(usage_pct) + let entry = snapshot.get(class)?; + let usage_pct = if matches!(entry.state, AdmissionState::Saturated) { + 100 + } else { + let limit = entry.limit?; + if limit == 0 { + return None; + } + entry + .active + .unwrap_or(0) + .saturating_mul(100) + .checked_div(limit) + .unwrap_or(100) + }; + + (usage_pct >= threshold_pct).then_some(ForegroundPressure { + class, + usage_pct, + threshold_pct, + }) + }) + .max_by_key(|pressure| pressure.usage_pct) } fn schedule_mainline_throttle_recheck(notify: Arc, delay: Duration) { @@ -870,12 +917,12 @@ impl HealManager { }); } - fn record_mainline_throttle_delay(usage_pct: usize, config: &HealConfig) { + fn record_mainline_throttle_delay(pressure: ForegroundPressure, config: &HealConfig) { counter!( "rustfs_heal_mainline_throttle_total", "source" => "background", "result" => "delayed", - "reason" => "foreground_read_pressure" + "reason" => pressure.reason() ) .increment(1); debug!( @@ -884,10 +931,12 @@ impl HealManager { component = LOG_COMPONENT_HEAL, subsystem = LOG_SUBSYSTEM_MANAGER, state = "delayed", - foreground_read_usage_pct = usage_pct, - threshold_pct = config.mainline_read_utilization_high_percent, + reason = pressure.reason(), + workload_class = pressure.class.as_str(), + foreground_usage_pct = pressure.usage_pct, + threshold_pct = pressure.threshold_pct, recheck_delay_ms = config.mainline_max_sleep.as_millis(), - "Heal scheduler delayed background work under foreground read pressure" + "Heal scheduler delayed background work under foreground pressure" ); } @@ -2294,7 +2343,7 @@ impl HealManager { } = context; let config = config.read().await; - let mainline_usage_pct = Self::mainline_throttle_active(&config, workload_provider); + let mainline_pressure = Self::mainline_throttle_active(&config, workload_provider); let mut active_heals_guard = active_heals.lock().await; publish_active_heal_count(&active_heals_guard); @@ -2320,13 +2369,13 @@ impl HealManager { let mut delayed_by_mainline_throttle = false; for _ in 0..available_slots { - let selected_request = if config.set_bulkhead_enable || mainline_usage_pct.is_some() { + let selected_request = if config.set_bulkhead_enable || mainline_pressure.is_some() { let max_concurrent_per_set = config.max_concurrent_per_set; let (selected_request, skipped_sets) = queue.pop_runnable_with_skips( |request| { let set_allowed = !config.set_bulkhead_enable || can_schedule_request(request, &running_per_set, max_concurrent_per_set); - let mainline_allowed = mainline_usage_pct.is_none() || Self::request_bypasses_mainline_throttle(request); + let mainline_allowed = mainline_pressure.is_none() || Self::request_bypasses_mainline_throttle(request); set_allowed && mainline_allowed }, |request| heal_request_set_key(request).map(|_| heal_request_set_metric_label(request)), @@ -2614,7 +2663,7 @@ impl HealManager { }); tasks_started += 1; } else { - delayed_by_mainline_throttle = mainline_usage_pct.is_some(); + delayed_by_mainline_throttle = mainline_pressure.is_some(); break; } } @@ -2626,8 +2675,8 @@ impl HealManager { publish_active_heal_count(&active_heals_guard); publish_heal_queue_length(&queue); - if delayed_by_mainline_throttle && let Some(usage_pct) = mainline_usage_pct { - Self::record_mainline_throttle_delay(usage_pct, &config); + if delayed_by_mainline_throttle && let Some(pressure) = mainline_pressure { + Self::record_mainline_throttle_delay(pressure, &config); Self::schedule_mainline_throttle_recheck(notify.clone(), config.mainline_max_sleep); } @@ -2757,6 +2806,7 @@ mod tests { #[derive(Debug)] struct FixedWorkloadProvider { + class: WorkloadClass, active: usize, limit: usize, state: AdmissionState, @@ -2764,13 +2814,11 @@ mod tests { impl WorkloadAdmissionSnapshotProvider for FixedWorkloadProvider { fn workload_admission_snapshot(&self) -> WorkloadAdmissionRegistrySnapshot { - WorkloadAdmissionRegistrySnapshot::new(vec![ - WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundRead, self.state).with_counts( - Some(self.active), - None, - Some(self.limit), - ), - ]) + WorkloadAdmissionRegistrySnapshot::new(vec![WorkloadAdmissionSnapshot::new(self.class, self.state).with_counts( + Some(self.active), + None, + Some(self.limit), + )]) } } @@ -4346,6 +4394,7 @@ mod tests { async fn test_mainline_throttle_delays_background_heal_start() { let storage: Arc = Arc::new(MockStorage); let provider: WorkloadSnapshotProviderRef = Arc::new(FixedWorkloadProvider { + class: WorkloadClass::ForegroundRead, active: 8, limit: 10, state: AdmissionState::Open, @@ -4356,6 +4405,40 @@ mod tests { max_concurrent_heals: 1, mainline_throttle_enable: true, mainline_read_utilization_high_percent: 80, + mainline_write_utilization_high_percent: 80, + mainline_max_sleep: Duration::from_millis(1), + ..HealConfig::default() + }), + Some(provider), + ); + + manager + .submit_heal_request(bucket_request("read-repair", HealPriority::Normal, HealRequestSource::ReadRepair)) + .await + .expect("read repair request should be queued"); + + process_manager_queue_once(&manager).await; + + assert_eq!(manager.get_queue_length().await, 1); + assert_eq!(manager.get_active_task_count().await, 0); + } + + #[tokio::test] + async fn test_mainline_throttle_delays_background_heal_start_under_write_pressure() { + let storage: Arc = Arc::new(MockStorage); + let provider: WorkloadSnapshotProviderRef = Arc::new(FixedWorkloadProvider { + class: WorkloadClass::ForegroundWrite, + active: 9, + limit: 10, + state: AdmissionState::Open, + }); + let manager = HealManager::new_with_workload_provider( + storage, + Some(HealConfig { + max_concurrent_heals: 1, + mainline_throttle_enable: true, + mainline_read_utilization_high_percent: 80, + mainline_write_utilization_high_percent: 80, mainline_max_sleep: Duration::from_millis(1), ..HealConfig::default() }), @@ -4377,6 +4460,7 @@ mod tests { async fn test_mainline_throttle_allows_admin_high_start() { let storage: Arc = Arc::new(MockStorage); let provider: WorkloadSnapshotProviderRef = Arc::new(FixedWorkloadProvider { + class: WorkloadClass::ForegroundRead, active: 10, limit: 10, state: AdmissionState::Saturated, @@ -4387,6 +4471,7 @@ mod tests { max_concurrent_heals: 1, mainline_throttle_enable: true, mainline_read_utilization_high_percent: 80, + mainline_write_utilization_high_percent: 80, mainline_max_sleep: Duration::from_millis(1), ..HealConfig::default() }), diff --git a/rustfs/src/admin/console.rs b/rustfs/src/admin/console.rs index c9721cc2b..ed739c079 100644 --- a/rustfs/src/admin/console.rs +++ b/rustfs/src/admin/console.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::admin::handlers::health::{HealthProbe, build_health_response_parts, collect_dependency_readiness}; +use crate::admin::handlers::health::{HealthProbe, build_health_response_parts, collect_probe_readiness}; use crate::admin::runtime_sources::{default_admin_usecase, resolve_oidc_handle}; use crate::admin::storage_api::access::RequestContext; use crate::license::has_valid_license; @@ -595,11 +595,7 @@ async fn health_check(method: Method, uri: Uri) -> Response { } else { HealthProbe::Liveness }; - let readiness_report = if probe == HealthProbe::Readiness { - Some(collect_dependency_readiness().await) - } else { - None - }; + let readiness_report = collect_probe_readiness(probe).await; let uptime = std::time::SystemTime::now() .duration_since(std::time::UNIX_EPOCH) .unwrap_or_default() diff --git a/rustfs/src/admin/handlers/health.rs b/rustfs/src/admin/handlers/health.rs index 787b69ed8..cfd8c569e 100644 --- a/rustfs/src/admin/handlers/health.rs +++ b/rustfs/src/admin/handlers/health.rs @@ -15,8 +15,9 @@ use super::profile::{TriggerProfileCPU, TriggerProfileMemory}; use crate::admin::router::{AdminOperation, Operation, S3Router}; use crate::server::{ - HEALTH_PREFIX, HEALTH_READY_PATH, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, - collect_dependency_readiness_report as collect_runtime_dependency_readiness_report, + HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_HEALTH_CLUSTER_PATH, MINIO_HEALTH_CLUSTER_READ_PATH, MINIO_HEALTH_READY_PATH, + PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, collect_cluster_read_health_report, collect_cluster_write_health_report, + collect_node_readiness_report, }; use http::{HeaderMap, HeaderValue}; use hyper::{Method, StatusCode}; @@ -55,6 +56,21 @@ pub(crate) struct HealthCheckState { pub(crate) enum HealthProbe { Liveness, Readiness, + ClusterWrite, + ClusterRead, +} + +impl HealthProbe { + const fn requires_lock_quorum(self) -> bool { + matches!(self, Self::ClusterWrite | Self::ClusterRead) + } +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum HealthReadinessSource { + Node, + ClusterWrite, + ClusterRead, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -76,8 +92,21 @@ pub(crate) struct HealthPayloadContext<'a> { pub(crate) include_dependency_details: bool, } -pub(crate) async fn collect_dependency_readiness() -> crate::server::DependencyReadinessReport { - collect_runtime_dependency_readiness_report().await +pub(crate) async fn collect_probe_readiness(probe: HealthProbe) -> Option { + match readiness_source_for_probe(probe)? { + HealthReadinessSource::Node => Some(collect_node_readiness_report().await), + HealthReadinessSource::ClusterWrite => Some(collect_cluster_write_health_report().await), + HealthReadinessSource::ClusterRead => Some(collect_cluster_read_health_report().await), + } +} + +fn readiness_source_for_probe(probe: HealthProbe) -> Option { + match probe { + HealthProbe::Liveness => None, + HealthProbe::Readiness => Some(HealthReadinessSource::Node), + HealthProbe::ClusterWrite => Some(HealthReadinessSource::ClusterWrite), + HealthProbe::ClusterRead => Some(HealthReadinessSource::ClusterRead), + } } pub(crate) fn health_check_state( @@ -94,7 +123,7 @@ pub(crate) fn health_check_state( }; } - let ready = storage_ready && iam_ready && lock_quorum_ready; + let ready = storage_ready && iam_ready && (!probe.requires_lock_quorum() || lock_quorum_ready); let status = if ready { "ok" } else { "degraded" }; let status_code = if ready { @@ -158,10 +187,11 @@ pub(crate) fn build_degraded_reasons(reasons: &[crate::server::ReadinessDegraded } pub(crate) fn probe_from_path(path: &str) -> HealthProbe { - if path == HEALTH_READY_PATH { - HealthProbe::Readiness - } else { - HealthProbe::Liveness + match path { + HEALTH_READY_PATH | MINIO_HEALTH_READY_PATH => HealthProbe::Readiness, + MINIO_HEALTH_CLUSTER_PATH => HealthProbe::ClusterWrite, + MINIO_HEALTH_CLUSTER_READ_PATH => HealthProbe::ClusterRead, + _ => HealthProbe::Liveness, } } @@ -175,7 +205,7 @@ pub(crate) fn build_health_response_parts( ) -> HealthResponseParts { let (storage_ready, iam_ready, lock_quorum_ready, mut health, mut degraded_reasons, include_dependency_details) = match (probe, readiness_report) { - (HealthProbe::Readiness, Some(readiness_report)) => { + (probe @ (HealthProbe::Readiness | HealthProbe::ClusterWrite | HealthProbe::ClusterRead), Some(readiness_report)) => { let storage_ready = readiness_report.readiness.storage_ready; let iam_ready = readiness_report.readiness.iam_ready; let lock_quorum_ready = readiness_report.readiness.lock_quorum_ready; @@ -188,7 +218,7 @@ pub(crate) fn build_health_response_parts( true, ) } - (HealthProbe::Readiness, None) => ( + (HealthProbe::Readiness | HealthProbe::ClusterWrite | HealthProbe::ClusterRead, None) => ( false, false, false, @@ -284,11 +314,7 @@ impl Operation for HealthCheckHandler { } let probe = probe_from_path(req.uri.path()); - let readiness_report = if probe == HealthProbe::Readiness { - Some(collect_dependency_readiness().await) - } else { - None - }; + let readiness_report = collect_probe_readiness(probe).await; let response_parts = build_health_response_parts(method.clone(), probe, readiness_report.as_ref(), "rustfs-endpoint", None, None); @@ -347,6 +373,14 @@ mod tests { #[test] fn test_readiness_state_lock_not_ready() { let state = health_check_state(true, true, false, HealthProbe::Readiness); + assert_eq!(state.status_code, StatusCode::OK); + assert_eq!(state.status, "ok"); + assert!(state.ready); + } + + #[test] + fn test_cluster_write_state_lock_not_ready() { + let state = health_check_state(true, true, false, HealthProbe::ClusterWrite); assert_eq!(state.status_code, StatusCode::SERVICE_UNAVAILABLE); assert_eq!(state.status, "degraded"); assert!(!state.ready); @@ -375,6 +409,27 @@ mod tests { #[test] fn test_probe_from_path_readiness() { assert_eq!(probe_from_path(HEALTH_READY_PATH), HealthProbe::Readiness); + assert_eq!(probe_from_path(MINIO_HEALTH_READY_PATH), HealthProbe::Readiness); + assert_eq!(probe_from_path(MINIO_HEALTH_CLUSTER_PATH), HealthProbe::ClusterWrite); + assert_eq!(probe_from_path(MINIO_HEALTH_CLUSTER_READ_PATH), HealthProbe::ClusterRead); + } + + #[test] + fn test_readiness_probe_uses_node_collector_only() { + assert_eq!(readiness_source_for_probe(HealthProbe::Readiness), Some(HealthReadinessSource::Node)); + assert_eq!(readiness_source_for_probe(HealthProbe::Liveness), None); + } + + #[test] + fn test_cluster_probes_use_cluster_collectors() { + assert_eq!( + readiness_source_for_probe(HealthProbe::ClusterWrite), + Some(HealthReadinessSource::ClusterWrite) + ); + assert_eq!( + readiness_source_for_probe(HealthProbe::ClusterRead), + Some(HealthReadinessSource::ClusterRead) + ); } #[test] diff --git a/rustfs/src/admin/handlers/system.rs b/rustfs/src/admin/handlers/system.rs index 806e09209..eca6e99fa 100644 --- a/rustfs/src/admin/handlers/system.rs +++ b/rustfs/src/admin/handlers/system.rs @@ -521,7 +521,7 @@ mod tests { CapabilityState, CapabilityStatus, MemorySamplingState, ObservabilitySnapshot, PlatformSupport, TopologyCapabilities, TopologySnapshot, UserspaceProfilingCapability, }; - use rustfs_concurrency::{AdmissionState, WorkloadClass}; + use rustfs_concurrency::WorkloadClass; use rustfs_madmin::{InfoMessage, StorageInfo}; #[tokio::test] @@ -539,20 +539,6 @@ mod tests { assert_eq!(response.cluster_snapshot_summary, None); assert_eq!(response.summary.cluster_snapshot.state, CapabilityState::Unknown); assert_eq!(response.observability.platform.os.as_deref(), Some(std::env::consts::OS)); - assert_eq!( - response - .workload_admission - .get(WorkloadClass::ForegroundRead) - .map(|snapshot| snapshot.state), - Some(AdmissionState::Open) - ); - assert_eq!( - response - .workload_admission - .get(WorkloadClass::ForegroundWrite) - .map(|snapshot| snapshot.state), - Some(AdmissionState::Disabled) - ); assert_eq!(response.workload_admission.entries().len(), WorkloadClass::REQUIRED.len()); } diff --git a/rustfs/src/server/compress.rs b/rustfs/src/server/compress.rs index d92d06e68..bccd80c8a 100644 --- a/rustfs/src/server/compress.rs +++ b/rustfs/src/server/compress.rs @@ -423,6 +423,7 @@ impl PathCategory { || path == "/minio/health/live" || path == "/minio/health/ready" || path == "/minio/health/cluster" + || path == "/minio/health/cluster/read" { PathCategory::Probe } else { @@ -774,6 +775,7 @@ mod tests { assert_eq!(PathCategory::classify("/minio/health/live"), PathCategory::Probe); assert_eq!(PathCategory::classify("/minio/health/ready"), PathCategory::Probe); assert_eq!(PathCategory::classify("/minio/health/cluster"), PathCategory::Probe); + assert_eq!(PathCategory::classify("/minio/health/cluster/read"), PathCategory::Probe); } #[test] diff --git a/rustfs/src/server/layer.rs b/rustfs/src/server/layer.rs index 54abd4a7b..cc7a94785 100644 --- a/rustfs/src/server/layer.rs +++ b/rustfs/src/server/layer.rs @@ -14,15 +14,15 @@ use super::runtime_sources; use crate::admin::console::is_console_path; -use crate::admin::handlers::health::{HealthProbe, build_health_response_parts}; +use crate::admin::handlers::health::{HealthProbe, build_health_response_parts, collect_probe_readiness}; use crate::error::ApiError; use crate::server::RemoteAddr; use crate::server::cors; use crate::server::hybrid::HybridBody; use crate::server::{ ADMIN_PREFIX, CONSOLE_PREFIX, HEALTH_COMPAT_LIVE_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_ADMIN_PREFIX, - MINIO_ADMIN_V3_PREFIX, MINIO_HEALTH_CLUSTER_PATH, MINIO_HEALTH_LIVE_PATH, MINIO_HEALTH_READY_PATH, RPC_PREFIX, - RUSTFS_ADMIN_PREFIX, active_http_requests, collect_dependency_readiness_report, has_path_prefix, is_admin_path, + MINIO_ADMIN_V3_PREFIX, MINIO_HEALTH_CLUSTER_PATH, MINIO_HEALTH_CLUSTER_READ_PATH, MINIO_HEALTH_LIVE_PATH, + MINIO_HEALTH_READY_PATH, RPC_PREFIX, RUSTFS_ADMIN_PREFIX, active_http_requests, has_path_prefix, is_admin_path, is_table_catalog_path, }; use crate::storage_api::server::apply_cors_headers; @@ -886,7 +886,9 @@ fn resolve_public_health_probe(method: &Method, path: &str) -> Option Some(HealthProbe::Liveness), - HEALTH_READY_PATH | MINIO_HEALTH_READY_PATH | MINIO_HEALTH_CLUSTER_PATH => Some(HealthProbe::Readiness), + HEALTH_READY_PATH | MINIO_HEALTH_READY_PATH => Some(HealthProbe::Readiness), + MINIO_HEALTH_CLUSTER_PATH => Some(HealthProbe::ClusterWrite), + MINIO_HEALTH_CLUSTER_READ_PATH => Some(HealthProbe::ClusterRead), _ => None, } } @@ -939,11 +941,7 @@ where .expect("failed to build health busy response"); } - let readiness_report = if probe == HealthProbe::Readiness { - Some(collect_dependency_readiness_report().await) - } else { - None - }; + let readiness_report = collect_probe_readiness(probe).await; let kms_ready = if probe == HealthProbe::Readiness && health_compat_kms_ready_check_enabled() { Some(health_kms_ready().await) } else { @@ -1293,6 +1291,7 @@ impl ConditionalCorsLayer { "/minio/health/live", "/minio/health/ready", "/minio/health/cluster", + "/minio/health/cluster/read", "/profile/cpu", "/profile/memory", ]; @@ -1952,6 +1951,31 @@ mod tests { .await; } + #[tokio::test] + #[serial] + async fn public_health_endpoint_layer_handles_minio_health_cluster_read_before_inner_service() { + async_with_vars([(rustfs_config::ENV_HEALTH_ENDPOINT_ENABLE, Some("true"))], async { + let inner = CountingHybridService::default(); + let calls = inner.calls(); + let mut service = PublicHealthEndpointLayer.layer(inner); + + let response = service + .call( + Request::builder() + .method(Method::GET) + .uri(MINIO_HEALTH_CLUSTER_READ_PATH) + .body(Full::::from(Bytes::new())) + .expect("request"), + ) + .await + .expect("health response"); + + assert!(response.status() == StatusCode::OK || response.status() == StatusCode::SERVICE_UNAVAILABLE); + assert_eq!(calls.load(Ordering::SeqCst), 0); + }) + .await; + } + #[tokio::test] #[serial] async fn public_health_endpoint_layer_forwards_unknown_health_path_when_endpoint_disabled() { diff --git a/rustfs/src/server/mod.rs b/rustfs/src/server/mod.rs index d3ba1bc46..3688ac862 100644 --- a/rustfs/src/server/mod.rs +++ b/rustfs/src/server/mod.rs @@ -52,18 +52,19 @@ pub(crate) use module_switch::{ }; pub(crate) use prefix::{ ADMIN_PREFIX, CONSOLE_PREFIX, FAVICON_PATH, HEALTH_COMPAT_LIVE_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, LICENSE, - MINIO_ADMIN_PREFIX, MINIO_ADMIN_V3_PREFIX, MINIO_HEALTH_CLUSTER_PATH, MINIO_HEALTH_LIVE_PATH, MINIO_HEALTH_READY_PATH, - PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, RPC_PREFIX, RUSTFS_ADMIN_PREFIX, TABLE_CATALOG_COMPAT_PREFIX, TABLE_CATALOG_PREFIX, - TONIC_PREFIX, VERSION, has_path_prefix, is_admin_path, is_table_catalog_path, + MINIO_ADMIN_PREFIX, MINIO_ADMIN_V3_PREFIX, MINIO_HEALTH_CLUSTER_PATH, MINIO_HEALTH_CLUSTER_READ_PATH, MINIO_HEALTH_LIVE_PATH, + MINIO_HEALTH_READY_PATH, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, RPC_PREFIX, RUSTFS_ADMIN_PREFIX, TABLE_CATALOG_COMPAT_PREFIX, + TABLE_CATALOG_PREFIX, TONIC_PREFIX, VERSION, has_path_prefix, is_admin_path, is_table_catalog_path, }; pub(crate) use readiness::DependencyReadiness; pub(crate) use readiness::DependencyReadinessReport; pub(crate) use readiness::ReadinessDegradedReason; pub(crate) use readiness::ReadinessGateLayer; pub(crate) use readiness::collect_dependency_readiness; -pub(crate) use readiness::collect_dependency_readiness_report; +pub(crate) use readiness::collect_node_readiness_report; pub use readiness::publish_ready_when_runtime_ready; pub(crate) use readiness::snapshot_dependency_readiness_report; +pub(crate) use readiness::{collect_cluster_read_health_report, collect_cluster_write_health_report}; #[derive(Clone, Copy, Debug)] pub struct RemoteAddr(pub std::net::SocketAddr); diff --git a/rustfs/src/server/prefix.rs b/rustfs/src/server/prefix.rs index a32d76b51..a0a4ec116 100644 --- a/rustfs/src/server/prefix.rs +++ b/rustfs/src/server/prefix.rs @@ -43,6 +43,9 @@ pub(crate) const MINIO_HEALTH_READY_PATH: &str = "/minio/health/ready"; /// MinIO-compatible cluster health probe alias path. pub(crate) const MINIO_HEALTH_CLUSTER_PATH: &str = "/minio/health/cluster"; +/// MinIO-compatible cluster read health probe alias path. +pub(crate) const MINIO_HEALTH_CLUSTER_READ_PATH: &str = "/minio/health/cluster/read"; + /// Predefined administrative prefix for RustFS server routes. /// This prefix is used for endpoints that handle administrative tasks /// such as configuration, monitoring, and management. diff --git a/rustfs/src/server/readiness.rs b/rustfs/src/server/readiness.rs index e737caf07..4f894a901 100644 --- a/rustfs/src/server/readiness.rs +++ b/rustfs/src/server/readiness.rs @@ -58,6 +58,7 @@ pub enum ReadinessDegradedReason { IamNotReady, LockQuorumUnavailable, KmsNotReady, + ClusterHealthTimeout, StorageAndIamUnavailable, StorageAndLockUnavailable, IamAndLockUnavailable, @@ -71,6 +72,7 @@ impl ReadinessDegradedReason { ReadinessDegradedReason::IamNotReady => "iam_not_ready", ReadinessDegradedReason::LockQuorumUnavailable => "lock_quorum_unavailable", ReadinessDegradedReason::KmsNotReady => "kms_not_ready", + ReadinessDegradedReason::ClusterHealthTimeout => "cluster_health_timeout", ReadinessDegradedReason::StorageAndIamUnavailable => "storage_and_iam_unavailable", ReadinessDegradedReason::StorageAndLockUnavailable => "storage_and_lock_unavailable", ReadinessDegradedReason::IamAndLockUnavailable => "iam_and_lock_unavailable", @@ -137,6 +139,7 @@ fn is_probe_path(path: &str) -> bool { | crate::server::MINIO_HEALTH_LIVE_PATH | crate::server::MINIO_HEALTH_READY_PATH | crate::server::MINIO_HEALTH_CLUSTER_PATH + | crate::server::MINIO_HEALTH_CLUSTER_READ_PATH | crate::server::FAVICON_PATH ); @@ -208,7 +211,7 @@ pub async fn publish_ready_when_runtime_ready( wait_for_runtime_readiness_with( STARTUP_RUNTIME_READINESS_MAX_WAIT, STARTUP_RUNTIME_READINESS_POLL_INTERVAL, - collect_dependency_readiness_uncached, + collect_node_readiness, |dependency_readiness| { readiness.mark_stage(rustfs_common::SystemStage::FullReady); if let Some(state_manager) = state_manager { @@ -218,7 +221,7 @@ pub async fn publish_ready_when_runtime_ready( target: "rustfs::server::readiness", storage_ready = dependency_readiness.storage_ready, iam_ready = dependency_readiness.iam_ready, - "Runtime readiness reached write quorum; publishing ready state" + "Runtime node readiness reached; publishing ready state" ); }, ) @@ -237,6 +240,18 @@ struct LockQuorumCacheEntry { status: LockQuorumStatus, } +#[derive(Debug, Clone)] +struct ClusterHealthReportCacheEntry { + captured_at: Instant, + report: DependencyReadinessReport, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +enum ClusterHealthProbeKind { + Write, + Read, +} + #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct LockQuorumStatus { pub ready: bool, @@ -256,6 +271,16 @@ fn health_readiness_cache_ttl() -> Duration { )) } +fn health_cluster_timeout() -> Duration { + Duration::from_millis( + rustfs_utils::get_env_u64( + rustfs_config::ENV_HEALTH_CLUSTER_TIMEOUT_MS, + rustfs_config::DEFAULT_HEALTH_CLUSTER_TIMEOUT_MS, + ) + .max(1), + ) +} + fn storage_readiness_cache() -> &'static Mutex> { static CACHE: OnceLock>> = OnceLock::new(); CACHE.get_or_init(|| Mutex::new(None)) @@ -266,6 +291,46 @@ fn lock_quorum_status_cache() -> &'static Mutex> { CACHE.get_or_init(|| Mutex::new(None)) } +fn cluster_write_health_report_cache() -> &'static Mutex> { + static CACHE: OnceLock>> = OnceLock::new(); + CACHE.get_or_init(|| Mutex::new(None)) +} + +fn cluster_read_health_report_cache() -> &'static Mutex> { + static CACHE: OnceLock>> = OnceLock::new(); + CACHE.get_or_init(|| Mutex::new(None)) +} + +fn cluster_write_health_singleflight() -> &'static Mutex<()> { + static LOCK: OnceLock> = OnceLock::new(); + LOCK.get_or_init(|| Mutex::new(())) +} + +fn cluster_read_health_singleflight() -> &'static Mutex<()> { + static LOCK: OnceLock> = OnceLock::new(); + LOCK.get_or_init(|| Mutex::new(())) +} + +fn cluster_health_report_cache(kind: ClusterHealthProbeKind) -> &'static Mutex> { + match kind { + ClusterHealthProbeKind::Write => cluster_write_health_report_cache(), + ClusterHealthProbeKind::Read => cluster_read_health_report_cache(), + } +} + +fn cluster_health_singleflight(kind: ClusterHealthProbeKind) -> &'static Mutex<()> { + match kind { + ClusterHealthProbeKind::Write => cluster_write_health_singleflight(), + ClusterHealthProbeKind::Read => cluster_read_health_singleflight(), + } +} + +#[cfg(test)] +async fn reset_cluster_health_report_caches() { + *cluster_write_health_report_cache().lock().await = None; + *cluster_read_health_report_cache().lock().await = None; +} + async fn load_cached_storage_readiness() -> Option { let ttl = health_readiness_cache_ttl(); if ttl.is_zero() { @@ -281,6 +346,33 @@ async fn load_cached_storage_readiness() -> Option { None } +async fn load_cached_cluster_health_report(kind: ClusterHealthProbeKind) -> Option { + let ttl = health_readiness_cache_ttl(); + if ttl.is_zero() { + return None; + } + + let cache = cluster_health_report_cache(kind).lock().await; + let entry = cache.as_ref()?; + if entry.captured_at.elapsed() <= ttl { + return Some(entry.report.clone()); + } + + None +} + +async fn update_cluster_health_report_cache(kind: ClusterHealthProbeKind, report: DependencyReadinessReport) { + if health_readiness_cache_ttl().is_zero() { + return; + } + + let mut cache = cluster_health_report_cache(kind).lock().await; + *cache = Some(ClusterHealthReportCacheEntry { + captured_at: Instant::now(), + report, + }); +} + async fn update_storage_readiness_cache(storage_ready: bool) { if health_readiness_cache_ttl().is_zero() { return; @@ -364,7 +456,24 @@ fn pool_write_quorum(info: &StorageInfo, pool_idx: usize, set_drive_count: usize write_quorum.max(1) } -fn storage_ready_from_runtime_state(info: &StorageInfo) -> bool { +fn pool_read_quorum(info: &StorageInfo, pool_idx: usize, set_drive_count: usize) -> usize { + if set_drive_count == 0 { + return 1; + } + + info.backend + .standard_sc_data + .get(pool_idx) + .copied() + .filter(|count| *count > 0) + .unwrap_or_else(|| (set_drive_count / 2).max(1)) + .max(1) +} + +fn storage_ready_from_runtime_state_with_quorum(info: &StorageInfo, quorum_for_set: F) -> bool +where + F: Fn(&StorageInfo, usize, usize) -> usize, +{ if info.disks.is_empty() { return false; } @@ -407,11 +516,19 @@ fn storage_ready_from_runtime_state(info: &StorageInfo) -> bool { set_drive_counts.into_iter().all(|((pool_idx, set_idx), set_drive_count)| { let online = set_online_counts.get(&(pool_idx, set_idx)).copied().unwrap_or_default(); - let write_quorum = pool_write_quorum(info, pool_idx, set_drive_count); - online >= write_quorum + let quorum = quorum_for_set(info, pool_idx, set_drive_count); + online >= quorum }) } +fn storage_ready_from_runtime_state(info: &StorageInfo) -> bool { + storage_ready_from_runtime_state_with_quorum(info, pool_write_quorum) +} + +fn storage_read_ready_from_runtime_state(info: &StorageInfo) -> bool { + storage_ready_from_runtime_state_with_quorum(info, pool_read_quorum) +} + fn degraded_reasons(storage_ready: bool, iam_ready_raw: bool, lock_quorum_ready: bool) -> Vec { match (storage_ready, iam_ready_raw, lock_quorum_ready) { (true, true, true) => Vec::new(), @@ -465,6 +582,80 @@ pub async fn collect_dependency_readiness_report() -> DependencyReadinessReport report } +pub async fn collect_cluster_write_health_report() -> DependencyReadinessReport { + collect_cluster_health_report_with(ClusterHealthProbeKind::Write, collect_dependency_readiness_report).await +} + +pub async fn collect_cluster_read_health_report() -> DependencyReadinessReport { + collect_cluster_health_report_with(ClusterHealthProbeKind::Read, collect_cluster_read_dependency_readiness_report).await +} + +pub async fn collect_node_readiness_report() -> DependencyReadinessReport { + let readiness = DependencyReadiness { + storage_ready: runtime_sources::object_store_handle().is_some(), + iam_ready: runtime_sources::iam_ready(), + lock_quorum_ready: true, + }; + let report = dependency_readiness_report_from_readiness(readiness); + record_readiness_report(&report); + report +} + +async fn collect_cluster_health_report_with( + kind: ClusterHealthProbeKind, + mut load_report: LoadFn, +) -> DependencyReadinessReport +where + LoadFn: FnMut() -> Fut, + Fut: Future, +{ + if let Some(cached) = load_cached_cluster_health_report(kind).await { + return cached; + } + + let _singleflight = cluster_health_singleflight(kind).lock().await; + if let Some(cached) = load_cached_cluster_health_report(kind).await { + return cached; + } + + let report = match tokio::time::timeout(health_cluster_timeout(), load_report()).await { + Ok(report) => report, + Err(_) => cluster_health_timeout_report(), + }; + update_cluster_health_report_cache(kind, report.clone()).await; + report +} + +fn cluster_health_timeout_report() -> DependencyReadinessReport { + DependencyReadinessReport { + readiness: DependencyReadiness { + storage_ready: false, + iam_ready: false, + lock_quorum_ready: false, + }, + degraded_reasons: vec![ReadinessDegradedReason::ClusterHealthTimeout], + } +} + +async fn collect_node_readiness() -> DependencyReadiness { + collect_node_readiness_report().await.readiness +} + +pub async fn collect_cluster_read_dependency_readiness_report() -> DependencyReadinessReport { + let iam_ready_raw = runtime_sources::iam_ready(); + let storage_ready = collect_storage_read_readiness_uncached().await; + let lock_quorum_status = collect_lock_quorum_status().await; + + let readiness = DependencyReadiness { + storage_ready, + iam_ready: iam_ready_raw, + lock_quorum_ready: lock_quorum_status.ready, + }; + let report = dependency_readiness_report_from_readiness(readiness); + record_readiness_report(&report); + report +} + pub(crate) async fn snapshot_dependency_readiness_report() -> DependencyReadinessReport { let readiness = DependencyReadiness { storage_ready: collect_storage_readiness_uncached().await, @@ -485,21 +676,6 @@ async fn collect_lock_quorum_status() -> LockQuorumStatus { } } -async fn collect_dependency_readiness_uncached() -> DependencyReadiness { - let iam_ready_raw = runtime_sources::iam_ready(); - let storage_ready = collect_storage_readiness_uncached().await; - let lock_quorum_status = collect_lock_quorum_status_uncached().await; - - let readiness = DependencyReadiness { - storage_ready, - iam_ready: iam_ready_raw, - lock_quorum_ready: lock_quorum_status.ready, - }; - let report = dependency_readiness_report_from_readiness(readiness); - record_readiness_report(&report); - report.readiness -} - async fn collect_storage_readiness_uncached() -> bool { if let Some(store) = runtime_sources::object_store_handle() { let storage_info = StorageAdminApi::storage_info(store.as_ref()).await; @@ -509,6 +685,15 @@ async fn collect_storage_readiness_uncached() -> bool { } } +async fn collect_storage_read_readiness_uncached() -> bool { + if let Some(store) = runtime_sources::object_store_handle() { + let storage_info = StorageAdminApi::storage_info(store.as_ref()).await; + storage_read_ready_from_runtime_state(&storage_info) + } else { + false + } +} + fn set_lock_quorum_status(online_hosts: &HashSet, set_endpoints: &[Endpoint]) -> LockQuorumStatus { let total_clients = set_endpoints .iter() @@ -627,7 +812,7 @@ where loop { let readiness = load_readiness().await; - if readiness.storage_ready && readiness.iam_ready && readiness.lock_quorum_ready { + if readiness.storage_ready && readiness.iam_ready { on_ready(readiness); return Ok(()); } @@ -648,7 +833,7 @@ where storage_ready = readiness.storage_ready, iam_ready = readiness.iam_ready, lock_quorum_ready = readiness.lock_quorum_ready, - "Runtime readiness has not reached write quorum yet; delaying ready state publication" + "Runtime node readiness has not been reached yet; delaying ready state publication" ); tokio::time::sleep(poll_interval).await; } @@ -658,7 +843,10 @@ where mod tests { use super::*; use rustfs_madmin::{BackendInfo, Disk}; + use serial_test::serial; use std::future; + use std::sync::atomic::{AtomicUsize, Ordering}; + use temp_env::async_with_vars; #[test] fn startup_runtime_readiness_wait_constants_are_ordered() { @@ -667,6 +855,91 @@ mod tests { assert_eq!(STARTUP_RUNTIME_READINESS_POLL_INTERVAL.as_secs(), 1); } + #[tokio::test] + #[serial] + async fn cluster_health_report_singleflight_reuses_inflight_result() { + reset_cluster_health_report_caches().await; + async_with_vars( + [ + (rustfs_config::ENV_HEALTH_READINESS_CACHE_TTL_MS, Some("60000")), + (rustfs_config::ENV_HEALTH_CLUSTER_TIMEOUT_MS, Some("1000")), + ], + async { + let calls = Arc::new(AtomicUsize::new(0)); + let ready_report = DependencyReadinessReport { + readiness: DependencyReadiness { + storage_ready: true, + iam_ready: true, + lock_quorum_ready: true, + }, + degraded_reasons: Vec::new(), + }; + + let first_calls = calls.clone(); + let first_report = ready_report.clone(); + let first = collect_cluster_health_report_with(ClusterHealthProbeKind::Write, move || { + let first_calls = first_calls.clone(); + let first_report = first_report.clone(); + async move { + first_calls.fetch_add(1, Ordering::SeqCst); + tokio::time::sleep(Duration::from_millis(20)).await; + first_report + } + }); + + let second_calls = calls.clone(); + let second_report = ready_report.clone(); + let second = collect_cluster_health_report_with(ClusterHealthProbeKind::Write, move || { + let second_calls = second_calls.clone(); + let second_report = second_report.clone(); + async move { + second_calls.fetch_add(1, Ordering::SeqCst); + second_report + } + }); + + let (first, second) = tokio::join!(first, second); + + assert_eq!(calls.load(Ordering::SeqCst), 1); + assert_eq!(first, ready_report); + assert_eq!(second, ready_report); + }, + ) + .await; + reset_cluster_health_report_caches().await; + } + + #[tokio::test] + #[serial] + async fn cluster_health_report_timeout_is_sanitized() { + reset_cluster_health_report_caches().await; + async_with_vars( + [ + (rustfs_config::ENV_HEALTH_READINESS_CACHE_TTL_MS, Some("0")), + (rustfs_config::ENV_HEALTH_CLUSTER_TIMEOUT_MS, Some("1")), + ], + async { + let report = collect_cluster_health_report_with(ClusterHealthProbeKind::Read, || async { + tokio::time::sleep(Duration::from_millis(50)).await; + DependencyReadinessReport { + readiness: DependencyReadiness { + storage_ready: true, + iam_ready: true, + lock_quorum_ready: true, + }, + degraded_reasons: Vec::new(), + } + }) + .await; + + assert_eq!(report, cluster_health_timeout_report()); + assert_eq!(report.degraded_reasons, vec![ReadinessDegradedReason::ClusterHealthTimeout]); + }, + ) + .await; + reset_cluster_health_report_caches().await; + } + #[tokio::test] async fn wait_for_runtime_readiness_with_does_not_publish_ready_when_runtime_readiness_is_not_reached() { let readiness = GlobalReadiness::new(); @@ -696,11 +969,11 @@ mod tests { } #[tokio::test] - async fn wait_for_runtime_readiness_with_does_not_publish_ready_when_lock_quorum_is_not_reached() { + async fn wait_for_runtime_readiness_with_publishes_ready_without_lock_quorum() { let readiness = GlobalReadiness::new(); let state_manager = ServiceStateManager::new(); - let err = wait_for_runtime_readiness_with( + let result = wait_for_runtime_readiness_with( Duration::ZERO, Duration::from_millis(1), || { @@ -715,12 +988,11 @@ mod tests { state_manager.update(ServiceState::Ready); }, ) - .await - .expect_err("lock quorum should block readiness publication"); + .await; - assert!(err.to_string().contains("lock_quorum_ready=false")); - assert!(!readiness.is_ready()); - assert_eq!(state_manager.current_state(), ServiceState::Starting); + assert!(result.is_ok(), "lock quorum must not block node readiness publication"); + assert!(readiness.is_ready()); + assert_eq!(state_manager.current_state(), ServiceState::Ready); } #[test] @@ -775,6 +1047,33 @@ mod tests { assert!(storage_ready_from_runtime_state(&info)); } + #[test] + fn storage_read_ready_from_runtime_state_allows_read_quorum_below_write_quorum() { + let disks = (0..4) + .map(|disk_index| Disk { + endpoint: format!("127.0.0.1:900{disk_index}"), + drive_path: format!("/data{disk_index}"), + pool_index: 0, + set_index: 0, + disk_index, + state: if disk_index < 2 { "ok" } else { "offline" }.to_string(), + runtime_state: Some(if disk_index < 2 { "online" } else { "offline" }.to_string()), + ..Default::default() + }) + .collect::>(); + let info = StorageInfo { + backend: BackendInfo { + standard_sc_data: vec![2], + drives_per_set: vec![4], + ..Default::default() + }, + disks, + }; + + assert!(storage_read_ready_from_runtime_state(&info)); + assert!(!storage_ready_from_runtime_state(&info)); + } + #[test] fn storage_ready_from_runtime_state_deduplicates_duplicate_disk_rows() { let duplicate_disk = Disk { diff --git a/rustfs/src/storage/concurrency/manager.rs b/rustfs/src/storage/concurrency/manager.rs index 94db26185..5eed85c5b 100644 --- a/rustfs/src/storage/concurrency/manager.rs +++ b/rustfs/src/storage/concurrency/manager.rs @@ -575,6 +575,28 @@ impl ConcurrencyManager { } } + /// Get a read-only workload admission snapshot for foreground writes. + pub fn put_object_admission_snapshot(&self) -> WorkloadAdmissionSnapshot { + let active = PutObjectGuard::concurrent_count(); + let limit = self.scheduler_config.max_concurrent_reads; + let state = if limit == 0 { + AdmissionState::Disabled + } else if active >= limit { + AdmissionState::Saturated + } else { + AdmissionState::Open + }; + + let admission = + WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, state).with_counts(Some(active), None, Some(limit)); + + match state { + AdmissionState::Disabled => admission.with_reason("foreground write pressure tracking disabled"), + AdmissionState::Saturated => admission.with_reason("foreground write concurrency reached local pressure limit"), + _ => admission, + } + } + /// Get a read-only workload admission registry snapshot for local storage concurrency. pub fn workload_admission_registry_snapshot(&self) -> WorkloadAdmissionRegistrySnapshot { let entries = WorkloadClass::REQUIRED @@ -582,6 +604,7 @@ impl ConcurrencyManager { .copied() .map(|class| match class { WorkloadClass::ForegroundRead => self.get_object_admission_snapshot(), + WorkloadClass::ForegroundWrite => self.put_object_admission_snapshot(), class => WorkloadAdmissionSnapshot::new(class, AdmissionState::Unknown) .with_reason("not exposed by storage concurrency manager"), }) @@ -695,6 +718,28 @@ mod integration_tests { assert_eq!(snapshot.limit, Some(manager.scheduler_config().max_concurrent_reads)); } + #[tokio::test] + #[serial] + async fn test_concurrency_manager_workload_admission_snapshot_tracks_put_requests() { + crate::storage::concurrency::reset_active_put_requests(); + let manager = ConcurrencyManager::new(); + let initial = manager.put_object_admission_snapshot(); + + assert_eq!(initial.class, WorkloadClass::ForegroundWrite); + assert_eq!(initial.state, AdmissionState::Open); + assert_eq!(initial.active, Some(0)); + assert_eq!(initial.queued, None); + assert_eq!(initial.limit, Some(manager.scheduler_config().max_concurrent_reads)); + + let guard = PutObjectGuard::new(); + let snapshot = manager.put_object_admission_snapshot(); + + assert_eq!(snapshot.active, Some(1)); + assert_eq!(snapshot.limit, Some(manager.scheduler_config().max_concurrent_reads)); + drop(guard); + crate::storage::concurrency::reset_active_put_requests(); + } + #[tokio::test] #[serial] async fn test_concurrency_manager_workload_admission_registry_covers_required_classes() { @@ -713,6 +758,10 @@ mod integration_tests { registry.get(WorkloadClass::ForegroundRead).map(|snapshot| snapshot.state), Some(AdmissionState::Open) ); + assert_eq!( + registry.get(WorkloadClass::ForegroundWrite).map(|snapshot| snapshot.state), + Some(AdmissionState::Open) + ); assert_eq!( registry.get(WorkloadClass::Scanner).map(|snapshot| snapshot.state), Some(AdmissionState::Unknown) diff --git a/rustfs/src/workload_admission.rs b/rustfs/src/workload_admission.rs index eadc3ce17..1bd8d3dca 100644 --- a/rustfs/src/workload_admission.rs +++ b/rustfs/src/workload_admission.rs @@ -22,7 +22,6 @@ use rustfs_concurrency::{ use crate::storage_api::workload::concurrency::get_concurrency_manager; const BUCKET_METADATA_RUNTIME_NOT_INITIALIZED: &str = "bucket metadata runtime not initialized"; -const FOREGROUND_WRITE_NOT_EXPOSED_BY_PROVIDER: &str = "foreground write admission not yet exposed by RustFS runtime"; const HEAL_MANAGER_NOT_INITIALIZED: &str = "heal manager not initialized"; const REPAIR_QUEUE_BACKLOG_PRESENT: &str = "repair queue has pending work"; const REPLICATION_RUNTIME_NOT_INITIALIZED: &str = "replication runtime not initialized"; @@ -33,6 +32,8 @@ const SCANNER_ADMISSION_SATURATED: &str = "scanner active work reached configure const SCANNER_ACTIVITY_IDLE_OR_NOT_INITIALIZED: &str = "scanner activity idle or not initialized"; const STORAGE_CONCURRENCY_PROVIDER_MISSING_FOREGROUND_READ: &str = "storage concurrency provider did not expose foreground read admission"; +const STORAGE_CONCURRENCY_PROVIDER_MISSING_FOREGROUND_WRITE: &str = + "storage concurrency provider did not expose foreground write admission"; #[derive(Debug, Default, Clone, Copy)] pub struct RustFsWorkloadAdmissionSnapshotProvider; @@ -75,9 +76,13 @@ pub fn foreground_read_workload_admission_snapshot() -> WorkloadAdmissionSnapsho } pub fn foreground_write_workload_admission_snapshot() -> WorkloadAdmissionSnapshot { - WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, AdmissionState::Disabled) - .with_counts(Some(0), None, Some(0)) - .with_reason(FOREGROUND_WRITE_NOT_EXPOSED_BY_PROVIDER) + storage_concurrency_workload_admission_snapshot() + .get(WorkloadClass::ForegroundWrite) + .cloned() + .unwrap_or_else(|| { + WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundWrite, AdmissionState::Unknown) + .with_reason(STORAGE_CONCURRENCY_PROVIDER_MISSING_FOREGROUND_WRITE) + }) } pub fn metadata_workload_admission_snapshot() -> WorkloadAdmissionSnapshot { @@ -260,14 +265,17 @@ mod tests { } #[test] - fn foreground_write_snapshot_reports_explicit_gap_reason() { + fn foreground_write_snapshot_uses_storage_concurrency_provider_contract() { let snapshot = foreground_write_workload_admission_snapshot(); + let storage_snapshot = storage_concurrency_workload_admission_snapshot() + .get(WorkloadClass::ForegroundWrite) + .cloned(); + assert_eq!(Some(snapshot.clone()), storage_snapshot); assert_eq!(snapshot.class, WorkloadClass::ForegroundWrite); - assert_eq!(snapshot.state, AdmissionState::Disabled); - assert_eq!(snapshot.active, Some(0)); - assert_eq!(snapshot.limit, Some(0)); - assert_eq!(snapshot.reason.as_deref(), Some(FOREGROUND_WRITE_NOT_EXPOSED_BY_PROVIDER)); + assert_ne!(snapshot.state, AdmissionState::Unknown); + assert!(snapshot.active.is_some()); + assert!(snapshot.limit.is_some()); } #[test] @@ -386,6 +394,10 @@ mod tests { registry.get(WorkloadClass::ForegroundRead).map(|snapshot| snapshot.state), Some(AdmissionState::Unknown) ); + assert_ne!( + registry.get(WorkloadClass::ForegroundWrite).map(|snapshot| snapshot.state), + Some(AdmissionState::Unknown) + ); assert_eq!(registry.get(WorkloadClass::Scanner).and_then(|snapshot| snapshot.active), Some(0)); } @@ -401,10 +413,12 @@ mod tests { assert_eq!(registry.get(WorkloadClass::Metadata), Some(&metadata_workload_admission_snapshot())); assert_eq!(registry.get(WorkloadClass::Scanner), Some(&scanner_workload_admission_snapshot())); assert_eq!( - registry - .get(WorkloadClass::ForegroundWrite) - .map(|snapshot| (snapshot.state, snapshot.reason.as_deref())), - Some((AdmissionState::Disabled, Some(FOREGROUND_WRITE_NOT_EXPOSED_BY_PROVIDER))) + registry.get(WorkloadClass::ForegroundWrite).map(|snapshot| snapshot.class), + Some(WorkloadClass::ForegroundWrite) + ); + assert_eq!( + registry.get(WorkloadClass::ForegroundWrite), + storage_registry.get(WorkloadClass::ForegroundWrite) ); } }