From e1a4b9e0b6c83efff8008c8b394a7d156bb435f4 Mon Sep 17 00:00:00 2001 From: Zhengchao An Date: Sat, 27 Jun 2026 10:51:06 +0800 Subject: [PATCH] refactor: batch cluster lock and health readiness (#3936) --- crates/config/src/constants/health.rs | 5 + crates/ecstore/src/core/sets.rs | 19 +-- crates/ecstore/src/runtime/sources.rs | 76 +++++++++- docs/architecture/migration-progress.md | 78 +++++++--- rustfs/src/admin/handlers/cluster_snapshot.rs | 4 + rustfs/src/admin/handlers/health.rs | 23 ++- rustfs/src/cluster_snapshot.rs | 12 +- rustfs/src/server/health.rs | 17 ++- rustfs/src/server/readiness.rs | 137 ++++++++++++++++-- 9 files changed, 312 insertions(+), 59 deletions(-) diff --git a/crates/config/src/constants/health.rs b/crates/config/src/constants/health.rs index 628d0f443..b54922797 100644 --- a/crates/config/src/constants/health.rs +++ b/crates/config/src/constants/health.rs @@ -48,3 +48,8 @@ pub const DEFAULT_HEALTH_COMPAT_BUSY_MAX_ACTIVE_REQUESTS: usize = 0; /// in running state if a global KMS manager exists. pub const ENV_HEALTH_COMPAT_KMS_READY_CHECK_ENABLE: &str = "RUSTFS_HEALTH_COMPAT_KMS_READY_CHECK_ENABLE"; pub const DEFAULT_HEALTH_COMPAT_KMS_READY_CHECK_ENABLE: bool = false; + +/// Enable peer-health readiness impact. +/// When disabled, peer-health state is reported but does not affect readiness. +pub const ENV_HEALTH_PEER_READY_CHECK_ENABLE: &str = "RUSTFS_HEALTH_PEER_READY_CHECK_ENABLE"; +pub const DEFAULT_HEALTH_PEER_READY_CHECK_ENABLE: bool = false; diff --git a/crates/ecstore/src/core/sets.rs b/crates/ecstore/src/core/sets.rs index 2b35b5678..5f765621d 100644 --- a/crates/ecstore/src/core/sets.rs +++ b/crates/ecstore/src/core/sets.rs @@ -46,7 +46,6 @@ use rustfs_common::heal_channel::HealOpts; use rustfs_common::heal_channel::{DriveState, HealItemType}; use rustfs_filemeta::FileInfo; use rustfs_lock::NamespaceLockWrapper; -use rustfs_lock::client::LockClient; use rustfs_madmin::heal_commands::{HealDriveInfo, HealResultItem}; use rustfs_utils::{crc_hash, path::path_join_buf, sip_hash}; use std::{collections::HashMap, sync::Arc}; @@ -102,28 +101,17 @@ impl Sets { let mut disk_set = Vec::with_capacity(set_count); - // Get lock clients from global storage - let lock_clients = runtime_sources::global_lock_clients(); + let lock_registry = runtime_sources::lock_registry(); for i in 0..set_count { let mut set_drive = Vec::with_capacity(set_drive_count); let mut set_endpoints = Vec::with_capacity(set_drive_count); - let mut set_lock_clients: HashMap> = HashMap::new(); for j in 0..set_drive_count { let idx = i * set_drive_count + j; let mut disk = disks[idx].clone(); let endpoint = endpoints.endpoints.as_ref()[idx].clone(); - if let Some(lock_clients_map) = lock_clients { - let host_port = endpoint.host_port(); - if let Some(lock_client) = lock_clients_map.get(&host_port) - && !set_lock_clients.contains_key(&host_port) - { - set_lock_clients.insert(host_port, lock_client.clone()); - } - } - set_endpoints.push(endpoint); if disk.is_none() { @@ -164,7 +152,10 @@ impl Sets { } } - let lockers = set_lock_clients.values().cloned().collect::>>(); + let lockers = lock_registry + .as_ref() + .map(|registry| registry.clients_for_endpoints(&set_endpoints)) + .unwrap_or_default(); let set_disks = SetDisks::new( runtime_sources::local_node_name().await, Arc::new(RwLock::new(set_drive)), diff --git a/crates/ecstore/src/runtime/sources.rs b/crates/ecstore/src/runtime/sources.rs index 948fb4d40..c5d9c60bc 100644 --- a/crates/ecstore/src/runtime/sources.rs +++ b/crates/ecstore/src/runtime/sources.rs @@ -13,7 +13,7 @@ // limitations under the License. use std::{ - collections::HashMap, + collections::{HashMap, HashSet}, sync::{Arc, OnceLock}, time::SystemTime, }; @@ -59,6 +59,35 @@ const TEST_RPC_SECRET: &str = "test-rpc-secret"; pub(crate) type WorkloadSnapshotProviderRef = Arc; +#[derive(Clone, Default)] +pub(crate) struct LockRegistry { + clients: HashMap>, +} + +impl LockRegistry { + pub(crate) fn new(clients: HashMap>) -> Self { + Self { clients } + } + + pub(crate) fn clients_for_endpoints(&self, endpoints: &[Endpoint]) -> Vec> { + let mut seen_hosts = HashSet::with_capacity(endpoints.len()); + let mut clients = Vec::with_capacity(endpoints.len()); + + for endpoint in endpoints { + let host_port = endpoint.host_port(); + if host_port.is_empty() || !seen_hosts.insert(host_port.clone()) { + continue; + } + + if let Some(client) = self.clients.get(&host_port) { + clients.push(client.clone()); + } + } + + clients + } +} + static WORKLOAD_ADMISSION_SNAPSHOT_PROVIDER: OnceLock = OnceLock::new(); pub(crate) fn set_workload_admission_snapshot_provider( @@ -256,6 +285,11 @@ pub(crate) fn global_lock_clients() -> Option<&'static HashMap Option { + global_lock_clients() + .map(|clients| LockRegistry::new(clients.iter().map(|(host, client)| (host.clone(), client.clone())).collect())) +} + pub(crate) fn set_primary_lock_client(client: Arc) -> std::result::Result<(), Arc> { set_global_lock_client(client) } @@ -469,3 +503,43 @@ pub(crate) async fn initialize_local_disk_maps(endpoint_pools: EndpointServerPoo pub(crate) async fn init_tier_config_mgr(store: Arc) -> Result<()> { GLOBAL_TierConfigMgr.write().await.init(store).await } + +#[cfg(test)] +mod tests { + use super::LockRegistry; + use crate::disk::endpoint::Endpoint; + use rustfs_lock::{LocalClient, LockClient}; + use std::{collections::HashMap, sync::Arc}; + + fn url_endpoint(raw: &str) -> Endpoint { + Endpoint { + url: url::Url::parse(raw).expect("test endpoint url"), + is_local: false, + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + } + } + + #[test] + fn lock_registry_selects_unique_clients_in_endpoint_order() { + let client_a: Arc = Arc::new(LocalClient::new()); + let client_b: Arc = Arc::new(LocalClient::new()); + let registry = LockRegistry::new(HashMap::from([ + ("node-a:9000".to_string(), client_a.clone()), + ("node-b:9000".to_string(), client_b.clone()), + ])); + let endpoints = vec![ + url_endpoint("http://node-a:9000/data-a"), + url_endpoint("http://node-a:9000/data-b"), + url_endpoint("http://node-missing:9000/data"), + url_endpoint("http://node-b:9000/data"), + ]; + + let clients = registry.clients_for_endpoints(&endpoints); + + assert_eq!(clients.len(), 2); + assert!(Arc::ptr_eq(&clients[0], &client_a)); + assert!(Arc::ptr_eq(&clients[1], &client_b)); + } +} diff --git a/docs/architecture/migration-progress.md b/docs/architecture/migration-progress.md index d3f82ff0e..5ca8446b7 100644 --- a/docs/architecture/migration-progress.md +++ b/docs/architecture/migration-progress.md @@ -5,28 +5,23 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block ## Current Context - Issue: [`rustfs/backlog#660`](https://github.com/rustfs/backlog/issues/660) -- Branch: `overtrue/arch-cluster-readonly-control-plane-batch` +- Branch: `overtrue/arch-cluster-lock-health-phase` - Baseline: completed `C-011/C-012/C-013/API-055/API-059/API-079/API-080/API-081/API-082/API-083/API-084/API-085/API-086/API-087/API-088/API-089/API-090/API-091/API-092/API-093/API-094/API-095/API-096/API-097/API-098/API-099/API-100/API-101/API-102/API-103/API-104/API-105/API-106/API-107/API-108/API-109/API-110/API-111/API-112/API-113/API-114/API-115/API-116/API-117/API-118/API-119/API-120/API-121/API-122/API-123/API-124/API-125/API-126/API-127/API-128/API-129/API-130/API-131/API-132/API-133/API-134/API-135/API-136/API-137/API-138/API-139/API-140/API-141/API-142/API-143/API-144/API-145/API-146/API-147/API-148/API-149/API-150/API-151/API-152/API-153/API-154/API-155/API-156/API-157/API-158/API-159/API-160/API-161/API-162/API-163/API-164/API-165/API-166/API-167/API-168/API-169/API-170/API-171/API-172/API-173/API-174/API-175/API-176/API-177/API-178/API-179/API-180/API-181/API-182/API-183/API-184/API-185/API-186/API-187/API-188/API-189/API-190/API-191/API-192/API-193/API-194/API-195/API-196/API-197/API-198/API-199/API-200/API-201/API-202/API-203/API-204/API-205/API-206/API-207/API-208/API-209/API-210/API-211/API-212/API-213/API-214/API-215/API-216/API-217/API-218/API-219/API-220/API-221/API-222/API-223/API-224/API-225/API-226/API-227/API-228/API-229/API-230/API-231/API-232/API-233/API-234/API-235/API-236/API-237/API-238/API-239/API-240/API-241/API-242/API-243/API-244/API-245/API-246/API-247/API-248/API-249/API-250/API-251/API-252/API-253/API-254/CTX-002`. - Current baseline also includes API-255 from PR #3923, API-256 from PR - #3925, and CFG-009 from PR #3927. -- Current phase PR: C-007/C-009 cluster read-only control-plane consumer, RPC - boundary, and readiness invariant batch. -- Based on: E-035/E-008 branch while PR #3933 and the E-031 through E-035 - follow-up ECStore owner batch are pending; rebase onto `origin/main` after - those PRs merge before opening this PR. + #3925, CFG-009 from PR #3927, and C-007/C-009 from PR #3935. +- Current phase PR: C-008/C-010 cluster lock-registry consumption and gated + peer-health readiness impact batch. +- Based on: `origin/main` after PR #3935 merged. - PR type for this branch: `contract`. -- Runtime behavior changes: admin readiness reads keep the existing readiness - report collector and, when endpoint context is available, project that same - report through the read-only cluster snapshot runtime status before returning - readiness; `FullReady` publication now uses the existing storage, IAM, and - lock quorum readiness invariant. -- Rust code changes: add a read-only cluster RPC boundary snapshot that models - metadata/lock/health/admin control RPC as gRPC and remote-disk streams as the - internode data transport, expose it through the ECStore public cluster facade, - include it in RustFS cluster snapshots and admin JSON views, route admin - readiness usecase reads through the snapshot runtime status using the existing - readiness report collector, and require lock quorum before publishing - `FullReady`. +- Runtime behavior changes: lock client selection now consumes a runtime + `LockRegistry` snapshot instead of directly scanning global clients inside + each set; peer-health readiness impact is gated by + `RUSTFS_HEALTH_PEER_READY_CHECK_ENABLE` and remains disabled by default. +- Rust code changes: add the ECStore lock-registry runtime source, route + `Sets::new` through endpoint-ordered registry selection, add a disabled-by + default peer-health readiness gate, carry the peer-health readiness bit + through health and cluster runtime status views, and cover default-disabled + and enabled-degraded readiness behavior. - CI/script changes: guard that the ECStore RPC boundary snapshot and internode data-transport control-plane separation note remain present; reject restoring ECStore root `store.rs`, `set_disk.rs`, @@ -1409,6 +1404,20 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block coverage, migration guard, formatting, diff hygiene, risk scan, PR quality gate, and three-expert review. +- [x] `C-008` Migrate lock group consumption. + - Completed slice: add an ECStore runtime `LockRegistry` snapshot and route + erasure set lock-client selection through endpoint-ordered registry + consumption. + - Acceptance: lock clients are selected by endpoint host-port in set order, + duplicate endpoint entries are deduplicated, missing hosts are skipped, and + existing host-port, quorum, connected-count, and client-count semantics stay + intact. + - Must preserve: remote lock behavior, namespace lock quorum checks, endpoint + layout ownership, and object hot-path behavior. + - Verification: focused ECStore lock-registry test, compile coverage, + formatting, diff hygiene, risk scan, PR quality gate, and three-expert + review. + - [x] `C-009` Model control RPC separately. - Completed slice: add a read-only RPC boundary snapshot that keeps metadata, lock, health, and administrative control RPC on the gRPC control @@ -1423,6 +1432,20 @@ Status values: `[ ]` not started, `[~]` in progress, `[x]` complete, `[!]` block compile coverage, migration guard, formatting, diff hygiene, risk scan, PR quality gate, and three-expert review. +- [x] `C-010` Peer health impact gate. + - Completed slice: add a disabled-by-default peer-health readiness gate that + consumes the read-only peer-health snapshot before impacting runtime + readiness. + - Acceptance: default-disabled behavior preserves existing readiness results; + when enabled, unsupported or unknown peer health degrades readiness with an + explicit `peer_health_unavailable` reason. + - Must preserve: object hot paths, peer-health snapshot read-only behavior, + storage/IAM/lock readiness behavior when the gate is disabled, and existing + health payload compatibility. + - Verification: focused RustFS readiness, health, and cluster snapshot tests, + compile coverage, formatting, diff hygiene, risk scan, PR quality gate, and + three-expert review. + - [x] `TEST-PRTYPE-001` Check PR type enum consistency. - Acceptance: `./scripts/check_architecture_migration_rules.sh` parses the allowed PR types from [`crate-boundaries.md`](crate-boundaries.md) and fails @@ -9398,6 +9421,23 @@ Notes: public APIs, boxed public errors, production println/eprintln, or relaxed ordering introduced in changed Rust files. +- Issue #660 C-008/C-010 current slice: + - `cargo test -p rustfs-ecstore --lib lock_registry`: passed, 1 passed. + - `cargo test -p rustfs --lib readiness`: passed, 33 passed. + - `cargo test -p rustfs --lib health`: passed, 41 passed. + - `cargo test -p rustfs --lib cluster_snapshot`: passed, 9 passed. + - `cargo check -p rustfs-ecstore --lib`: passed. + - `cargo check -p rustfs --lib`: passed. + - `cargo fmt --all --check`: passed. + - `./scripts/check_architecture_migration_rules.sh`: passed. + - `./scripts/check_layer_dependencies.sh`: passed. + - `git diff --check`: passed. + - `make pre-pr`: passed. + - Rust risk scan: passed; only new `.expect()` matches are focused test URL + builders, with no new production unwrap/expect, numeric casts, string error + public APIs, boxed public errors, production println/eprintln, or relaxed + ordering introduced in changed Rust files. + ## Handoff Notes - Continue with larger consumer-migration batches outside the cleaned diff --git a/rustfs/src/admin/handlers/cluster_snapshot.rs b/rustfs/src/admin/handlers/cluster_snapshot.rs index 63db2ac84..10886f575 100644 --- a/rustfs/src/admin/handlers/cluster_snapshot.rs +++ b/rustfs/src/admin/handlers/cluster_snapshot.rs @@ -434,6 +434,7 @@ pub(crate) struct ClusterRuntimeStatusView { pub storage_ready: bool, pub iam_ready: bool, pub lock_quorum_ready: bool, + pub peer_health_ready: bool, pub degraded_reasons: Vec<&'static str>, } @@ -444,6 +445,7 @@ impl From for ClusterRuntimeStatusView { storage_ready: runtime.readiness.storage_ready, iam_ready: runtime.readiness.iam_ready, lock_quorum_ready: runtime.readiness.lock_quorum_ready, + peer_health_ready: runtime.readiness.peer_health_ready, degraded_reasons: runtime.degraded_reasons.into_iter().map(|reason| reason.as_str()).collect(), } } @@ -767,6 +769,7 @@ mod tests { storage_ready: false, iam_ready: true, lock_quorum_ready: false, + peer_health_ready: true, }, state: ClusterRuntimeReadinessState::Degraded, degraded_reasons: vec![ReadinessDegradedReason::StorageAndLockUnavailable], @@ -808,6 +811,7 @@ mod tests { storage_ready: true, iam_ready: true, lock_quorum_ready: true, + peer_health_ready: true, }, state: ClusterRuntimeReadinessState::Ready, degraded_reasons: Vec::new(), diff --git a/rustfs/src/admin/handlers/health.rs b/rustfs/src/admin/handlers/health.rs index 207b3257e..855195d5e 100644 --- a/rustfs/src/admin/handlers/health.rs +++ b/rustfs/src/admin/handlers/health.rs @@ -92,7 +92,7 @@ mod tests { #[test] fn test_readiness_state_ready() { - let state = health_check_state(true, true, true, HealthProbe::Readiness); + let state = health_check_state(true, true, true, true, HealthProbe::Readiness); assert_eq!(state.status_code, StatusCode::OK); assert_eq!(state.status, "ok"); assert!(state.ready); @@ -100,7 +100,7 @@ mod tests { #[test] fn test_readiness_state_storage_not_ready() { - let state = health_check_state(false, true, true, HealthProbe::Readiness); + let state = health_check_state(false, true, true, true, HealthProbe::Readiness); assert_eq!(state.status_code, StatusCode::SERVICE_UNAVAILABLE); assert_eq!(state.status, "degraded"); assert!(!state.ready); @@ -108,7 +108,7 @@ mod tests { #[test] fn test_liveness_state_iam_not_ready() { - let state = health_check_state(true, false, true, HealthProbe::Liveness); + let state = health_check_state(true, false, true, true, HealthProbe::Liveness); assert_eq!(state.status_code, StatusCode::OK); assert_eq!(state.status, "ok"); assert!(state.ready); @@ -116,7 +116,7 @@ mod tests { #[test] fn test_readiness_state_iam_not_ready() { - let state = health_check_state(true, false, true, HealthProbe::Readiness); + let state = health_check_state(true, false, true, true, HealthProbe::Readiness); assert_eq!(state.status_code, StatusCode::SERVICE_UNAVAILABLE); assert_eq!(state.status, "degraded"); assert!(!state.ready); @@ -124,7 +124,7 @@ mod tests { #[test] fn test_readiness_state_lock_not_ready() { - let state = health_check_state(true, true, false, HealthProbe::Readiness); + let state = health_check_state(true, true, false, true, HealthProbe::Readiness); assert_eq!(state.status_code, StatusCode::OK); assert_eq!(state.status, "ok"); assert!(state.ready); @@ -132,7 +132,7 @@ mod tests { #[test] fn test_cluster_write_state_lock_not_ready() { - let state = health_check_state(true, true, false, HealthProbe::ClusterWrite); + let state = health_check_state(true, true, false, true, HealthProbe::ClusterWrite); assert_eq!(state.status_code, StatusCode::SERVICE_UNAVAILABLE); assert_eq!(state.status, "degraded"); assert!(!state.ready); @@ -197,6 +197,7 @@ mod tests { storage_ready: false, iam_ready: true, lock_quorum_ready: true, + peer_health_ready: true, }, degraded_reasons: vec![crate::server::ReadinessDegradedReason::StorageQuorumUnavailable], }; @@ -218,6 +219,7 @@ mod tests { storage_ready: true, iam_ready: true, lock_quorum_ready: true, + peer_health_ready: true, }, degraded_reasons: Vec::new(), }; @@ -239,6 +241,7 @@ mod tests { storage_ready: false, iam_ready: false, lock_quorum_ready: false, + peer_health_ready: true, }, degraded_reasons: vec![crate::server::ReadinessDegradedReason::StorageAndIamUnavailable], }; @@ -265,6 +268,7 @@ mod tests { storage_ready: false, iam_ready: false, lock_quorum_ready: false, + peer_health_ready: true, }, degraded_reasons: vec![crate::server::ReadinessDegradedReason::StorageAndIamUnavailable], }; @@ -282,7 +286,7 @@ mod tests { #[test] fn test_build_health_payload_minimal_mode_returns_status_and_ready_only() { - let health = health_check_state(true, false, true, HealthProbe::Readiness); + let health = health_check_state(true, false, true, true, HealthProbe::Readiness); with_var(rustfs_config::ENV_HEALTH_MINIMAL_RESPONSE_ENABLE, Some("true"), || { let payload = build_health_payload(HealthPayloadContext { health, @@ -306,7 +310,7 @@ mod tests { #[test] fn test_build_health_payload_includes_degraded_reasons() { - let health = health_check_state(false, false, false, HealthProbe::Readiness); + let health = health_check_state(false, false, false, true, HealthProbe::Readiness); let payload = build_health_payload(HealthPayloadContext { health, storage_ready: false, @@ -328,6 +332,7 @@ mod tests { storage_ready: true, iam_ready: true, lock_quorum_ready: true, + peer_health_ready: true, }, degraded_reasons: Vec::new(), }; @@ -344,6 +349,7 @@ mod tests { storage_ready: false, iam_ready: true, lock_quorum_ready: true, + peer_health_ready: true, }, degraded_reasons: vec![crate::server::ReadinessDegradedReason::StorageQuorumUnavailable], }; @@ -363,6 +369,7 @@ mod tests { storage_ready: true, iam_ready: true, lock_quorum_ready: true, + peer_health_ready: true, }, degraded_reasons: Vec::new(), }; diff --git a/rustfs/src/cluster_snapshot.rs b/rustfs/src/cluster_snapshot.rs index e647d1726..0cae00cf5 100644 --- a/rustfs/src/cluster_snapshot.rs +++ b/rustfs/src/cluster_snapshot.rs @@ -55,7 +55,11 @@ pub enum ClusterRuntimeReadinessState { impl ClusterRuntimeReadinessState { fn from_report(report: &DependencyReadinessReport) -> Self { - if report.readiness.storage_ready && report.readiness.iam_ready && report.readiness.lock_quorum_ready { + if report.readiness.storage_ready + && report.readiness.iam_ready + && report.readiness.lock_quorum_ready + && report.readiness.peer_health_ready + { Self::Ready } else if report.degraded_reasons.is_empty() { Self::Unknown @@ -131,6 +135,7 @@ mod tests { storage_ready: true, iam_ready: true, lock_quorum_ready: true, + peer_health_ready: true, }, degraded_reasons: Vec::new(), }); @@ -141,6 +146,7 @@ mod tests { storage_ready: false, iam_ready: true, lock_quorum_ready: true, + peer_health_ready: true, }, degraded_reasons: vec![ReadinessDegradedReason::StorageQuorumUnavailable], }); @@ -156,6 +162,7 @@ mod tests { storage_ready: false, iam_ready: true, lock_quorum_ready: false, + peer_health_ready: true, }, degraded_reasons: vec![ReadinessDegradedReason::StorageAndLockUnavailable], }); @@ -208,6 +215,7 @@ mod tests { storage_ready: true, iam_ready: true, lock_quorum_ready: true, + peer_health_ready: true, }, state: ClusterRuntimeReadinessState::Ready, degraded_reasons: Vec::new(), @@ -221,6 +229,7 @@ mod tests { storage_ready: false, iam_ready: true, lock_quorum_ready: true, + peer_health_ready: true, }, state: ClusterRuntimeReadinessState::Degraded, degraded_reasons: vec![ReadinessDegradedReason::StorageQuorumUnavailable], @@ -270,6 +279,7 @@ mod tests { storage_ready: true, iam_ready: true, lock_quorum_ready: true, + peer_health_ready: true, }, state: ClusterRuntimeReadinessState::Ready, degraded_reasons: Vec::new(), diff --git a/rustfs/src/server/health.rs b/rustfs/src/server/health.rs index 72e451bbb..8fe315dc9 100644 --- a/rustfs/src/server/health.rs +++ b/rustfs/src/server/health.rs @@ -88,6 +88,7 @@ pub(crate) fn health_check_state( storage_ready: bool, iam_ready: bool, lock_quorum_ready: bool, + peer_health_ready: bool, probe: HealthProbe, ) -> HealthCheckState { if probe == HealthProbe::Liveness { @@ -98,7 +99,7 @@ pub(crate) fn health_check_state( }; } - let ready = storage_ready && iam_ready && (!probe.requires_lock_quorum() || lock_quorum_ready); + let ready = storage_ready && iam_ready && peer_health_ready && (!probe.requires_lock_quorum() || lock_quorum_ready); let status = if ready { "ok" } else { "degraded" }; let status_code = if ready { @@ -184,11 +185,12 @@ pub(crate) fn build_health_response_parts( 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; + let peer_health_ready = readiness_report.readiness.peer_health_ready; ( storage_ready, iam_ready, lock_quorum_ready, - health_check_state(storage_ready, iam_ready, lock_quorum_ready, probe), + health_check_state(storage_ready, iam_ready, lock_quorum_ready, peer_health_ready, probe), readiness_report.degraded_reasons.clone(), true, ) @@ -205,9 +207,14 @@ pub(crate) fn build_health_response_parts( vec![ReadinessDegradedReason::StorageIamAndLockUnavailable], true, ), - (HealthProbe::Liveness, _) => { - (false, false, false, health_check_state(false, false, false, probe), Vec::new(), false) - } + (HealthProbe::Liveness, _) => ( + false, + false, + false, + health_check_state(false, false, false, false, probe), + Vec::new(), + false, + ), }; if probe == HealthProbe::Readiness && matches!(kms_ready, Some(false)) { diff --git a/rustfs/src/server/readiness.rs b/rustfs/src/server/readiness.rs index b3598ada4..46459d1e7 100644 --- a/rustfs/src/server/readiness.rs +++ b/rustfs/src/server/readiness.rs @@ -15,6 +15,7 @@ use crate::server::runtime_sources; use crate::server::{ServiceState, ServiceStateManager}; use crate::server::{has_path_prefix, is_table_catalog_path}; +use crate::storage_api::cluster::control_plane::ClusterControlPlane; use crate::storage_api::server::readiness::contract::admin::StorageAdminApi; use crate::storage_api::server::readiness::{Endpoint, EndpointServerPools, is_dist_erasure}; #[cfg(test)] @@ -51,6 +52,7 @@ pub struct DependencyReadiness { pub storage_ready: bool, pub iam_ready: bool, pub lock_quorum_ready: bool, + pub peer_health_ready: bool, } #[derive(Debug, Clone, Copy, PartialEq, Eq)] @@ -60,6 +62,7 @@ pub enum ReadinessDegradedReason { LockQuorumUnavailable, KmsNotReady, ClusterHealthTimeout, + PeerHealthUnavailable, StorageAndIamUnavailable, StorageAndLockUnavailable, IamAndLockUnavailable, @@ -74,6 +77,7 @@ impl ReadinessDegradedReason { ReadinessDegradedReason::LockQuorumUnavailable => "lock_quorum_unavailable", ReadinessDegradedReason::KmsNotReady => "kms_not_ready", ReadinessDegradedReason::ClusterHealthTimeout => "cluster_health_timeout", + ReadinessDegradedReason::PeerHealthUnavailable => "peer_health_unavailable", ReadinessDegradedReason::StorageAndIamUnavailable => "storage_and_iam_unavailable", ReadinessDegradedReason::StorageAndLockUnavailable => "storage_and_lock_unavailable", ReadinessDegradedReason::IamAndLockUnavailable => "iam_and_lock_unavailable", @@ -223,6 +227,7 @@ pub async fn publish_ready_when_runtime_ready( storage_ready = dependency_readiness.storage_ready, iam_ready = dependency_readiness.iam_ready, lock_quorum_ready = dependency_readiness.lock_quorum_ready, + peer_health_ready = dependency_readiness.peer_health_ready, "Runtime node readiness reached; publishing ready state" ); }, @@ -531,7 +536,36 @@ 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 { +fn peer_health_ready_check_enabled() -> bool { + rustfs_utils::get_env_bool( + rustfs_config::ENV_HEALTH_PEER_READY_CHECK_ENABLE, + rustfs_config::DEFAULT_HEALTH_PEER_READY_CHECK_ENABLE, + ) +} + +fn peer_health_ready_from_endpoint_pools(endpoint_pools: &EndpointServerPools) -> bool { + if !peer_health_ready_check_enabled() { + return true; + } + + ClusterControlPlane::new(endpoint_pools.clone()) + .peer_health_snapshot() + .peers + .iter() + .all(|peer| peer.status.state.is_supported()) +} + +fn collect_peer_health_readiness() -> bool { + if !peer_health_ready_check_enabled() { + return true; + } + + runtime_sources::endpoints_handle() + .as_ref() + .is_some_and(peer_health_ready_from_endpoint_pools) +} + +fn base_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(), (false, false, false) => vec![ReadinessDegradedReason::StorageIamAndLockUnavailable], @@ -544,8 +578,19 @@ fn degraded_reasons(storage_ready: bool, iam_ready_raw: bool, lock_quorum_ready: } } +fn degraded_reasons(readiness: DependencyReadiness) -> Vec { + let mut reasons = base_degraded_reasons(readiness.storage_ready, readiness.iam_ready, readiness.lock_quorum_ready); + if !readiness.peer_health_ready { + reasons.push(ReadinessDegradedReason::PeerHealthUnavailable); + } + reasons +} + fn record_readiness_report(report: &DependencyReadinessReport) { - let ready = report.readiness.storage_ready && report.readiness.iam_ready && report.readiness.lock_quorum_ready; + let ready = report.readiness.storage_ready + && report.readiness.iam_ready + && report.readiness.lock_quorum_ready + && report.readiness.peer_health_ready; gauge!(METRIC_RUNTIME_READINESS_READY).set(if ready { 1.0 } else { 0.0 }); for reason in &report.degraded_reasons { counter!(METRIC_RUNTIME_READINESS_DEGRADED_TOTAL, "reason" => reason.as_str()).increment(1); @@ -554,7 +599,7 @@ fn record_readiness_report(report: &DependencyReadinessReport) { fn dependency_readiness_report_from_readiness(readiness: DependencyReadiness) -> DependencyReadinessReport { DependencyReadinessReport { - degraded_reasons: degraded_reasons(readiness.storage_ready, readiness.iam_ready, readiness.lock_quorum_ready), + degraded_reasons: degraded_reasons(readiness), readiness, } } @@ -574,6 +619,7 @@ pub async fn collect_dependency_readiness_report() -> DependencyReadinessReport storage_ready, iam_ready: iam_ready_raw, lock_quorum_ready: lock_quorum_status.ready, + peer_health_ready: collect_peer_health_readiness(), }; let report = dependency_readiness_report_from_readiness(readiness); record_readiness_report(&report); @@ -593,6 +639,7 @@ pub async fn collect_node_readiness_report() -> DependencyReadinessReport { storage_ready: runtime_sources::object_store_handle().is_some(), iam_ready: runtime_sources::iam_ready(), lock_quorum_ready: collect_lock_quorum_status().await.ready, + peer_health_ready: collect_peer_health_readiness(), }; let report = dependency_readiness_report_from_readiness(readiness); record_readiness_report(&report); @@ -630,6 +677,7 @@ fn cluster_health_timeout_report() -> DependencyReadinessReport { storage_ready: false, iam_ready: false, lock_quorum_ready: false, + peer_health_ready: false, }, degraded_reasons: vec![ReadinessDegradedReason::ClusterHealthTimeout], } @@ -648,6 +696,7 @@ pub async fn collect_cluster_read_dependency_readiness_report() -> DependencyRea storage_ready, iam_ready: iam_ready_raw, lock_quorum_ready: lock_quorum_status.ready, + peer_health_ready: collect_peer_health_readiness(), }; let report = dependency_readiness_report_from_readiness(readiness); record_readiness_report(&report); @@ -659,6 +708,7 @@ pub(crate) async fn snapshot_dependency_readiness_report() -> DependencyReadines storage_ready: collect_storage_readiness_uncached().await, iam_ready: runtime_sources::iam_ready(), lock_quorum_ready: collect_lock_quorum_status_uncached().await.ready, + peer_health_ready: collect_peer_health_readiness(), }; dependency_readiness_report_from_readiness(readiness) @@ -810,18 +860,19 @@ 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 && readiness.lock_quorum_ready && readiness.peer_health_ready { on_ready(readiness); return Ok(()); } if tokio::time::Instant::now() >= startup_deadline { let reason = format!( - "startup readiness timed out after {}s: storage_ready={}, iam_ready={}, lock_quorum_ready={}", + "startup readiness timed out after {}s: storage_ready={}, iam_ready={}, lock_quorum_ready={}, peer_health_ready={}", max_wait.as_secs(), readiness.storage_ready, readiness.iam_ready, - readiness.lock_quorum_ready + readiness.lock_quorum_ready, + readiness.peer_health_ready ); return Err(std::io::Error::other(reason)); } @@ -831,6 +882,7 @@ where storage_ready = readiness.storage_ready, iam_ready = readiness.iam_ready, lock_quorum_ready = readiness.lock_quorum_ready, + peer_health_ready = readiness.peer_health_ready, "Runtime node readiness has not been reached yet; delaying ready state publication" ); tokio::time::sleep(poll_interval).await; @@ -844,7 +896,7 @@ mod tests { use serial_test::serial; use std::future; use std::sync::atomic::{AtomicUsize, Ordering}; - use temp_env::async_with_vars; + use temp_env::{async_with_vars, with_var}; #[test] fn startup_runtime_readiness_wait_constants_are_ordered() { @@ -853,6 +905,46 @@ mod tests { assert_eq!(STARTUP_RUNTIME_READINESS_POLL_INTERVAL.as_secs(), 1); } + fn peer_health_test_pools() -> EndpointServerPools { + EndpointServerPools::from(vec![PoolEndpoints { + legacy: false, + set_count: 1, + drives_per_set: 2, + endpoints: Endpoints::from(vec![ + Endpoint { + url: url::Url::parse("http://node1:9000/data1").expect("test endpoint url"), + is_local: true, + pool_idx: 0, + set_idx: 0, + disk_idx: 0, + }, + Endpoint { + url: url::Url::parse("http://node2:9000/data2").expect("test endpoint url"), + is_local: false, + pool_idx: 0, + set_idx: 0, + disk_idx: 1, + }, + ]), + cmd_line: String::new(), + platform: String::new(), + }]) + } + + #[test] + fn peer_health_gate_defaults_to_ready() { + with_var(rustfs_config::ENV_HEALTH_PEER_READY_CHECK_ENABLE, Some("false"), || { + assert!(peer_health_ready_from_endpoint_pools(&peer_health_test_pools())); + }); + } + + #[test] + fn peer_health_gate_degrades_unknown_peer_health_when_enabled() { + with_var(rustfs_config::ENV_HEALTH_PEER_READY_CHECK_ENABLE, Some("true"), || { + assert!(!peer_health_ready_from_endpoint_pools(&peer_health_test_pools())); + }); + } + #[tokio::test] #[serial] async fn cluster_health_report_singleflight_reuses_inflight_result() { @@ -869,6 +961,7 @@ mod tests { storage_ready: true, iam_ready: true, lock_quorum_ready: true, + peer_health_ready: true, }, degraded_reasons: Vec::new(), }; @@ -924,6 +1017,7 @@ mod tests { storage_ready: true, iam_ready: true, lock_quorum_ready: true, + peer_health_ready: true, }, degraded_reasons: Vec::new(), } @@ -951,6 +1045,7 @@ mod tests { storage_ready: false, iam_ready: false, lock_quorum_ready: false, + peer_health_ready: true, }) }, |_| { @@ -979,6 +1074,7 @@ mod tests { storage_ready: true, iam_ready: true, lock_quorum_ready: false, + peer_health_ready: true, }) }, |_| { @@ -1007,6 +1103,7 @@ mod tests { storage_ready: true, iam_ready: true, lock_quorum_ready: true, + peer_health_ready: true, }) }, |_| { @@ -1276,18 +1373,36 @@ mod tests { #[test] fn degraded_reasons_include_lock_quorum_failures() { - assert_eq!(degraded_reasons(true, true, false), vec![ReadinessDegradedReason::LockQuorumUnavailable]); assert_eq!( - degraded_reasons(false, true, false), + base_degraded_reasons(true, true, false), + vec![ReadinessDegradedReason::LockQuorumUnavailable] + ); + assert_eq!( + base_degraded_reasons(false, true, false), vec![ReadinessDegradedReason::StorageAndLockUnavailable] ); - assert_eq!(degraded_reasons(true, false, false), vec![ReadinessDegradedReason::IamAndLockUnavailable]); assert_eq!( - degraded_reasons(false, false, false), + base_degraded_reasons(true, false, false), + vec![ReadinessDegradedReason::IamAndLockUnavailable] + ); + assert_eq!( + base_degraded_reasons(false, false, false), vec![ReadinessDegradedReason::StorageIamAndLockUnavailable] ); } + #[test] + fn degraded_reasons_append_peer_health_gate_failures() { + let readiness = DependencyReadiness { + storage_ready: true, + iam_ready: true, + lock_quorum_ready: true, + peer_health_ready: false, + }; + + assert_eq!(degraded_reasons(readiness), vec![ReadinessDegradedReason::PeerHealthUnavailable]); + } + #[tokio::test] #[serial] async fn lock_quorum_status_cache_roundtrip() {