diff --git a/crates/config/README.md b/crates/config/README.md index ea2aaa77b..bb921c2a5 100644 --- a/crates/config/README.md +++ b/crates/config/README.md @@ -71,6 +71,24 @@ Current guidance: - `RUSTFS_SCANNER_IDLE_MODE` (canonical) - `RUSTFS_SCANNER_CACHE_SAVE_TIMEOUT_SECS` (canonical) +## Health compatibility switches + +- `RUSTFS_HEALTH_ENDPOINT_ENABLE` + - controls canonical `/health`, `/health/live`, and `/health/ready` endpoint exposure. +- `RUSTFS_HEALTH_MINIMAL_RESPONSE_ENABLE` + - enables minimal payload mode for GET health responses (`status`, `ready` only). +- `RUSTFS_HEALTH_READINESS_CACHE_TTL_MS` + - TTL for readiness cache evaluation. +- `RUSTFS_HEALTH_COMPAT_BUSY_CHECK_ENABLE` + - enables busy protection behavior for health probes. + - default is `false`. +- `RUSTFS_HEALTH_COMPAT_BUSY_MAX_ACTIVE_REQUESTS` + - max active HTTP requests; health probes return `429` when active requests reach or exceed this value. + - `0` disables thresholding even if busy protection is enabled. +- `RUSTFS_HEALTH_COMPAT_KMS_READY_CHECK_ENABLE` + - enables KMS readiness enforcement for `/health/ready`. + - default is `false`. + ## Drive timeout environment variables - `RUSTFS_DRIVE_METADATA_TIMEOUT_SECS` diff --git a/crates/config/src/constants/health.rs b/crates/config/src/constants/health.rs index 86fea1126..04b528dba 100644 --- a/crates/config/src/constants/health.rs +++ b/crates/config/src/constants/health.rs @@ -26,3 +26,20 @@ pub const DEFAULT_HEALTH_READINESS_CACHE_TTL_MS: u64 = 1000; /// When enabled, only `status` and `ready` fields are returned. pub const ENV_HEALTH_MINIMAL_RESPONSE_ENABLE: &str = "RUSTFS_HEALTH_MINIMAL_RESPONSE_ENABLE"; pub const DEFAULT_HEALTH_MINIMAL_RESPONSE_ENABLE: bool = false; + +/// Enable busy protection for health probes. +/// When enabled with a positive request threshold, alias health probes may +/// return 429 when active HTTP requests exceed the threshold. +pub const ENV_HEALTH_COMPAT_BUSY_CHECK_ENABLE: &str = "RUSTFS_HEALTH_COMPAT_BUSY_CHECK_ENABLE"; +pub const DEFAULT_HEALTH_COMPAT_BUSY_CHECK_ENABLE: bool = false; + +/// Max active HTTP requests; alias health probes report busy (429) when active requests reach or exceed this value. +/// Set to 0 to disable thresholding even when busy protection is enabled. +pub const ENV_HEALTH_COMPAT_BUSY_MAX_ACTIVE_REQUESTS: &str = "RUSTFS_HEALTH_COMPAT_BUSY_MAX_ACTIVE_REQUESTS"; +pub const DEFAULT_HEALTH_COMPAT_BUSY_MAX_ACTIVE_REQUESTS: usize = 0; + +/// Enable KMS readiness check for alias readiness probes. +/// When enabled, `/health/ready` additionally requires KMS service to be +/// 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; diff --git a/crates/e2e_test/src/kms/README.md b/crates/e2e_test/src/kms/README.md index e730b74bd..4565999d8 100644 --- a/crates/e2e_test/src/kms/README.md +++ b/crates/e2e_test/src/kms/README.md @@ -191,7 +191,7 @@ RUST_LOG=debug cargo test test_local_kms_end_to_end -- --nocapture 4. **Monitor ports** ```bash netstat -an | grep 9050 - curl http://127.0.0.1:9050/minio/health/ready + curl http://127.0.0.1:9050/health/ready ``` ## 📊 Coverage diff --git a/crates/ecstore/src/rpc/remote_disk.rs b/crates/ecstore/src/rpc/remote_disk.rs index f018df5cb..29f7cad86 100644 --- a/crates/ecstore/src/rpc/remote_disk.rs +++ b/crates/ecstore/src/rpc/remote_disk.rs @@ -450,7 +450,7 @@ impl RemoteDisk { async fn mark_faulty_and_evict(&self, reason: &'static str) { let previous_state = self.runtime_state(); - let became_offline = self.mark_suspect_or_offline(reason); + let transitioned_to_offline = self.mark_suspect_or_offline(reason); let state = self.runtime_state(); if state != previous_state { @@ -461,7 +461,7 @@ impl RemoteDisk { "reason" => reason.to_string() ) .increment(1); - if became_offline || state == RuntimeDriveHealthState::Offline { + if transitioned_to_offline { warn!( "Remote disk marked faulty after timeout: endpoint={}, addr={}, reason={}", self.endpoint, self.addr, reason diff --git a/crates/ecstore/src/rpc/remote_locker.rs b/crates/ecstore/src/rpc/remote_locker.rs index 3d6f4f911..a89854caf 100644 --- a/crates/ecstore/src/rpc/remote_locker.rs +++ b/crates/ecstore/src/rpc/remote_locker.rs @@ -14,12 +14,15 @@ use crate::rpc::client::{TonicInterceptor, gen_tonic_signature_interceptor, node_service_time_out_client}; use async_trait::async_trait; +use bytes::Bytes; use rustfs_lock::{ LockClient, LockError, LockInfo, LockRequest, LockResponse, LockStats, LockStatus, LockType, Result, types::{LockId, LockMetadata, LockPriority}, }; use rustfs_protos::proto_gen::node_service::{BatchGenerallyLockRequest, GenerallyLockRequest, PingRequest}; -use rustfs_protos::{evict_failed_connection, proto_gen::node_service::node_service_client::NodeServiceClient}; +use rustfs_protos::{ + evict_failed_connection, models::PingBodyBuilder, proto_gen::node_service::node_service_client::NodeServiceClient, +}; use std::time::Duration; use tokio::time::timeout; use tonic::Request; @@ -42,6 +45,20 @@ impl RemoteClient { Self { addr: url.to_string() } } + fn build_ping_request() -> PingRequest { + let mut fbb = flatbuffers::FlatBufferBuilder::new(); + let payload = fbb.create_vector(b"health-check"); + let mut builder = PingBodyBuilder::new(&mut fbb); + builder.add_payload(payload); + let root = builder.finish(); + fbb.finish(root, None); + + PingRequest { + version: 1, + body: Bytes::copy_from_slice(fbb.finished_data()), + } + } + /// Create a minimal LockRequest for unlock operations using only lock_id fn create_unlock_request(lock_id: &LockId) -> LockRequest { LockRequest { @@ -510,10 +527,7 @@ impl LockClient for RemoteClient { } }; - let ping_req = Request::new(PingRequest { - version: 1, - body: bytes::Bytes::new(), - }); + let ping_req = Request::new(Self::build_ping_request()); match client.ping(ping_req).await { Ok(_) => { diff --git a/rustfs/src/admin/console.rs b/rustfs/src/admin/console.rs index 0b7529c58..ad70be8f8 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_payload, collect_dependency_readiness, health_check_state}; +use crate::admin::handlers::health::{HealthProbe, build_health_response_parts, collect_dependency_readiness}; use crate::license::has_valid_license; use crate::server::{CONSOLE_PREFIX, FAVICON_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, LICENSE, RUSTFS_ADMIN_PREFIX, VERSION}; use crate::version::build; @@ -465,6 +465,10 @@ fn setup_console_middleware_stack( if rustfs_utils::get_env_bool(rustfs_config::ENV_HEALTH_ENDPOINT_ENABLE, rustfs_config::DEFAULT_HEALTH_ENDPOINT_ENABLE) { app = app .route(&format!("{CONSOLE_PREFIX}{HEALTH_PREFIX}"), get(health_check).head(health_check)) + .route( + &format!("{CONSOLE_PREFIX}{}", crate::server::HEALTH_COMPAT_LIVE_PATH), + get(health_check).head(health_check), + ) .route(&format!("{CONSOLE_PREFIX}{HEALTH_READY_PATH}"), get(health_check).head(health_check)); } else { // Keep disabled health probes from falling through to the SPA fallback. @@ -473,6 +477,10 @@ fn setup_console_middleware_stack( &format!("{CONSOLE_PREFIX}{HEALTH_PREFIX}"), get(health_route_disabled).head(health_route_disabled), ) + .route( + &format!("{CONSOLE_PREFIX}{}", crate::server::HEALTH_COMPAT_LIVE_PATH), + get(health_route_disabled).head(health_route_disabled), + ) .route( &format!("{CONSOLE_PREFIX}{HEALTH_READY_PATH}"), get(health_route_disabled).head(health_route_disabled), @@ -523,31 +531,26 @@ async fn health_check(method: Method, uri: Uri) -> Response { HealthProbe::Liveness }; let readiness_report = collect_dependency_readiness().await; - 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 health = health_check_state(storage_ready, iam_ready, lock_quorum_ready, probe); + let uptime = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .unwrap_or_default() + .as_secs(); + let response_parts = + build_health_response_parts(method.clone(), probe, &readiness_report, "rustfs-console", Some(uptime), None); let builder = Response::builder() - .status(health.status_code) + .status(response_parts.status_code) .header("content-type", "application/json"); match method { // GET: Returns complete JSON Method::GET => { - let uptime = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .unwrap_or_default() - .as_secs(); - let body_json = build_health_payload( - health, - storage_ready, - iam_ready, - lock_quorum_ready, - &readiness_report.degraded_reasons, - "rustfs-console", - Some(uptime), - ); + let body_json = response_parts.payload.unwrap_or_else(|| { + serde_json::json!({ + "status": "error", + "service": "rustfs-console", + }) + }); // Return a minimal JSON when serialization fails to avoid panic let body_str = serde_json::to_string(&body_json).unwrap_or_else(|e| { diff --git a/rustfs/src/admin/handlers/health.rs b/rustfs/src/admin/handlers/health.rs index e76a472e2..79ab62d8f 100644 --- a/rustfs/src/admin/handlers/health.rs +++ b/rustfs/src/admin/handlers/health.rs @@ -57,6 +57,24 @@ pub(crate) enum HealthProbe { Readiness, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub(crate) struct HealthResponseParts { + pub(crate) status_code: StatusCode, + pub(crate) payload: Option, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct HealthPayloadContext<'a> { + pub(crate) health: HealthCheckState, + pub(crate) storage_ready: bool, + pub(crate) iam_ready: bool, + pub(crate) lock_quorum_ready: bool, + pub(crate) degraded_reasons: &'a [crate::server::ReadinessDegradedReason], + pub(crate) service: &'a str, + pub(crate) uptime: Option, + pub(crate) kms_ready: Option, +} + pub(crate) async fn collect_dependency_readiness() -> crate::server::DependencyReadinessReport { collect_runtime_dependency_readiness_report().await } @@ -90,8 +108,13 @@ pub(crate) fn health_minimal_response_enabled() -> bool { ) } -pub(crate) fn build_component_details(storage_ready: bool, iam_ready: bool, lock_quorum_ready: bool) -> Value { - json!({ +pub(crate) fn build_component_details( + storage_ready: bool, + iam_ready: bool, + lock_quorum_ready: bool, + kms_ready: Option, +) -> Value { + let mut details = json!({ "storage": { "status": if storage_ready { "connected" } else { "disconnected" }, "ready": storage_ready, @@ -104,7 +127,16 @@ pub(crate) fn build_component_details(storage_ready: bool, iam_ready: bool, lock "status": if lock_quorum_ready { "connected" } else { "disconnected" }, "ready": lock_quorum_ready, } - }) + }); + + if let Some(kms_ready) = kms_ready { + details["kms"] = json!({ + "status": if kms_ready { "connected" } else { "disconnected" }, + "ready": kms_ready, + }); + } + + details } pub(crate) fn build_degraded_reasons(reasons: &[crate::server::ReadinessDegradedReason]) -> Value { @@ -124,71 +156,77 @@ pub(crate) fn probe_from_path(path: &str) -> HealthProbe { } } -pub(crate) fn build_health_payload( - health: HealthCheckState, - storage_ready: bool, - iam_ready: bool, - lock_quorum_ready: bool, - degraded_reasons: &[crate::server::ReadinessDegradedReason], +pub(crate) fn build_health_response_parts( + method: Method, + probe: HealthProbe, + readiness_report: &crate::server::DependencyReadinessReport, service: &str, uptime: Option, -) -> Value { + kms_ready: Option, +) -> HealthResponseParts { + 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 mut health = health_check_state(storage_ready, iam_ready, lock_quorum_ready, probe); + let mut degraded_reasons = readiness_report.degraded_reasons.clone(); + + if probe == HealthProbe::Readiness && matches!(kms_ready, Some(false)) { + health = HealthCheckState { + status_code: StatusCode::SERVICE_UNAVAILABLE, + status: "degraded", + ready: false, + }; + if !degraded_reasons.contains(&crate::server::ReadinessDegradedReason::KmsNotReady) { + degraded_reasons.push(crate::server::ReadinessDegradedReason::KmsNotReady); + } + } + + let payload = if method == Method::HEAD { + None + } else { + Some(build_health_payload(HealthPayloadContext { + health, + storage_ready, + iam_ready, + lock_quorum_ready, + degraded_reasons: °raded_reasons, + service, + uptime, + kms_ready, + })) + }; + + HealthResponseParts { + status_code: health.status_code, + payload, + } +} + +pub(crate) fn build_health_payload(ctx: HealthPayloadContext<'_>) -> Value { if health_minimal_response_enabled() { return json!({ - "status": health.status, - "ready": health.ready, + "status": ctx.health.status, + "ready": ctx.health.ready, }); } let mut payload = json!({ - "status": health.status, - "ready": health.ready, - "service": service, + "status": ctx.health.status, + "ready": ctx.health.ready, + "service": ctx.service, "timestamp": jiff::Zoned::now().to_string(), "version": env!("CARGO_PKG_VERSION"), - "details": build_component_details(storage_ready, iam_ready, lock_quorum_ready), - "degradedReasons": build_degraded_reasons(degraded_reasons), + "details": build_component_details(ctx.storage_ready, ctx.iam_ready, ctx.lock_quorum_ready, ctx.kms_ready), + "degradedReasons": build_degraded_reasons(ctx.degraded_reasons), }); - if let Some(uptime) = uptime { + if let Some(uptime) = ctx.uptime { payload["uptime"] = json!(uptime); } payload } -pub(crate) fn build_health_response( - method: Method, - probe: HealthProbe, - storage_ready: bool, - iam_ready: bool, - lock_quorum_ready: bool, - degraded_reasons: &[crate::server::ReadinessDegradedReason], -) -> S3Response<(StatusCode, Body)> { - let health = health_check_state(storage_ready, iam_ready, lock_quorum_ready, probe); - let health_info = build_health_payload( - health, - storage_ready, - iam_ready, - lock_quorum_ready, - degraded_reasons, - "rustfs-endpoint", - None, - ); - - let mut headers = HeaderMap::new(); - headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); - - if method == Method::HEAD { - return S3Response::with_headers((health.status_code, Body::empty()), headers); - } - - let body_str = serde_json::to_string(&health_info).unwrap_or_else(|_| "{}".to_string()); - let body = Body::from(body_str); - - S3Response::with_headers((health.status_code, body), headers) -} - #[async_trait::async_trait] impl Operation for HealthCheckHandler { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { @@ -209,14 +247,19 @@ impl Operation for HealthCheckHandler { let probe = probe_from_path(req.uri.path()); let readiness_report = collect_dependency_readiness().await; - Ok(build_health_response( - method, - probe, - readiness_report.readiness.storage_ready, - readiness_report.readiness.iam_ready, - readiness_report.readiness.lock_quorum_ready, - &readiness_report.degraded_reasons, - )) + let response_parts = build_health_response_parts(method.clone(), probe, &readiness_report, "rustfs-endpoint", None, None); + + let mut headers = HeaderMap::new(); + headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + + let response = if let Some(payload) = response_parts.payload { + let body_str = serde_json::to_string(&payload).unwrap_or_else(|_| "{}".to_string()); + S3Response::with_headers((response_parts.status_code, Body::from(body_str)), headers) + } else { + S3Response::with_headers((response_parts.status_code, Body::empty()), headers) + }; + + Ok(response) } } @@ -267,7 +310,7 @@ mod tests { #[test] fn test_health_check_component_details() { - let details = build_component_details(true, false, false); + let details = build_component_details(true, false, false, None); assert_eq!(details["storage"]["status"], "connected"); assert_eq!(details["storage"]["ready"], true); @@ -275,6 +318,14 @@ mod tests { assert_eq!(details["iam"]["ready"], false); assert_eq!(details["lock"]["status"], "disconnected"); assert_eq!(details["lock"]["ready"], false); + assert!(details.get("kms").is_none()); + } + + #[test] + fn test_health_check_component_details_include_kms_when_present() { + let details = build_component_details(true, true, true, Some(false)); + assert_eq!(details["kms"]["status"], "disconnected"); + assert_eq!(details["kms"]["ready"], false); } #[test] @@ -290,62 +341,79 @@ mod tests { #[test] fn test_build_health_response_readiness_returns_503_when_deps_not_ready() { - let resp = build_health_response( - Method::GET, - HealthProbe::Readiness, - false, - true, - true, - &[crate::server::ReadinessDegradedReason::StorageQuorumUnavailable], - ); - assert_eq!(resp.output.0, StatusCode::SERVICE_UNAVAILABLE); + let readiness_report = crate::server::DependencyReadinessReport { + readiness: crate::server::DependencyReadiness { + storage_ready: false, + iam_ready: true, + lock_quorum_ready: true, + }, + degraded_reasons: vec![crate::server::ReadinessDegradedReason::StorageQuorumUnavailable], + }; + let parts = + build_health_response_parts(Method::GET, HealthProbe::Readiness, &readiness_report, "rustfs-endpoint", None, None); + assert_eq!(parts.status_code, StatusCode::SERVICE_UNAVAILABLE); } #[test] fn test_build_health_response_readiness_returns_200_when_deps_ready() { - let resp = build_health_response(Method::GET, HealthProbe::Readiness, true, true, true, &[]); - assert_eq!(resp.output.0, StatusCode::OK); + let readiness_report = crate::server::DependencyReadinessReport { + readiness: crate::server::DependencyReadiness { + storage_ready: true, + iam_ready: true, + lock_quorum_ready: true, + }, + degraded_reasons: Vec::new(), + }; + let parts = + build_health_response_parts(Method::GET, HealthProbe::Readiness, &readiness_report, "rustfs-endpoint", None, None); + assert_eq!(parts.status_code, StatusCode::OK); } #[test] fn test_build_health_response_liveness_returns_200_when_deps_not_ready() { - let resp = build_health_response( - Method::GET, - HealthProbe::Liveness, - false, - false, - false, - &[crate::server::ReadinessDegradedReason::StorageAndIamUnavailable], - ); - assert_eq!(resp.output.0, StatusCode::OK); + let readiness_report = crate::server::DependencyReadinessReport { + readiness: crate::server::DependencyReadiness { + storage_ready: false, + iam_ready: false, + lock_quorum_ready: false, + }, + degraded_reasons: vec![crate::server::ReadinessDegradedReason::StorageAndIamUnavailable], + }; + let parts = + build_health_response_parts(Method::GET, HealthProbe::Liveness, &readiness_report, "rustfs-endpoint", None, None); + assert_eq!(parts.status_code, StatusCode::OK); } #[test] fn test_build_health_response_head_returns_empty_body() { - let resp = build_health_response( - Method::HEAD, - HealthProbe::Readiness, - false, - false, - false, - &[crate::server::ReadinessDegradedReason::StorageAndIamUnavailable], - ); - assert_eq!(resp.output.0, StatusCode::SERVICE_UNAVAILABLE); + let readiness_report = crate::server::DependencyReadinessReport { + readiness: crate::server::DependencyReadiness { + storage_ready: false, + iam_ready: false, + lock_quorum_ready: false, + }, + degraded_reasons: vec![crate::server::ReadinessDegradedReason::StorageAndIamUnavailable], + }; + let parts = + build_health_response_parts(Method::HEAD, HealthProbe::Readiness, &readiness_report, "rustfs-endpoint", None, None); + assert_eq!(parts.status_code, StatusCode::SERVICE_UNAVAILABLE); + assert!(parts.payload.is_none()); } #[test] fn test_build_health_payload_minimal_mode_returns_status_and_ready_only() { let health = health_check_state(true, false, true, HealthProbe::Readiness); with_var(rustfs_config::ENV_HEALTH_MINIMAL_RESPONSE_ENABLE, Some("true"), || { - let payload = build_health_payload( + let payload = build_health_payload(HealthPayloadContext { health, - true, - false, - true, - &[crate::server::ReadinessDegradedReason::IamNotReady], - "rustfs-endpoint", - Some(123), - ); + storage_ready: true, + iam_ready: false, + lock_quorum_ready: true, + degraded_reasons: &[crate::server::ReadinessDegradedReason::IamNotReady], + service: "rustfs-endpoint", + uptime: Some(123), + kms_ready: None, + }); assert_eq!(payload["status"], "degraded"); assert_eq!(payload["ready"], false); assert!(payload.get("version").is_none()); @@ -358,15 +426,69 @@ mod tests { #[test] fn test_build_health_payload_includes_degraded_reasons() { let health = health_check_state(false, false, false, HealthProbe::Readiness); - let payload = build_health_payload( + let payload = build_health_payload(HealthPayloadContext { health, - false, - false, - false, - &[crate::server::ReadinessDegradedReason::StorageAndIamUnavailable], - "rustfs-endpoint", - None, - ); + storage_ready: false, + iam_ready: false, + lock_quorum_ready: false, + degraded_reasons: &[crate::server::ReadinessDegradedReason::StorageAndIamUnavailable], + service: "rustfs-endpoint", + uptime: None, + kms_ready: None, + }); assert_eq!(payload["degradedReasons"][0], "storage_and_iam_unavailable"); } + + #[test] + fn test_build_health_response_parts_head_has_no_payload() { + let report = crate::server::DependencyReadinessReport { + readiness: crate::server::DependencyReadiness { + storage_ready: true, + iam_ready: true, + lock_quorum_ready: true, + }, + degraded_reasons: Vec::new(), + }; + let parts = build_health_response_parts(Method::HEAD, HealthProbe::Readiness, &report, "rustfs-endpoint", None, None); + assert_eq!(parts.status_code, StatusCode::OK); + assert!(parts.payload.is_none()); + } + + #[test] + fn test_build_health_response_parts_get_includes_payload() { + let report = crate::server::DependencyReadinessReport { + readiness: crate::server::DependencyReadiness { + storage_ready: false, + iam_ready: true, + lock_quorum_ready: true, + }, + degraded_reasons: vec![crate::server::ReadinessDegradedReason::StorageQuorumUnavailable], + }; + let parts = build_health_response_parts(Method::GET, HealthProbe::Readiness, &report, "rustfs-endpoint", None, None); + assert_eq!(parts.status_code, StatusCode::SERVICE_UNAVAILABLE); + let payload = parts.payload.expect("GET should include payload"); + assert_eq!(payload["status"], "degraded"); + assert_eq!(payload["ready"], false); + assert_eq!(payload["degradedReasons"][0], "storage_quorum_unavailable"); + } + + #[test] + fn test_build_health_response_parts_readiness_marks_kms_not_ready() { + let report = crate::server::DependencyReadinessReport { + readiness: crate::server::DependencyReadiness { + storage_ready: true, + iam_ready: true, + lock_quorum_ready: true, + }, + degraded_reasons: Vec::new(), + }; + let parts = + build_health_response_parts(Method::GET, HealthProbe::Readiness, &report, "rustfs-endpoint", None, Some(false)); + assert_eq!(parts.status_code, StatusCode::SERVICE_UNAVAILABLE); + let payload = parts.payload.expect("GET should include payload"); + assert_eq!(payload["ready"], false); + assert_eq!(payload["details"]["lock"]["ready"], true); + assert_eq!(payload["details"]["kms"]["ready"], false); + assert_eq!(payload["degradedReasons"][0], "kms_not_ready"); + } } diff --git a/rustfs/src/server/compress.rs b/rustfs/src/server/compress.rs index 2ba127830..64ef6dae3 100644 --- a/rustfs/src/server/compress.rs +++ b/rustfs/src/server/compress.rs @@ -380,7 +380,7 @@ impl PathCategory { PathCategory::AdminApi } else if path.starts_with("/rustfs/console") { PathCategory::Console - } else if path.starts_with("/minio/health/") { + } else if path == "/health" || path.starts_with("/health/") { PathCategory::Probe } else { PathCategory::S3DataPlane @@ -725,8 +725,9 @@ mod tests { #[test] fn test_path_category_classify_probe() { - assert_eq!(PathCategory::classify("/minio/health/live"), PathCategory::Probe); - assert_eq!(PathCategory::classify("/minio/health/ready"), PathCategory::Probe); + assert_eq!(PathCategory::classify("/health"), PathCategory::Probe); + assert_eq!(PathCategory::classify("/health/live"), PathCategory::Probe); + assert_eq!(PathCategory::classify("/health/ready"), PathCategory::Probe); } #[test] diff --git a/rustfs/src/server/layer.rs b/rustfs/src/server/layer.rs index 12a8cb451..0c4cd1662 100644 --- a/rustfs/src/server/layer.rs +++ b/rustfs/src/server/layer.rs @@ -13,13 +13,13 @@ // limitations under the License. use crate::admin::console::is_console_path; -use crate::admin::handlers::health::{build_health_payload, health_check_state, probe_from_path}; +use crate::admin::handlers::health::{HealthProbe, build_health_response_parts}; use crate::error::ApiError; use crate::server::cors; use crate::server::hybrid::HybridBody; use crate::server::{ - ADMIN_PREFIX, CONSOLE_PREFIX, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_ADMIN_PREFIX, MINIO_ADMIN_V3_PREFIX, RPC_PREFIX, - RUSTFS_ADMIN_PREFIX, collect_dependency_readiness_report, + ADMIN_PREFIX, CONSOLE_PREFIX, HEALTH_COMPAT_LIVE_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_ADMIN_PREFIX, + MINIO_ADMIN_V3_PREFIX, RPC_PREFIX, RUSTFS_ADMIN_PREFIX, active_http_requests, collect_dependency_readiness_report, }; use crate::storage::apply_cors_headers; use crate::storage::request_context::{RequestContext, extract_request_id_from_headers}; @@ -625,10 +625,58 @@ fn health_endpoint_enabled() -> bool { rustfs_utils::get_env_bool(rustfs_config::ENV_HEALTH_ENDPOINT_ENABLE, rustfs_config::DEFAULT_HEALTH_ENDPOINT_ENABLE) } +fn health_compat_busy_check_enabled() -> bool { + rustfs_utils::get_env_bool( + rustfs_config::ENV_HEALTH_COMPAT_BUSY_CHECK_ENABLE, + rustfs_config::DEFAULT_HEALTH_COMPAT_BUSY_CHECK_ENABLE, + ) +} + +fn health_compat_busy_max_active_requests() -> u64 { + rustfs_utils::get_env_usize( + rustfs_config::ENV_HEALTH_COMPAT_BUSY_MAX_ACTIVE_REQUESTS, + rustfs_config::DEFAULT_HEALTH_COMPAT_BUSY_MAX_ACTIVE_REQUESTS, + ) as u64 +} + +fn health_compat_kms_ready_check_enabled() -> bool { + rustfs_utils::get_env_bool( + rustfs_config::ENV_HEALTH_COMPAT_KMS_READY_CHECK_ENABLE, + rustfs_config::DEFAULT_HEALTH_COMPAT_KMS_READY_CHECK_ENABLE, + ) +} + +fn resolve_public_health_probe(method: &Method, path: &str) -> Option { + if (method != Method::GET && method != Method::HEAD) || !health_endpoint_enabled() { + return None; + } + + match path { + HEALTH_PREFIX | HEALTH_COMPAT_LIVE_PATH => Some(HealthProbe::Liveness), + HEALTH_READY_PATH => Some(HealthProbe::Readiness), + _ => None, + } +} + +fn alias_busy_threshold_exceeded(active_requests: u64) -> bool { + if !health_compat_busy_check_enabled() { + return false; + } + + let max_active_requests = health_compat_busy_max_active_requests(); + max_active_requests > 0 && active_requests >= max_active_requests +} + fn is_public_health_endpoint_request(method: &Method, path: &str) -> bool { - (method == Method::GET || method == Method::HEAD) - && (path == HEALTH_PREFIX || path == HEALTH_READY_PATH) - && health_endpoint_enabled() + resolve_public_health_probe(method, path).is_some() +} + +async fn health_kms_ready() -> bool { + let Some(service_manager) = rustfs_kms::get_global_kms_service_manager() else { + return true; + }; + + matches!(service_manager.get_status().await, rustfs_kms::KmsServiceStatus::Running) } async fn build_public_health_http_response( @@ -638,29 +686,41 @@ async fn build_public_health_http_response( where RestBody: From, { - let probe = probe_from_path(&path); + let probe = resolve_public_health_probe(&method, path.as_str()) + .expect("public health endpoint request should always resolve health probe"); + + if probe == HealthProbe::Readiness && alias_busy_threshold_exceeded(active_http_requests()) { + let retry_after = HeaderValue::from_static("5"); + let body_bytes = Bytes::from_static(b"{\"status\":\"busy\",\"ready\":false}"); + let mut builder = Response::builder() + .status(StatusCode::TOO_MANY_REQUESTS) + .header(http::header::CONTENT_TYPE, "application/json") + .header(http::header::RETRY_AFTER, retry_after); + if let Ok(val) = HeaderValue::from_str(&body_bytes.len().to_string()) { + builder = builder.header(http::header::CONTENT_LENGTH, val); + } + return builder + .body(HybridBody::Rest { + rest_body: RestBody::from(body_bytes), + }) + .expect("failed to build health busy response"); + } + let readiness_report = collect_dependency_readiness_report().await; - 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 health = health_check_state(storage_ready, iam_ready, lock_quorum_ready, probe); - let body = if method == Method::HEAD { - Bytes::new() + let kms_ready = if probe == HealthProbe::Readiness && health_compat_kms_ready_check_enabled() { + Some(health_kms_ready().await) } else { - let payload = build_health_payload( - health, - storage_ready, - iam_ready, - lock_quorum_ready, - &readiness_report.degraded_reasons, - "rustfs-endpoint", - None, - ); - Bytes::from(serde_json::to_vec(&payload).unwrap_or_else(|_| b"{}".to_vec())) + None }; + let response_parts = build_health_response_parts(method, probe, &readiness_report, "rustfs-endpoint", None, kms_ready); + let body = response_parts + .payload + .map(|payload| Bytes::from(serde_json::to_vec(&payload).unwrap_or_else(|_| b"{}".to_vec()))) + .unwrap_or_default(); + Response::builder() - .status(health.status_code) + .status(response_parts.status_code) .header(http::header::CONTENT_TYPE, "application/json") .body(HybridBody::Rest { rest_body: RestBody::from(body), @@ -829,7 +889,8 @@ impl ConditionalCorsLayer { } /// Exact paths that should be excluded from being treated as S3 paths. - const EXCLUDED_EXACT_PATHS: &'static [&'static str] = &["/health", "/health/ready", "/profile/cpu", "/profile/memory"]; + const EXCLUDED_EXACT_PATHS: &'static [&'static str] = + &["/health", "/health/live", "/health/ready", "/profile/cpu", "/profile/memory"]; fn is_s3_path(path: &str) -> bool { // Exclude Admin, Console, RPC, and configured special paths @@ -1218,6 +1279,73 @@ mod tests { .await; } + #[tokio::test] + #[serial] + async fn public_health_endpoint_layer_handles_health_live_path() { + 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(HEALTH_COMPAT_LIVE_PATH) + .body(Full::::from(Bytes::new())) + .expect("request"), + ) + .await + .expect("health response"); + + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(calls.load(Ordering::SeqCst), 0); + }) + .await; + } + + #[tokio::test] + #[serial] + async fn public_health_endpoint_layer_forwards_unknown_health_path_when_endpoint_disabled() { + async_with_vars([(rustfs_config::ENV_HEALTH_ENDPOINT_ENABLE, Some("false"))], 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("/health/live") + .body(Full::::from(Bytes::new())) + .expect("request"), + ) + .await + .expect("inner response"); + + assert_eq!(response.status(), StatusCode::IM_A_TEAPOT); + assert_eq!(calls.load(Ordering::SeqCst), 1); + }) + .await; + } + + #[test] + fn alias_busy_threshold_exceeded_requires_switch_and_positive_threshold() { + with_var(rustfs_config::ENV_HEALTH_COMPAT_BUSY_CHECK_ENABLE, Some("true"), || { + with_var(rustfs_config::ENV_HEALTH_COMPAT_BUSY_MAX_ACTIVE_REQUESTS, Some("2"), || { + assert!(!alias_busy_threshold_exceeded(1)); + assert!(alias_busy_threshold_exceeded(2)); + assert!(alias_busy_threshold_exceeded(3)); + }); + }); + + with_var(rustfs_config::ENV_HEALTH_COMPAT_BUSY_CHECK_ENABLE, Some("false"), || { + with_var(rustfs_config::ENV_HEALTH_COMPAT_BUSY_MAX_ACTIVE_REQUESTS, Some("1"), || { + assert!(!alias_busy_threshold_exceeded(100)); + }); + }); + } + #[tokio::test] #[serial] async fn public_health_endpoint_layer_forwards_health_when_endpoint_disabled() { diff --git a/rustfs/src/server/mod.rs b/rustfs/src/server/mod.rs index 66134e908..386d48065 100644 --- a/rustfs/src/server/mod.rs +++ b/rustfs/src/server/mod.rs @@ -47,8 +47,9 @@ pub(crate) use module_switch::{ refresh_persisted_module_switches_from_store, save_persisted_module_switches_to_store, validate_module_switch_update, }; pub(crate) use prefix::{ - ADMIN_PREFIX, CONSOLE_PREFIX, FAVICON_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, LICENSE, MINIO_ADMIN_PREFIX, - MINIO_ADMIN_V3_PREFIX, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, RPC_PREFIX, RUSTFS_ADMIN_PREFIX, TONIC_PREFIX, VERSION, + ADMIN_PREFIX, CONSOLE_PREFIX, FAVICON_PATH, HEALTH_COMPAT_LIVE_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, LICENSE, + MINIO_ADMIN_PREFIX, MINIO_ADMIN_V3_PREFIX, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, RPC_PREFIX, RUSTFS_ADMIN_PREFIX, + TONIC_PREFIX, VERSION, }; pub(crate) use readiness::DependencyReadiness; pub(crate) use readiness::DependencyReadinessReport; diff --git a/rustfs/src/server/prefix.rs b/rustfs/src/server/prefix.rs index d9a3bd455..be0c64a56 100644 --- a/rustfs/src/server/prefix.rs +++ b/rustfs/src/server/prefix.rs @@ -31,6 +31,9 @@ pub(crate) const HEALTH_PREFIX: &str = "/health"; /// This path is used to check dependency readiness and may return 503. pub(crate) const HEALTH_READY_PATH: &str = "/health/ready"; +/// Health liveness probe compatibility alias path. +pub(crate) const HEALTH_COMPAT_LIVE_PATH: &str = "/health/live"; + /// 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 a7880b515..663e21127 100644 --- a/rustfs/src/server/readiness.rs +++ b/rustfs/src/server/readiness.rs @@ -57,6 +57,7 @@ pub enum ReadinessDegradedReason { StorageQuorumUnavailable, IamNotReady, LockQuorumUnavailable, + KmsNotReady, StorageAndIamUnavailable, StorageAndLockUnavailable, IamAndLockUnavailable, @@ -69,6 +70,7 @@ impl ReadinessDegradedReason { ReadinessDegradedReason::StorageQuorumUnavailable => "storage_quorum_unavailable", ReadinessDegradedReason::IamNotReady => "iam_not_ready", ReadinessDegradedReason::LockQuorumUnavailable => "lock_quorum_unavailable", + ReadinessDegradedReason::KmsNotReady => "kms_not_ready", ReadinessDegradedReason::StorageAndIamUnavailable => "storage_and_iam_unavailable", ReadinessDegradedReason::StorageAndLockUnavailable => "storage_and_lock_unavailable", ReadinessDegradedReason::IamAndLockUnavailable => "iam_and_lock_unavailable", @@ -154,6 +156,7 @@ where crate::server::PROFILE_MEMORY_PATH | crate::server::PROFILE_CPU_PATH | crate::server::HEALTH_PREFIX + | crate::server::HEALTH_COMPAT_LIVE_PATH | crate::server::HEALTH_READY_PATH | crate::server::FAVICON_PATH ); @@ -222,6 +225,12 @@ struct StorageReadinessCacheEntry { storage_ready: bool, } +#[derive(Debug, Clone, Copy)] +struct LockQuorumCacheEntry { + captured_at: Instant, + status: LockQuorumStatus, +} + #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct LockQuorumStatus { pub ready: bool, @@ -246,6 +255,11 @@ fn storage_readiness_cache() -> &'static Mutex &'static Mutex> { + static CACHE: OnceLock>> = OnceLock::new(); + CACHE.get_or_init(|| Mutex::new(None)) +} + async fn load_cached_storage_readiness() -> Option { let ttl = health_readiness_cache_ttl(); if ttl.is_zero() { @@ -273,6 +287,33 @@ async fn update_storage_readiness_cache(storage_ready: bool) { }); } +async fn load_cached_lock_quorum_status() -> Option { + let ttl = health_readiness_cache_ttl(); + if ttl.is_zero() { + return None; + } + + let cache = lock_quorum_status_cache().lock().await; + let entry = cache.as_ref()?; + if entry.captured_at.elapsed() <= ttl { + return Some(entry.status); + } + + None +} + +async fn update_lock_quorum_status_cache(status: LockQuorumStatus) { + if health_readiness_cache_ttl().is_zero() { + return; + } + + let mut cache = lock_quorum_status_cache().lock().await; + *cache = Some(LockQuorumCacheEntry { + captured_at: Instant::now(), + status, + }); +} + fn disk_is_online_for_readiness(disk: &Disk) -> bool { let state_is_acceptable = disk.state.eq_ignore_ascii_case(DISK_STATE_OK) || disk.state.eq_ignore_ascii_case(rustfs_madmin::ITEM_ONLINE) @@ -399,7 +440,7 @@ pub async fn collect_dependency_readiness_report() -> DependencyReadinessReport update_storage_readiness_cache(computed).await; computed }; - let lock_quorum_status = collect_lock_quorum_status_uncached().await; + let lock_quorum_status = collect_lock_quorum_status().await; let readiness = DependencyReadiness { storage_ready, @@ -414,6 +455,16 @@ pub async fn collect_dependency_readiness_report() -> DependencyReadinessReport report } +async fn collect_lock_quorum_status() -> LockQuorumStatus { + if let Some(cached) = load_cached_lock_quorum_status().await { + cached + } else { + let computed = collect_lock_quorum_status_uncached().await; + update_lock_quorum_status_cache(computed).await; + computed + } +} + async fn collect_dependency_readiness_uncached() -> DependencyReadiness { let iam_ready_raw = get_global_iam_sys().is_some_and(|sys| sys.is_ready()); let storage_ready = collect_storage_readiness_uncached().await; @@ -894,4 +945,32 @@ mod tests { vec![ReadinessDegradedReason::StorageIamAndLockUnavailable] ); } + + #[tokio::test] + async fn lock_quorum_status_cache_roundtrip() { + let cache = lock_quorum_status_cache(); + { + let mut guard = cache.lock().await; + *guard = None; + } + + update_lock_quorum_status_cache(LockQuorumStatus { + ready: true, + connected_clients: 2, + total_clients: 3, + required_quorum: 2, + }) + .await; + + let cached = load_cached_lock_quorum_status().await; + assert_eq!( + cached, + Some(LockQuorumStatus { + ready: true, + connected_clients: 2, + total_clients: 3, + required_quorum: 2, + }) + ); + } } diff --git a/scripts/run_four_node_cluster_failover_bench.sh b/scripts/run_four_node_cluster_failover_bench.sh index a87c7a45e..4b5303249 100755 --- a/scripts/run_four_node_cluster_failover_bench.sh +++ b/scripts/run_four_node_cluster_failover_bench.sh @@ -30,6 +30,7 @@ FAILOVER_INTERVAL_SECS="${FAILOVER_INTERVAL_SECS:-1}" BENCH_WAIT_MODE="${BENCH_WAIT_MODE:-ready}" BENCH_ENDPOINT="${BENCH_ENDPOINT:-http://127.0.0.1:9000}" +BENCH_WARP_HOSTS="${BENCH_WARP_HOSTS:-http://127.0.0.1:9000,http://127.0.0.1:9001,http://127.0.0.1:9002,http://127.0.0.1:9003}" BENCH_BUCKET="${BENCH_BUCKET:-rustfs-four-node-bench}" BENCH_AUTO_NEW_BUCKET="${BENCH_AUTO_NEW_BUCKET:-true}" BENCH_BUCKET_PREFIX="${BENCH_BUCKET_PREFIX:-rustfs-four-node-bench}" @@ -58,6 +59,7 @@ Options: --failover-node node to stop during failover test (default: node4) --obs-endpoint RUSTFS_OBS_ENDPOINT (default: auto-select by mode) --bench-endpoint benchmark endpoint (default: http://127.0.0.1:9000) + --bench-warp-hosts comma-separated warp hosts (default: node1..node4) --bench-sizes comma list (default: 1KiB,4KiB,11Mi) --bench-concurrency benchmark concurrency --bench-concurrencies benchmark concurrency list (default: 8,16,32,64,128) @@ -75,6 +77,7 @@ Environment: WAIT_PROBE_MODE (service|ready, default: service) WAIT_TIMEOUT_SECS FAILOVER_NODE FAILOVER_WARMUP_SECS FAILOVER_SAMPLE_SECS FAILOVER_INTERVAL_SECS BENCH_ENDPOINT BENCH_BUCKET BENCH_CONCURRENCY + BENCH_WARP_HOSTS (default: http://127.0.0.1:9000,http://127.0.0.1:9001,http://127.0.0.1:9002,http://127.0.0.1:9003) BENCH_CONCURRENCIES BENCH_DURATION BENCH_SIZES OUT_DIR BENCH_WAIT_MODE (ready|service, default: ready) BENCH_READY_TIMEOUT_SECS (default: 180) @@ -585,7 +588,7 @@ run_benchmark() { cd "${PROJECT_ROOT}" ./scripts/run_object_batch_bench.sh \ --tool warp \ - --endpoint "${BENCH_ENDPOINT}" \ + --endpoint "${BENCH_WARP_HOSTS}" \ --access-key "${RUSTFS_ACCESS_KEY}" \ --secret-key "${RUSTFS_SECRET_KEY}" \ --bucket "${bench_bucket}" \ @@ -664,6 +667,10 @@ parse_args() { BENCH_ENDPOINT="$2" shift 2 ;; + --bench-warp-hosts) + BENCH_WARP_HOSTS="$2" + shift 2 + ;; --bench-sizes) BENCH_SIZES="$2" shift 2 diff --git a/scripts/run_object_batch_bench.sh b/scripts/run_object_batch_bench.sh index 76b9cc70a..521b96469 100755 --- a/scripts/run_object_batch_bench.sh +++ b/scripts/run_object_batch_bench.sh @@ -108,12 +108,28 @@ print_dry_run_command() { normalize_warp_host() { local raw="$1" - raw="${raw#http://}" - raw="${raw#https://}" - raw="${raw%%/*}" - raw="${raw%%\?*}" - raw="${raw%%\#*}" - echo "$raw" + local item normalized + local -a parts + local -a hosts=() + + IFS=',' read -r -a parts <<< "$raw" + for item in "${parts[@]}"; do + item="$(echo "$item" | xargs)" + [[ -z "$item" ]] && continue + item="${item#http://}" + item="${item#https://}" + item="${item%%/*}" + item="${item%%\?*}" + item="${item%%\#*}" + [[ -n "$item" ]] && hosts+=("$item") + done + + if [[ ${#hosts[@]} -eq 0 ]]; then + return 0 + fi + + normalized="$(IFS=','; echo "${hosts[*]}")" + echo "$normalized" } parse_args() {