gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-07-27 00:38:16 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix(health): decouple liveness and lock RPC deadlines (#3824)

Browse Source
This commit is contained in:
cxymds
2026-06-24 18:34:51 +08:00
committed by GitHub
parent 57e12e56d2
commit 0ffd561daf
6 changed files with 140 additions and 93 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/rpc/remote_locker.rs
+18 -31
View File
@@ -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<T, F>(
&self,
op: &'static str,
timeout_duration: Duration,
resource_summary: &str,
future: F,
) -> std::result::Result<T, LockError>
async fn execute_rpc<T, F>(&self, op: &'static str, resource_summary: &str, future: F) -> std::result::Result<T, LockError>
where
F: std::future::Future<Output = std::result::Result<T, tonic::Status>>,
{
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<String>) -> LockInfo {
if let Some(lock_info_json) = lock_info_json {
match serde_json::from_str::<LockInfo>(&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));
});
}
}
rustfs/src/admin/console.rs
+6 -5
View File
@@ -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)
rustfs/src/admin/handlers/health.rs
+104 -39
View File
@@ -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<u64>,
pub(crate) kms_ready: Option<bool>,
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<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();
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);
rustfs/src/server/layer.rs
+12 -6
View File
@@ -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;
}
rustfs/src/server/mod.rs
-1
View File
@@ -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;
rustfs/src/server/readiness.rs
-11
View File
@@ -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
}