From 0ffd561daf9470db0690d057a8e8e3eac444ad5b Mon Sep 17 00:00:00 2001 From: cxymds Date: Wed, 24 Jun 2026 18:34:51 +0800 Subject: [PATCH] fix(health): decouple liveness and lock RPC deadlines (#3824) --- crates/ecstore/src/rpc/remote_locker.rs | 49 +++----- rustfs/src/admin/console.rs | 11 +- rustfs/src/admin/handlers/health.rs | 143 +++++++++++++++++------- rustfs/src/server/layer.rs | 18 ++- rustfs/src/server/mod.rs | 1 - rustfs/src/server/readiness.rs | 11 -- 6 files changed, 140 insertions(+), 93 deletions(-) diff --git a/crates/ecstore/src/rpc/remote_locker.rs b/crates/ecstore/src/rpc/remote_locker.rs index dec62b75c..dda07e623 100644 --- a/crates/ecstore/src/rpc/remote_locker.rs +++ b/crates/ecstore/src/rpc/remote_locker.rs @@ -38,8 +38,6 @@ pub struct RemoteClient { } impl RemoteClient { - const RPC_TIMEOUT_GRACE: Duration = Duration::from_millis(500); - pub fn new(endpoint: String) -> Self { Self { addr: endpoint } } @@ -124,28 +122,21 @@ impl RemoteClient { resources.join(", ") } - fn rpc_timeout(lock_wait_budget: Duration) -> Duration { - let configured = Duration::from_millis( + fn rpc_timeout() -> Duration { + Duration::from_millis( rustfs_utils::get_env_u64( rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS, rustfs_config::DEFAULT_OBJECT_LOCK_RPC_TIMEOUT_MS, ) .max(1), - ); - configured.max(lock_wait_budget.saturating_add(Self::RPC_TIMEOUT_GRACE)) + ) } - async fn execute_rpc( - &self, - op: &'static str, - timeout_duration: Duration, - resource_summary: &str, - future: F, - ) -> std::result::Result + async fn execute_rpc(&self, op: &'static str, resource_summary: &str, future: F) -> std::result::Result where F: std::future::Future>, { - let lock_timeout = Self::rpc_timeout(timeout_duration); + let lock_timeout = Self::rpc_timeout(); match timeout(lock_timeout, future).await { Ok(Ok(response)) => Ok(response), Ok(Err(err)) => { @@ -221,14 +212,6 @@ impl RemoteClient { .collect() } - fn batch_rpc_timeout(requests: &[LockRequest]) -> Duration { - requests - .iter() - .map(|request| request.acquire_timeout) - .max() - .unwrap_or_else(|| Duration::from_millis(1)) - } - fn build_lock_info(request: &LockRequest, lock_info_json: Option) -> LockInfo { if let Some(lock_info_json) = lock_info_json { match serde_json::from_str::(&lock_info_json) { @@ -279,10 +262,7 @@ impl LockClient for RemoteClient { .map_err(|e| LockError::internal(format!("Failed to serialize request: {e}")))?, }); - let resp = match self - .execute_rpc("lock", request.acquire_timeout, &resource_summary, client.lock(req)) - .await - { + let resp = match self.execute_rpc("lock", &resource_summary, client.lock(req)).await { Ok(resp) => resp.into_inner(), Err(err @ LockError::Timeout { .. }) => return Ok(Self::rpc_timeout_failure_response(request, &err)), Err(err) => return Ok(Self::rpc_failure_response(request, &err)), @@ -321,7 +301,7 @@ impl LockClient for RemoteClient { }); let resp = match self - .execute_rpc("lock_batch", Self::batch_rpc_timeout(requests), &resource_summary, client.lock_batch(req)) + .execute_rpc("lock_batch", &resource_summary, client.lock_batch(req)) .await { Ok(resp) => resp.into_inner(), @@ -686,11 +666,18 @@ mod tests { #[test] #[serial_test::serial] - fn test_remote_client_rpc_timeout_uses_configured_floor_and_grace() { + fn test_remote_client_rpc_timeout_honors_configured_deadline() { + temp_env::with_var(rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS, None::<&str>, || { + assert_eq!( + RemoteClient::rpc_timeout(), + Duration::from_millis(rustfs_config::DEFAULT_OBJECT_LOCK_RPC_TIMEOUT_MS) + ); + }); temp_env::with_var(rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS, Some("50"), || { - assert_eq!(RemoteClient::rpc_timeout(Duration::ZERO), Duration::from_millis(500)); - assert_eq!(RemoteClient::rpc_timeout(Duration::from_millis(25)), Duration::from_millis(525)); - assert_eq!(RemoteClient::rpc_timeout(Duration::from_secs(1)), Duration::from_millis(1500)); + assert_eq!(RemoteClient::rpc_timeout(), Duration::from_millis(50)); + }); + temp_env::with_var(rustfs_config::ENV_OBJECT_LOCK_RPC_TIMEOUT_MS, Some("0"), || { + assert_eq!(RemoteClient::rpc_timeout(), Duration::from_millis(1)); }); } } diff --git a/rustfs/src/admin/console.rs b/rustfs/src/admin/console.rs index f76678b90..5e7aa18e3 100644 --- a/rustfs/src/admin/console.rs +++ b/rustfs/src/admin/console.rs @@ -19,7 +19,7 @@ use crate::license::has_valid_license; use crate::server::has_path_prefix; use crate::server::{ CONSOLE_PREFIX, FAVICON_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, HeaderMapCarrier, LICENSE, RUSTFS_ADMIN_PREFIX, - RequestContextLayer, VERSION, liveness_dependency_readiness_report, + RequestContextLayer, VERSION, }; use crate::storage::request_context::RequestContext; use crate::version::build; @@ -597,16 +597,17 @@ async fn health_check(method: Method, uri: Uri) -> Response { } else { HealthProbe::Liveness }; - let readiness_report = match probe { - HealthProbe::Liveness => liveness_dependency_readiness_report(), - HealthProbe::Readiness => collect_dependency_readiness().await, + let readiness_report = if probe == HealthProbe::Readiness { + Some(collect_dependency_readiness().await) + } else { + None }; 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); + build_health_response_parts(method.clone(), probe, readiness_report.as_ref(), "rustfs-console", Some(uptime), None); let builder = Response::builder() .status(response_parts.status_code) diff --git a/rustfs/src/admin/handlers/health.rs b/rustfs/src/admin/handlers/health.rs index cd795b62c..787b69ed8 100644 --- a/rustfs/src/admin/handlers/health.rs +++ b/rustfs/src/admin/handlers/health.rs @@ -16,7 +16,7 @@ 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, liveness_dependency_readiness_report, + collect_dependency_readiness_report as collect_runtime_dependency_readiness_report, }; use http::{HeaderMap, HeaderValue}; use hyper::{Method, StatusCode}; @@ -73,6 +73,7 @@ pub(crate) struct HealthPayloadContext<'a> { pub(crate) service: &'a str, pub(crate) uptime: Option, pub(crate) kms_ready: Option, + pub(crate) include_dependency_details: bool, } pub(crate) async fn collect_dependency_readiness() -> crate::server::DependencyReadinessReport { @@ -85,13 +86,21 @@ pub(crate) fn health_check_state( lock_quorum_ready: bool, probe: HealthProbe, ) -> HealthCheckState { + if probe == HealthProbe::Liveness { + return HealthCheckState { + status_code: StatusCode::OK, + status: "ok", + ready: true, + }; + } + let ready = storage_ready && iam_ready && lock_quorum_ready; let status = if ready { "ok" } else { "degraded" }; - let status_code = match probe { - HealthProbe::Liveness => StatusCode::OK, - HealthProbe::Readiness if ready => StatusCode::OK, - HealthProbe::Readiness => StatusCode::SERVICE_UNAVAILABLE, + let status_code = if ready { + StatusCode::OK + } else { + StatusCode::SERVICE_UNAVAILABLE }; HealthCheckState { @@ -159,16 +168,42 @@ pub(crate) fn probe_from_path(path: &str) -> HealthProbe { pub(crate) fn build_health_response_parts( method: Method, probe: HealthProbe, - readiness_report: &crate::server::DependencyReadinessReport, + readiness_report: Option<&crate::server::DependencyReadinessReport>, service: &str, uptime: Option, 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(); + 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)) => { + 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; + ( + storage_ready, + iam_ready, + lock_quorum_ready, + health_check_state(storage_ready, iam_ready, lock_quorum_ready, probe), + readiness_report.degraded_reasons.clone(), + true, + ) + } + (HealthProbe::Readiness, None) => ( + false, + false, + false, + HealthCheckState { + status_code: StatusCode::SERVICE_UNAVAILABLE, + status: "degraded", + ready: false, + }, + vec![crate::server::ReadinessDegradedReason::StorageIamAndLockUnavailable], + true, + ), + (HealthProbe::Liveness, _) => { + (false, false, false, health_check_state(false, false, false, probe), Vec::new(), false) + } + }; if probe == HealthProbe::Readiness && matches!(kms_ready, Some(false)) { health = HealthCheckState { @@ -193,6 +228,7 @@ pub(crate) fn build_health_response_parts( service, uptime, kms_ready, + include_dependency_details, })) }; @@ -216,10 +252,13 @@ pub(crate) fn build_health_payload(ctx: HealthPayloadContext<'_>) -> Value { "service": ctx.service, "timestamp": jiff::Zoned::now().to_string(), "version": env!("CARGO_PKG_VERSION"), - "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 ctx.include_dependency_details { + payload["details"] = build_component_details(ctx.storage_ready, ctx.iam_ready, ctx.lock_quorum_ready, ctx.kms_ready); + payload["degradedReasons"] = build_degraded_reasons(ctx.degraded_reasons); + } + if let Some(uptime) = ctx.uptime { payload["uptime"] = json!(uptime); } @@ -245,12 +284,14 @@ impl Operation for HealthCheckHandler { } let probe = probe_from_path(req.uri.path()); - let readiness_report = match probe { - HealthProbe::Liveness => liveness_dependency_readiness_report(), - HealthProbe::Readiness => collect_dependency_readiness().await, + let readiness_report = if probe == HealthProbe::Readiness { + Some(collect_dependency_readiness().await) + } else { + None }; - let response_parts = build_health_response_parts(method.clone(), probe, &readiness_report, "rustfs-endpoint", None, None); + let response_parts = + build_health_response_parts(method.clone(), probe, readiness_report.as_ref(), "rustfs-endpoint", None, None); let mut headers = HeaderMap::new(); headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); @@ -291,17 +332,8 @@ mod tests { fn test_liveness_state_iam_not_ready() { let state = health_check_state(true, false, true, HealthProbe::Liveness); assert_eq!(state.status_code, StatusCode::OK); - assert_eq!(state.status, "degraded"); - assert!(!state.ready); - } - - #[test] - fn test_liveness_dependency_readiness_report_is_lightweight_ready() { - let report = liveness_dependency_readiness_report(); - assert!(report.readiness.storage_ready); - assert!(report.readiness.iam_ready); - assert!(report.readiness.lock_quorum_ready); - assert!(report.degraded_reasons.is_empty()); + assert_eq!(state.status, "ok"); + assert!(state.ready); } #[test] @@ -361,8 +393,14 @@ mod tests { }, degraded_reasons: vec![crate::server::ReadinessDegradedReason::StorageQuorumUnavailable], }; - let parts = - build_health_response_parts(Method::GET, HealthProbe::Readiness, &readiness_report, "rustfs-endpoint", None, None); + let parts = build_health_response_parts( + Method::GET, + HealthProbe::Readiness, + Some(&readiness_report), + "rustfs-endpoint", + None, + None, + ); assert_eq!(parts.status_code, StatusCode::SERVICE_UNAVAILABLE); } @@ -376,8 +414,14 @@ mod tests { }, degraded_reasons: Vec::new(), }; - let parts = - build_health_response_parts(Method::GET, HealthProbe::Readiness, &readiness_report, "rustfs-endpoint", None, None); + let parts = build_health_response_parts( + Method::GET, + HealthProbe::Readiness, + Some(&readiness_report), + "rustfs-endpoint", + None, + None, + ); assert_eq!(parts.status_code, StatusCode::OK); } @@ -391,9 +435,20 @@ mod tests { }, degraded_reasons: vec![crate::server::ReadinessDegradedReason::StorageAndIamUnavailable], }; - let parts = - build_health_response_parts(Method::GET, HealthProbe::Liveness, &readiness_report, "rustfs-endpoint", None, None); + let parts = build_health_response_parts( + Method::GET, + HealthProbe::Liveness, + Some(&readiness_report), + "rustfs-endpoint", + None, + None, + ); assert_eq!(parts.status_code, StatusCode::OK); + let payload = parts.payload.expect("GET should include payload"); + assert_eq!(payload["status"], "ok"); + assert_eq!(payload["ready"], true); + assert!(payload.get("details").is_none()); + assert!(payload.get("degradedReasons").is_none()); } #[test] @@ -406,8 +461,14 @@ mod tests { }, degraded_reasons: vec![crate::server::ReadinessDegradedReason::StorageAndIamUnavailable], }; - let parts = - build_health_response_parts(Method::HEAD, HealthProbe::Readiness, &readiness_report, "rustfs-endpoint", None, None); + let parts = build_health_response_parts( + Method::HEAD, + HealthProbe::Readiness, + Some(&readiness_report), + "rustfs-endpoint", + None, + None, + ); assert_eq!(parts.status_code, StatusCode::SERVICE_UNAVAILABLE); assert!(parts.payload.is_none()); } @@ -425,6 +486,7 @@ mod tests { service: "rustfs-endpoint", uptime: Some(123), kms_ready: None, + include_dependency_details: true, }); assert_eq!(payload["status"], "degraded"); assert_eq!(payload["ready"], false); @@ -447,6 +509,7 @@ mod tests { service: "rustfs-endpoint", uptime: None, kms_ready: None, + include_dependency_details: true, }); assert_eq!(payload["degradedReasons"][0], "storage_and_iam_unavailable"); } @@ -461,7 +524,8 @@ mod tests { }, degraded_reasons: Vec::new(), }; - let parts = build_health_response_parts(Method::HEAD, HealthProbe::Readiness, &report, "rustfs-endpoint", None, None); + let parts = + build_health_response_parts(Method::HEAD, HealthProbe::Readiness, Some(&report), "rustfs-endpoint", None, None); assert_eq!(parts.status_code, StatusCode::OK); assert!(parts.payload.is_none()); } @@ -476,7 +540,8 @@ mod tests { }, degraded_reasons: vec![crate::server::ReadinessDegradedReason::StorageQuorumUnavailable], }; - let parts = build_health_response_parts(Method::GET, HealthProbe::Readiness, &report, "rustfs-endpoint", None, None); + let parts = + build_health_response_parts(Method::GET, HealthProbe::Readiness, Some(&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"); @@ -495,7 +560,7 @@ mod tests { degraded_reasons: Vec::new(), }; let parts = - build_health_response_parts(Method::GET, HealthProbe::Readiness, &report, "rustfs-endpoint", None, Some(false)); + build_health_response_parts(Method::GET, HealthProbe::Readiness, Some(&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); diff --git a/rustfs/src/server/layer.rs b/rustfs/src/server/layer.rs index 6ac89e84a..800ba7a41 100644 --- a/rustfs/src/server/layer.rs +++ b/rustfs/src/server/layer.rs @@ -23,7 +23,6 @@ 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_LIVE_PATH, MINIO_HEALTH_READY_PATH, RPC_PREFIX, RUSTFS_ADMIN_PREFIX, active_http_requests, collect_dependency_readiness_report, has_path_prefix, is_admin_path, is_table_catalog_path, - liveness_dependency_readiness_report, }; use crate::storage::apply_cors_headers; use crate::storage::request_context::{ @@ -939,9 +938,10 @@ where .expect("failed to build health busy response"); } - let readiness_report = match probe { - HealthProbe::Liveness => liveness_dependency_readiness_report(), - HealthProbe::Readiness => collect_dependency_readiness_report().await, + let readiness_report = if probe == HealthProbe::Readiness { + Some(collect_dependency_readiness_report().await) + } else { + None }; let kms_ready = if probe == HealthProbe::Readiness && health_compat_kms_ready_check_enabled() { Some(health_kms_ready().await) @@ -949,7 +949,8 @@ where None }; - let response_parts = build_health_response_parts(method, probe, &readiness_report, "rustfs-endpoint", None, kms_ready); + let response_parts = + build_health_response_parts(method, probe, readiness_report.as_ref(), "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()))) @@ -1654,7 +1655,12 @@ mod tests { ); let body = BodyExt::collect(response.into_body()).await.expect("body").to_bytes(); - assert!(body.windows(br#""status":"#.len()).any(|window| window == br#""status":"#)); + let payload: serde_json::Value = + serde_json::from_slice(&body).expect("public liveness health response should be valid JSON"); + assert_eq!(payload["status"], "ok"); + assert_eq!(payload["ready"], true); + assert!(payload.get("details").is_none()); + assert!(payload.get("degradedReasons").is_none()); }) .await; } diff --git a/rustfs/src/server/mod.rs b/rustfs/src/server/mod.rs index bc3ec4d30..fe0765749 100644 --- a/rustfs/src/server/mod.rs +++ b/rustfs/src/server/mod.rs @@ -61,7 +61,6 @@ 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::liveness_dependency_readiness_report; pub use readiness::publish_ready_when_runtime_ready; pub(crate) use readiness::snapshot_dependency_readiness_report; diff --git a/rustfs/src/server/readiness.rs b/rustfs/src/server/readiness.rs index 9fde5629c..581c0e456 100644 --- a/rustfs/src/server/readiness.rs +++ b/rustfs/src/server/readiness.rs @@ -442,17 +442,6 @@ fn dependency_readiness_report_from_readiness(readiness: DependencyReadiness) -> } } -pub(crate) fn liveness_dependency_readiness_report() -> DependencyReadinessReport { - DependencyReadinessReport { - readiness: DependencyReadiness { - storage_ready: true, - iam_ready: true, - lock_quorum_ready: true, - }, - degraded_reasons: Vec::new(), - } -} - pub async fn collect_dependency_readiness() -> DependencyReadiness { collect_dependency_readiness_report().await.readiness }