diff --git a/rustfs/src/server/health.rs b/rustfs/src/server/health.rs index a9ed2e431..47f4357c1 100644 --- a/rustfs/src/server/health.rs +++ b/rustfs/src/server/health.rs @@ -321,13 +321,15 @@ pub(crate) fn build_health_response_parts( ), }; - let object_traffic_stalled = degraded_reasons.iter().any(|reason| { + let readiness_overlay_degraded = degraded_reasons.iter().any(|reason| { matches!( reason, - ReadinessDegradedReason::ObjectReadStalled | ReadinessDegradedReason::ObjectWriteStalled + ReadinessDegradedReason::ObjectReadStalled + | ReadinessDegradedReason::ObjectWriteStalled + | ReadinessDegradedReason::StartupFinalizationPending ) }); - if probe == HealthProbe::Readiness && (object_traffic_stalled || matches!(kms_ready, Some(false))) { + if probe == HealthProbe::Readiness && (readiness_overlay_degraded || matches!(kms_ready, Some(false))) { health = HealthCheckState { status_code: StatusCode::SERVICE_UNAVAILABLE, status: "degraded", diff --git a/rustfs/src/server/http.rs b/rustfs/src/server/http.rs index b3ac5e93d..131413b65 100644 --- a/rustfs/src/server/http.rs +++ b/rustfs/src/server/http.rs @@ -1677,7 +1677,10 @@ fn process_connection( .option_layer(if is_console { Some(RedirectLayer) } else { None }) .layer(BodylessStatusFixLayer) .layer(HeadRequestBodyFixLayer) - .layer(PublicHealthEndpointLayer::new(Arc::clone(&server_ctx))) + .layer(PublicHealthEndpointLayer::new( + Arc::clone(&server_ctx), + Arc::clone(&readiness), + )) .option_layer((!server_domains_configured && !is_console).then_some(VirtualHostStyleHintLayer)) .layer(DoubleSlashListBucketsCompatLayer) .service(service) diff --git a/rustfs/src/server/layer.rs b/rustfs/src/server/layer.rs index 7fea8a838..b6e585380 100644 --- a/rustfs/src/server/layer.rs +++ b/rustfs/src/server/layer.rs @@ -26,6 +26,7 @@ use crate::server::{ build_health_response_parts, collect_probe_readiness, has_path_prefix, is_admin_path, is_table_catalog_path, kms_probe_staleness_limit, kms_ready_from_probe, }; +use crate::shared_types::ReadinessDegradedReason; use crate::storage_api::server::layer::apply_cors_headers; use crate::storage_api::server::layer::request_context::{RequestContext, extract_request_id_from_headers, spawn_traced}; use bytes::{Bytes, BytesMut}; @@ -36,6 +37,7 @@ use http_body_util::{BodyExt, Full}; use hyper::body::Incoming; use pin_project_lite::pin_project; use quick_xml::events::Event; +use rustfs_common::GlobalReadiness; use rustfs_obs::HTTP_SERVER_LOG_TARGET; #[cfg(feature = "swift")] use rustfs_protocols::swift::SwiftRouter; @@ -1243,11 +1245,12 @@ where #[derive(Clone)] pub struct PublicHealthEndpointLayer { server_ctx: Arc, + readiness: Arc, } impl PublicHealthEndpointLayer { - pub fn new(server_ctx: Arc) -> Self { - Self { server_ctx } + pub fn new(server_ctx: Arc, readiness: Arc) -> Self { + Self { server_ctx, readiness } } } @@ -1258,6 +1261,7 @@ impl Layer for PublicHealthEndpointLayer { PublicHealthEndpointService { inner, server_ctx: Arc::clone(&self.server_ctx), + readiness: Arc::clone(&self.readiness), } } } @@ -1266,6 +1270,7 @@ impl Layer for PublicHealthEndpointLayer { pub struct PublicHealthEndpointService { inner: S, server_ctx: Arc, + readiness: Arc, } fn health_endpoint_enabled() -> bool { @@ -1334,6 +1339,7 @@ async fn build_public_health_http_response( method: Method, path: String, object_traffic_health: Option>, + readiness: &GlobalReadiness, ) -> Response> where RestBody: From, @@ -1358,7 +1364,15 @@ where .expect("failed to build health busy response"); } - let readiness_report = collect_probe_readiness(probe, object_traffic_health.as_deref()).await; + let mut readiness_report = collect_probe_readiness(probe, object_traffic_health.as_deref()).await; + if probe == HealthProbe::Readiness + && !readiness.is_ready() + && let Some(report) = readiness_report.as_mut() + { + report + .degraded_reasons + .push(ReadinessDegradedReason::StartupFinalizationPending); + } let kms_ready = if probe == HealthProbe::Readiness && health_compat_kms_ready_check_enabled() { Some(health_kms_ready().await) } else { @@ -1408,7 +1422,10 @@ where .server_ctx .installed_app_context() .map(|context| context.object_traffic_health()); - return Box::pin(async move { Ok(build_public_health_http_response(method, path, object_traffic_health).await) }); + let readiness = Arc::clone(&self.readiness); + return Box::pin(async move { + Ok(build_public_health_http_response(method, path, object_traffic_health, readiness.as_ref()).await) + }); } let mut inner = self.inner.clone(); @@ -2210,14 +2227,25 @@ mod tests { use tracing_subscriber::{Registry, fmt::MakeWriter, layer::SubscriberExt}; fn public_health_layer() -> PublicHealthEndpointLayer { - PublicHealthEndpointLayer::new(crate::runtime_sources::ServerContextSlot::new()) + let readiness = Arc::new(GlobalReadiness::new()); + readiness.mark_stage(rustfs_common::SystemStage::FullReady); + PublicHealthEndpointLayer::new(crate::runtime_sources::ServerContextSlot::new(), readiness) } async fn public_health_layer_with_tracker(object_traffic_health: Arc) -> PublicHealthEndpointLayer { + let readiness = Arc::new(GlobalReadiness::new()); + readiness.mark_stage(rustfs_common::SystemStage::FullReady); + public_health_layer_with_tracker_and_readiness(object_traffic_health, readiness).await + } + + async fn public_health_layer_with_tracker_and_readiness( + object_traffic_health: Arc, + readiness: Arc, + ) -> PublicHealthEndpointLayer { let app_context = crate::app::gating_test_env::app_context_with_object_traffic_health(object_traffic_health).await; let server_ctx = crate::runtime_sources::ServerContextSlot::new(); assert!(server_ctx.install(app_context)); - PublicHealthEndpointLayer::new(server_ctx) + PublicHealthEndpointLayer::new(server_ctx, readiness) } #[derive(Clone, Debug)] @@ -2987,6 +3015,82 @@ mod tests { .await; } + #[tokio::test] + #[serial] + async fn public_readiness_waits_for_s3_admission_publication() { + async_with_vars( + [ + (rustfs_config::ENV_HEALTH_ENDPOINT_ENABLE, Some("true")), + (rustfs_config::ENV_HEALTH_MINIMAL_RESPONSE_ENABLE, Some("false")), + ], + async { + let object_traffic_health = Arc::new(ObjectTrafficHealth::enabled_for_test(Duration::ZERO)); + let readiness = Arc::new(GlobalReadiness::new()); + let inner = CountingHybridService::default(); + let calls = inner.calls(); + let mut service = public_health_layer_with_tracker_and_readiness(object_traffic_health, Arc::clone(&readiness)) + .await + .layer(inner); + + let response = service + .call( + Request::builder() + .method(Method::GET) + .uri(HEALTH_READY_PATH) + .body(Full::::from(Bytes::new())) + .expect("readiness request before admission publication"), + ) + .await + .expect("readiness response before admission publication"); + assert_eq!(response.status(), StatusCode::SERVICE_UNAVAILABLE); + let body = BodyExt::collect(response.into_body()) + .await + .expect("readiness body before admission publication") + .to_bytes(); + let payload: serde_json::Value = serde_json::from_slice(&body).expect("readiness JSON"); + assert_eq!(payload["ready"], false); + assert_eq!(payload["details"]["storage"]["ready"], true); + assert_eq!(payload["details"]["iam"]["ready"], true); + assert_eq!(payload["details"]["lock"]["ready"], true); + assert_eq!(payload["degradedReasons"], serde_json::json!(["startup_finalization_pending"])); + + let response = service + .call( + Request::builder() + .method(Method::GET) + .uri(HEALTH_COMPAT_LIVE_PATH) + .body(Full::::from(Bytes::new())) + .expect("liveness request before admission publication"), + ) + .await + .expect("liveness response before admission publication"); + assert_eq!(response.status(), StatusCode::OK); + let body = BodyExt::collect(response.into_body()) + .await + .expect("liveness body before admission publication") + .to_bytes(); + let payload: serde_json::Value = serde_json::from_slice(&body).expect("liveness JSON"); + assert_eq!(payload["status"], "ok"); + assert!(payload.get("ready").is_none()); + + readiness.mark_stage(rustfs_common::SystemStage::FullReady); + let response = service + .call( + Request::builder() + .method(Method::HEAD) + .uri(MINIO_HEALTH_READY_PATH) + .body(Full::::from(Bytes::new())) + .expect("readiness request after admission publication"), + ) + .await + .expect("readiness response after admission publication"); + assert_eq!(response.status(), StatusCode::OK); + assert_eq!(calls.load(Ordering::SeqCst), 0); + }, + ) + .await; + } + #[tokio::test] #[serial] async fn public_readiness_aliases_use_the_installed_object_progress() { diff --git a/rustfs/src/shared_types.rs b/rustfs/src/shared_types.rs index 2f351383b..d307d5520 100644 --- a/rustfs/src/shared_types.rs +++ b/rustfs/src/shared_types.rs @@ -45,6 +45,7 @@ pub enum ReadinessDegradedReason { ObjectWriteStalled, ClusterHealthTimeout, PeerHealthUnavailable, + StartupFinalizationPending, StorageAndIamUnavailable, StorageAndLockUnavailable, IamAndLockUnavailable, @@ -62,6 +63,7 @@ impl ReadinessDegradedReason { ReadinessDegradedReason::ObjectWriteStalled => "object_write_stalled", ReadinessDegradedReason::ClusterHealthTimeout => "cluster_health_timeout", ReadinessDegradedReason::PeerHealthUnavailable => "peer_health_unavailable", + ReadinessDegradedReason::StartupFinalizationPending => "startup_finalization_pending", ReadinessDegradedReason::StorageAndIamUnavailable => "storage_and_iam_unavailable", ReadinessDegradedReason::StorageAndLockUnavailable => "storage_and_lock_unavailable", ReadinessDegradedReason::IamAndLockUnavailable => "iam_and_lock_unavailable",