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

fix: decouple readiness from cluster health (#3912)

Browse Source 
* 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 <housemecn@gmail.com>
This commit is contained in:
cxymds
2026-06-26 21:55:18 +08:00
committed by GitHub
parent 6469d6ada8
commit 7820a55fdc
14 changed files with 769 additions and 165 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/config/src/constants/heal.rs
+7
View File
@@ -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;
crates/config/src/constants/health.rs
+5
View File
@@ -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";
crates/ecstore/src/data_movement_backpressure.rs
+115 -37
View File
@@ -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<Instant> = 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<usize> {
if !config.enabled || config.foreground_read_high_percent == 0 {
) -> Option<ForegroundPressure> {
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<Instant>,
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]
crates/heal/src/heal/manager.rs
+121 -36
View File
@@ -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<WorkloadSnapshotProviderRef>) -> Option<usize> {
if !config.mainline_throttle_enable || config.mainline_read_utilization_high_percent == 0 {
fn mainline_throttle_active(
config: &HealConfig,
provider: &Option<WorkloadSnapshotProviderRef>,
) -> Option<ForegroundPressure> {
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<Notify>, 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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<dyn HealStorageAPI> = 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()
}),
rustfs/src/admin/console.rs
+2 -6
View File
@@ -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()
rustfs/src/admin/handlers/health.rs
+71 -16
View File
@@ -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<crate::server::DependencyReadinessReport> {
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<HealthReadinessSource> {
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]
rustfs/src/admin/handlers/system.rs
+1 -15
View File
@@ -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());
}
rustfs/src/server/compress.rs
+2
View File
@@ -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]
rustfs/src/server/layer.rs
+33 -9
View File
@@ -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<HealthProb
match path {
HEALTH_PREFIX | HEALTH_COMPAT_LIVE_PATH | MINIO_HEALTH_LIVE_PATH => 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::<Bytes>::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() {
rustfs/src/server/mod.rs
+5 -4
View File
@@ -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);
rustfs/src/server/prefix.rs
+3
View File
@@ -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.
rustfs/src/server/readiness.rs
+328 -29
View File
@@ -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<Option<StorageReadinessCacheEntry>> {
static CACHE: OnceLock<Mutex<Option<StorageReadinessCacheEntry>>> = OnceLock::new();
CACHE.get_or_init(|| Mutex::new(None))
@@ -266,6 +291,46 @@ fn lock_quorum_status_cache() -> &'static Mutex<Option<LockQuorumCacheEntry>> {
CACHE.get_or_init(|| Mutex::new(None))
}
fn cluster_write_health_report_cache() -> &'static Mutex<Option<ClusterHealthReportCacheEntry>> {
static CACHE: OnceLock<Mutex<Option<ClusterHealthReportCacheEntry>>> = OnceLock::new();
CACHE.get_or_init(|| Mutex::new(None))
}
fn cluster_read_health_report_cache() -> &'static Mutex<Option<ClusterHealthReportCacheEntry>> {
static CACHE: OnceLock<Mutex<Option<ClusterHealthReportCacheEntry>>> = OnceLock::new();
CACHE.get_or_init(|| Mutex::new(None))
}
fn cluster_write_health_singleflight() -> &'static Mutex<()> {
static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
LOCK.get_or_init(|| Mutex::new(()))
}
fn cluster_read_health_singleflight() -> &'static Mutex<()> {
static LOCK: OnceLock<Mutex<()>> = OnceLock::new();
LOCK.get_or_init(|| Mutex::new(()))
}
fn cluster_health_report_cache(kind: ClusterHealthProbeKind) -> &'static Mutex<Option<ClusterHealthReportCacheEntry>> {
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<bool> {
let ttl = health_readiness_cache_ttl();
if ttl.is_zero() {
@@ -281,6 +346,33 @@ async fn load_cached_storage_readiness() -> Option<bool> {
None
}
async fn load_cached_cluster_health_report(kind: ClusterHealthProbeKind) -> Option<DependencyReadinessReport> {
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<F>(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<ReadinessDegradedReason> {
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<LoadFn, Fut>(
kind: ClusterHealthProbeKind,
mut load_report: LoadFn,
) -> DependencyReadinessReport
where
LoadFn: FnMut() -> Fut,
Fut: Future<Output = DependencyReadinessReport>,
{
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<String>, 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::<Vec<_>>();
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 {
rustfs/src/storage/concurrency/manager.rs
+49
View File
@@ -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)
rustfs/src/workload_admission.rs
+27 -13
View File
@@ -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)
);
}
}