fix(ecstore): send valid ping body in remote locker (#3112)

* fix(ecstore): send valid ping body in remote locker

Build ping requests with a flatbuffer payload so health checks remain compatible with the ping response parser after restart.

* fix(bench): use multi-host warp target during failover

Normalize comma-separated warp host lists in run_object_batch_bench and let four-node failover bench pass BENCH_WARP_HOSTS so rolling restart does not pin load to a single restarting node.

* feat(health): add compat health probes with busy/KMS checks

  - Add /health/live liveness probe endpoint
  - Add busy protection (429) for readiness probes, gated by RUSTFS_HEALTH_COMPAT_BUSY_CHECK_ENABLE
  - Add KMS readiness check for /health/ready, gated by RUSTFS_HEALTH_COMPAT_KMS_READY_CHECK_ENABLE
  - Add lock quorum status caching with TTL to reduce RPC pressure
  - Consolidate health response building into build_health_response_parts
  - Register /health/live in console router and readiness gate
  - Remove MinIO references from newly added health code

* fix(health): decouple kms readiness from lock quorum
This commit is contained in:
houseme
2026-05-29 16:02:50 +08:00
committed by GitHub
parent c257043b63
commit 2b82432f9e
14 changed files with 578 additions and 169 deletions
+18
View File
@@ -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`
+17
View File
@@ -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;
+1 -1
View File
@@ -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
+2 -2
View File
@@ -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
+19 -5
View File
@@ -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(_) => {
+22 -19
View File
@@ -12,7 +12,7 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::admin::handlers::health::{HealthProbe, build_health_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| {
+231 -109
View File
@@ -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<Value>,
}
#[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<u64>,
pub(crate) kms_ready: Option<bool>,
}
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<bool>,
) -> 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,69 +156,75 @@ 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],
service: &str,
uptime: Option<u64>,
) -> Value {
if health_minimal_response_enabled() {
return json!({
"status": health.status,
"ready": health.ready,
});
}
let mut payload = json!({
"status": health.status,
"ready": health.ready,
"service": 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),
});
if let Some(uptime) = uptime {
payload["uptime"] = json!(uptime);
}
payload
}
pub(crate) fn build_health_response(
pub(crate) fn build_health_response_parts(
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(
readiness_report: &crate::server::DependencyReadinessReport,
service: &str,
uptime: Option<u64>,
kms_ready: Option<bool>,
) -> 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,
"rustfs-endpoint",
None,
);
degraded_reasons: &degraded_reasons,
service,
uptime,
kms_ready,
}))
};
let mut headers = HeaderMap::new();
headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
HealthResponseParts {
status_code: health.status_code,
payload,
}
}
if method == Method::HEAD {
return S3Response::with_headers((health.status_code, Body::empty()), headers);
pub(crate) fn build_health_payload(ctx: HealthPayloadContext<'_>) -> Value {
if health_minimal_response_enabled() {
return json!({
"status": ctx.health.status,
"ready": ctx.health.ready,
});
}
let body_str = serde_json::to_string(&health_info).unwrap_or_else(|_| "{}".to_string());
let body = Body::from(body_str);
let mut payload = json!({
"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(ctx.storage_ready, ctx.iam_ready, ctx.lock_quorum_ready, ctx.kms_ready),
"degradedReasons": build_degraded_reasons(ctx.degraded_reasons),
});
S3Response::with_headers((health.status_code, body), headers)
if let Some(uptime) = ctx.uptime {
payload["uptime"] = json!(uptime);
}
payload
}
#[async_trait::async_trait]
@@ -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");
}
}
+4 -3
View File
@@ -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]
+153 -25
View File
@@ -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<HealthProbe> {
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<RestBody, GrpcBody>(
@@ -638,29 +686,41 @@ async fn build_public_health_http_response<RestBody, GrpcBody>(
where
RestBody: From<Bytes>,
{
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::<Bytes>::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::<Bytes>::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() {
+3 -2
View File
@@ -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;
+3
View File
@@ -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.
+80 -1
View File
@@ -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<Option<StorageReadinessCacheEntry
CACHE.get_or_init(|| Mutex::new(None))
}
fn lock_quorum_status_cache() -> &'static Mutex<Option<LockQuorumCacheEntry>> {
static CACHE: OnceLock<Mutex<Option<LockQuorumCacheEntry>>> = OnceLock::new();
CACHE.get_or_init(|| Mutex::new(None))
}
async fn load_cached_storage_readiness() -> Option<bool> {
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<LockQuorumStatus> {
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,
})
);
}
}
@@ -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 <nodeN> node to stop during failover test (default: node4)
--obs-endpoint <url> RUSTFS_OBS_ENDPOINT (default: auto-select by mode)
--bench-endpoint <url> benchmark endpoint (default: http://127.0.0.1:9000)
--bench-warp-hosts <hosts> comma-separated warp hosts (default: node1..node4)
--bench-sizes <sizes> comma list (default: 1KiB,4KiB,11Mi)
--bench-concurrency <n> benchmark concurrency
--bench-concurrencies <list> 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
+22 -6
View File
@@ -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() {