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

fix(readiness): gate on lock quorum health (#3100)

Browse Source 
* fix(readiness): gate on lock quorum health

Include distributed lock quorum in runtime readiness decisions and expose the signal in health responses.

Map namespace lock quorum failures to ServiceUnavailable instead of InternalError so clients can safely retry while keeping quorum enforcement unchanged.

* fix(ecstore): restore namespace lock error decoding

Add the missing from_u32 arm for NamespaceLockQuorumUnavailable and cover the numeric roundtrip in error tests.
This commit is contained in:
houseme
2026-05-28 01:12:52 +08:00
committed by GitHub
parent 4648de9e62
commit 5dfc1b5f07
9 changed files with 502 additions and 113 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/error.rs
+45
View File
@@ -160,6 +160,14 @@ pub enum StorageError {
NotModified,
#[error("Invalid range specified: {0}")]
InvalidRangeSpec(String),
#[error("Namespace lock quorum unavailable for {mode} lock on {bucket}/{object}: required {required}, achieved {achieved}")]
NamespaceLockQuorumUnavailable {
mode: &'static str,
bucket: String,
object: String,
required: usize,
achieved: usize,
},
// ── Generic ──────────────────────────────────────────────────────
#[error("Unexpected error")]
@@ -204,6 +212,7 @@ impl StorageError {
| StorageError::ErasureWriteQuorum
| StorageError::InsufficientReadQuorum(_, _)
| StorageError::InsufficientWriteQuorum(_, _)
| StorageError::NamespaceLockQuorumUnavailable { .. }
)
}
}
@@ -433,6 +442,19 @@ impl Clone for StorageError {
StorageError::NotModified => StorageError::NotModified,
StorageError::InvalidPartNumber(a) => StorageError::InvalidPartNumber(*a),
StorageError::InvalidRangeSpec(a) => StorageError::InvalidRangeSpec(a.clone()),
StorageError::NamespaceLockQuorumUnavailable {
mode,
bucket,
object,
required,
achieved,
} => StorageError::NamespaceLockQuorumUnavailable {
mode,
bucket: bucket.clone(),
object: object.clone(),
required: *required,
achieved: *achieved,
},
}
}
}
@@ -505,6 +527,7 @@ impl StorageError {
StorageError::InvalidRangeSpec(_) => 0x3D,
StorageError::NotModified => 0x3E,
StorageError::InvalidPartNumber(_) => 0x3F,
StorageError::NamespaceLockQuorumUnavailable { .. } => 0x42,
}
}
@@ -579,6 +602,13 @@ impl StorageError {
0x3D => Some(StorageError::InvalidRangeSpec(Default::default())),
0x3E => Some(StorageError::NotModified),
0x3F => Some(StorageError::InvalidPartNumber(Default::default())),
0x42 => Some(StorageError::NamespaceLockQuorumUnavailable {
mode: "write",
bucket: Default::default(),
object: Default::default(),
required: Default::default(),
achieved: Default::default(),
}),
_ => None,
}
}
@@ -984,6 +1014,17 @@ mod tests {
assert_eq!(StorageError::DecommissionAlreadyRunning.to_u32(), 0x30);
assert_eq!(StorageError::RebalanceAlreadyRunning.to_u32(), 0x40);
assert_eq!(StorageError::OperationCanceled.to_u32(), 0x41);
assert_eq!(
StorageError::NamespaceLockQuorumUnavailable {
mode: "write",
bucket: "bucket".into(),
object: "object".into(),
required: 3,
achieved: 2,
}
.to_u32(),
0x42
);
}
#[test]
@@ -998,6 +1039,10 @@ mod tests {
assert!(matches!(StorageError::from_u32(0x30), Some(StorageError::DecommissionAlreadyRunning)));
assert!(matches!(StorageError::from_u32(0x40), Some(StorageError::RebalanceAlreadyRunning)));
assert!(matches!(StorageError::from_u32(0x41), Some(StorageError::OperationCanceled)));
assert!(matches!(
StorageError::from_u32(0x42),
Some(StorageError::NamespaceLockQuorumUnavailable { .. })
));
// Test invalid code returns None
assert!(StorageError::from_u32(0xFF).is_none());
crates/ecstore/src/set_disk.rs
+42 -72
View File
@@ -622,12 +622,7 @@ impl ObjectIO for SetDisks {
.await?
.get_read_lock(get_lock_acquire_timeout())
.await
.map_err(|e| {
Error::other(format!(
"Failed to acquire read lock: {}",
self.format_lock_error_from_error(bucket, object, "read", &e)
))
})?;
.map_err(|e| self.map_namespace_lock_error(bucket, object, "read", e))?;
// Record lock acquisition for deadlock detection
let _lock_id = record_lock_acquire(bucket, object, "read");
@@ -756,12 +751,12 @@ impl ObjectIO for SetDisks {
if let Some(http_preconditions) = opts.http_preconditions.clone() {
if !opts.no_lock {
let ns_lock = self.new_ns_lock(bucket, object).await?;
object_lock_guard = Some(ns_lock.get_write_lock(get_lock_acquire_timeout()).await.map_err(|e| {
StorageError::other(format!(
"Failed to acquire write lock: {}",
self.format_lock_error_from_error(bucket, object, "write", &e)
))
})?);
object_lock_guard = Some(
ns_lock
.get_write_lock(get_lock_acquire_timeout())
.await
.map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?,
);
}
if let Some(err) = self.check_write_precondition(bucket, object, opts).await {
@@ -975,12 +970,12 @@ impl ObjectIO for SetDisks {
if !opts.no_lock && object_lock_guard.is_none() {
let ns_lock = self.new_ns_lock(bucket, object).await?;
object_lock_guard = Some(ns_lock.get_write_lock(get_lock_acquire_timeout()).await.map_err(|e| {
StorageError::other(format!(
"Failed to acquire write lock: {}",
self.format_lock_error_from_error(bucket, object, "write", &e)
))
})?);
object_lock_guard = Some(
ns_lock
.get_write_lock(get_lock_acquire_timeout())
.await
.map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?,
);
}
let (online_disks, _, op_old_dir) = Self::rename_data(
@@ -1398,12 +1393,7 @@ impl ObjectOperations for SetDisks {
.await?
.get_write_lock(get_lock_acquire_timeout())
.await
.map_err(|e| {
Error::other(format!(
"Failed to acquire write lock: {}",
self.format_lock_error_from_error(src_bucket, src_object, "write", &e)
))
})?;
.map_err(|e| self.map_namespace_lock_error(src_bucket, src_object, "write", e))?;
let disks = self.get_disks_internal().await;
@@ -1785,12 +1775,7 @@ impl ObjectOperations for SetDisks {
.await?
.get_write_lock(get_lock_acquire_timeout())
.await
.map_err(|e| {
Error::other(format!(
"Failed to acquire write lock: {}",
self.format_lock_error_from_error(bucket, object, "write", &e)
))
})?,
.map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?,
)
} else {
None
@@ -1931,12 +1916,7 @@ impl ObjectOperations for SetDisks {
.await?
.get_read_lock(get_lock_acquire_timeout())
.await
.map_err(|e| {
Error::other(format!(
"Failed to acquire read lock: {}",
self.format_lock_error_from_error(bucket, object, "read", &e)
))
})?,
.map_err(|e| self.map_namespace_lock_error(bucket, object, "read", e))?,
)
} else {
None
@@ -1990,12 +1970,7 @@ impl ObjectOperations for SetDisks {
.await?
.get_write_lock(get_lock_acquire_timeout())
.await
.map_err(|e| {
Error::other(format!(
"Failed to acquire write lock: {}",
self.format_lock_error_from_error(bucket, object, "write", &e)
))
})?,
.map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?,
)
} else {
None
@@ -2527,12 +2502,7 @@ impl SetDisks {
.await?
.get_write_lock(get_lock_acquire_timeout())
.await
.map_err(|e| {
Error::other(format!(
"Failed to acquire write lock: {}",
self.format_lock_error_from_error(bucket, object, "write", &e)
))
})?,
.map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?,
)
} else {
None
@@ -3067,12 +3037,12 @@ impl MultipartOperations for SetDisks {
if let Some(http_preconditions) = opts.http_preconditions.clone() {
let object_lock_guard = if !opts.no_lock {
let ns_lock = self.new_ns_lock(bucket, object).await?;
Some(ns_lock.get_write_lock(get_lock_acquire_timeout()).await.map_err(|e| {
StorageError::other(format!(
"Failed to acquire write lock: {}",
self.format_lock_error_from_error(bucket, object, "write", &e)
))
})?)
Some(
ns_lock
.get_write_lock(get_lock_acquire_timeout())
.await
.map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?,
)
} else {
None
};
@@ -3249,12 +3219,12 @@ impl MultipartOperations for SetDisks {
if let Some(http_preconditions) = opts.http_preconditions.clone() {
if !opts.no_lock {
let ns_lock = self.new_ns_lock(bucket, object).await?;
object_lock_guard = Some(ns_lock.get_write_lock(get_lock_acquire_timeout()).await.map_err(|e| {
StorageError::other(format!(
"Failed to acquire write lock: {}",
self.format_lock_error_from_error(bucket, object, "write", &e)
))
})?);
object_lock_guard = Some(
ns_lock
.get_write_lock(get_lock_acquire_timeout())
.await
.map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?,
);
}
if let Some(err) = self.check_write_precondition(bucket, object, opts).await {
@@ -3599,12 +3569,12 @@ impl MultipartOperations for SetDisks {
if !opts.no_lock && object_lock_guard.is_none() {
let ns_lock = self.new_ns_lock(bucket, object).await?;
object_lock_guard = Some(ns_lock.get_write_lock(get_lock_acquire_timeout()).await.map_err(|e| {
StorageError::other(format!(
"Failed to acquire write lock: {}",
self.format_lock_error_from_error(bucket, object, "write", &e)
))
})?);
object_lock_guard = Some(
ns_lock
.get_write_lock(get_lock_acquire_timeout())
.await
.map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?,
);
}
self.cleanup_multipart_path(&parts).await;
@@ -3685,12 +3655,12 @@ impl HealOperations for SetDisks {
) -> Result<(HealResultItem, Option<Error>)> {
let _write_lock_guard = if !opts.no_lock {
let ns_lock = self.new_ns_lock(bucket, object).await?;
Some(ns_lock.get_write_lock(get_lock_acquire_timeout()).await.map_err(|e| {
StorageError::other(format!(
"Failed to acquire write lock: {}",
self.format_lock_error_from_error(bucket, object, "write", &e)
))
})?)
Some(
ns_lock
.get_write_lock(get_lock_acquire_timeout())
.await
.map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?,
)
} else {
None
};
crates/ecstore/src/set_disk/heal.rs
+7 -13
View File
@@ -39,12 +39,12 @@ impl SetDisks {
let write_lock_guard = if !opts.no_lock {
let ns_lock = self.new_ns_lock(bucket, object).await?;
Some(ns_lock.get_write_lock(get_lock_acquire_timeout()).await.map_err(|e| {
StorageError::other(format!(
"Failed to acquire write lock: {}",
self.format_lock_error_from_error(bucket, object, "write", &e)
))
})?)
Some(
ns_lock
.get_write_lock(get_lock_acquire_timeout())
.await
.map_err(|e| self.map_namespace_lock_error(bucket, object, "write", e))?,
)
} else {
None
};
@@ -713,13 +713,7 @@ impl SetDisks {
.await?
.get_write_lock(get_lock_acquire_timeout())
.await
.map_err(|e| {
let message = format!(
"Failed to acquire write lock: {}",
self.format_lock_error_from_error(bucket, object, "write", &e)
);
DiskError::other(message)
})?;
.map_err(|e| DiskError::other(self.map_namespace_lock_error(bucket, object, "write", e).to_string()))?;
self.heal_object_dir_locked(bucket, object, dry_run, remove).await
}
crates/ecstore/src/set_disk/lock.rs
+24
View File
@@ -50,6 +50,30 @@ impl SetDisks {
}
}
pub(super) fn map_namespace_lock_error(
&self,
bucket: &str,
object: &str,
mode: &'static str,
err: rustfs_lock::error::LockError,
) -> StorageError {
match err {
rustfs_lock::error::LockError::QuorumNotReached { required, achieved } => {
StorageError::NamespaceLockQuorumUnavailable {
mode,
bucket: bucket.to_string(),
object: object.to_string(),
required,
achieved,
}
}
other => StorageError::other(format!(
"Failed to acquire {mode} lock: {}",
self.format_lock_error_from_error(bucket, object, mode, &other)
)),
}
}
pub(super) async fn get_disks_internal(&self) -> Vec<Option<DiskStore>> {
let rl = self.disks.read().await;
rustfs/src/admin/console.rs
+3 -1
View File
@@ -526,7 +526,8 @@ async fn health_check(method: Method, uri: Uri) -> Response {
let readiness_report = collect_dependency_readiness().await;
let storage_ready = readiness_report.readiness.storage_ready;
let iam_ready = readiness_report.readiness.iam_ready;
let health = health_check_state(storage_ready, iam_ready, probe);
let lock_quorum_ready = readiness_report.readiness.lock_quorum_ready;
let health = health_check_state(storage_ready, iam_ready, lock_quorum_ready, probe);
let builder = Response::builder()
.status(health.status_code)
@@ -543,6 +544,7 @@ async fn health_check(method: Method, uri: Uri) -> Response {
health,
storage_ready,
iam_ready,
lock_quorum_ready,
&readiness_report.degraded_reasons,
"rustfs-console",
Some(uptime),
rustfs/src/admin/handlers/health.rs
+49 -14
View File
@@ -61,8 +61,13 @@ pub(crate) async fn collect_dependency_readiness() -> crate::server::DependencyR
collect_runtime_dependency_readiness_report().await
}
pub(crate) fn health_check_state(storage_ready: bool, iam_ready: bool, probe: HealthProbe) -> HealthCheckState {
let ready = storage_ready && iam_ready;
pub(crate) fn health_check_state(
storage_ready: bool,
iam_ready: bool,
lock_quorum_ready: bool,
probe: HealthProbe,
) -> HealthCheckState {
let ready = storage_ready && iam_ready && lock_quorum_ready;
let status = if ready { "ok" } else { "degraded" };
let status_code = match probe {
@@ -85,7 +90,7 @@ pub(crate) fn health_minimal_response_enabled() -> bool {
)
}
pub(crate) fn build_component_details(storage_ready: bool, iam_ready: bool) -> Value {
pub(crate) fn build_component_details(storage_ready: bool, iam_ready: bool, lock_quorum_ready: bool) -> Value {
json!({
"storage": {
"status": if storage_ready { "connected" } else { "disconnected" },
@@ -94,6 +99,10 @@ pub(crate) fn build_component_details(storage_ready: bool, iam_ready: bool) -> V
"iam": {
"status": if iam_ready { "connected" } else { "disconnected" },
"ready": iam_ready,
},
"lock": {
"status": if lock_quorum_ready { "connected" } else { "disconnected" },
"ready": lock_quorum_ready,
}
})
}
@@ -119,6 +128,7 @@ 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>,
@@ -136,7 +146,7 @@ pub(crate) fn build_health_payload(
"service": service,
"timestamp": jiff::Zoned::now().to_string(),
"version": env!("CARGO_PKG_VERSION"),
"details": build_component_details(storage_ready, iam_ready),
"details": build_component_details(storage_ready, iam_ready, lock_quorum_ready),
"degradedReasons": build_degraded_reasons(degraded_reasons),
});
@@ -152,10 +162,19 @@ pub(crate) fn build_health_response(
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, probe);
let health_info = build_health_payload(health, storage_ready, iam_ready, degraded_reasons, "rustfs-endpoint", None);
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"));
@@ -195,6 +214,7 @@ impl Operation for HealthCheckHandler {
probe,
readiness_report.readiness.storage_ready,
readiness_report.readiness.iam_ready,
readiness_report.readiness.lock_quorum_ready,
&readiness_report.degraded_reasons,
))
}
@@ -207,7 +227,7 @@ mod tests {
#[test]
fn test_readiness_state_ready() {
let state = health_check_state(true, true, HealthProbe::Readiness);
let state = health_check_state(true, true, true, HealthProbe::Readiness);
assert_eq!(state.status_code, StatusCode::OK);
assert_eq!(state.status, "ok");
assert!(state.ready);
@@ -215,7 +235,7 @@ mod tests {
#[test]
fn test_readiness_state_storage_not_ready() {
let state = health_check_state(false, true, HealthProbe::Readiness);
let state = health_check_state(false, true, true, HealthProbe::Readiness);
assert_eq!(state.status_code, StatusCode::SERVICE_UNAVAILABLE);
assert_eq!(state.status, "degraded");
assert!(!state.ready);
@@ -223,7 +243,7 @@ mod tests {
#[test]
fn test_liveness_state_iam_not_ready() {
let state = health_check_state(true, false, HealthProbe::Liveness);
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);
@@ -231,7 +251,15 @@ mod tests {
#[test]
fn test_readiness_state_iam_not_ready() {
let state = health_check_state(true, false, HealthProbe::Readiness);
let state = health_check_state(true, false, true, HealthProbe::Readiness);
assert_eq!(state.status_code, StatusCode::SERVICE_UNAVAILABLE);
assert_eq!(state.status, "degraded");
assert!(!state.ready);
}
#[test]
fn test_readiness_state_lock_not_ready() {
let state = health_check_state(true, true, false, HealthProbe::Readiness);
assert_eq!(state.status_code, StatusCode::SERVICE_UNAVAILABLE);
assert_eq!(state.status, "degraded");
assert!(!state.ready);
@@ -239,12 +267,14 @@ mod tests {
#[test]
fn test_health_check_component_details() {
let details = build_component_details(true, false);
let details = build_component_details(true, false, false);
assert_eq!(details["storage"]["status"], "connected");
assert_eq!(details["storage"]["ready"], true);
assert_eq!(details["iam"]["status"], "disconnected");
assert_eq!(details["iam"]["ready"], false);
assert_eq!(details["lock"]["status"], "disconnected");
assert_eq!(details["lock"]["ready"], false);
}
#[test]
@@ -265,6 +295,7 @@ mod tests {
HealthProbe::Readiness,
false,
true,
true,
&[crate::server::ReadinessDegradedReason::StorageQuorumUnavailable],
);
assert_eq!(resp.output.0, StatusCode::SERVICE_UNAVAILABLE);
@@ -272,7 +303,7 @@ mod tests {
#[test]
fn test_build_health_response_readiness_returns_200_when_deps_ready() {
let resp = build_health_response(Method::GET, HealthProbe::Readiness, true, true, &[]);
let resp = build_health_response(Method::GET, HealthProbe::Readiness, true, true, true, &[]);
assert_eq!(resp.output.0, StatusCode::OK);
}
@@ -283,6 +314,7 @@ mod tests {
HealthProbe::Liveness,
false,
false,
false,
&[crate::server::ReadinessDegradedReason::StorageAndIamUnavailable],
);
assert_eq!(resp.output.0, StatusCode::OK);
@@ -295,6 +327,7 @@ mod tests {
HealthProbe::Readiness,
false,
false,
false,
&[crate::server::ReadinessDegradedReason::StorageAndIamUnavailable],
);
assert_eq!(resp.output.0, StatusCode::SERVICE_UNAVAILABLE);
@@ -302,12 +335,13 @@ mod tests {
#[test]
fn test_build_health_payload_minimal_mode_returns_status_and_ready_only() {
let health = health_check_state(true, false, HealthProbe::Readiness);
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(
health,
true,
false,
true,
&[crate::server::ReadinessDegradedReason::IamNotReady],
"rustfs-endpoint",
Some(123),
@@ -323,11 +357,12 @@ mod tests {
#[test]
fn test_build_health_payload_includes_degraded_reasons() {
let health = health_check_state(false, false, HealthProbe::Readiness);
let health = health_check_state(false, false, false, HealthProbe::Readiness);
let payload = build_health_payload(
health,
false,
false,
false,
&[crate::server::ReadinessDegradedReason::StorageAndIamUnavailable],
"rustfs-endpoint",
None,
rustfs/src/error.rs
+28
View File
@@ -217,6 +217,7 @@ impl From<StorageError> for ApiError {
StorageError::BucketExists(_) => S3ErrorCode::BucketAlreadyOwnedByYou,
StorageError::StorageFull => S3ErrorCode::ServiceUnavailable,
StorageError::SlowDown => S3ErrorCode::SlowDown,
StorageError::NamespaceLockQuorumUnavailable { .. } => S3ErrorCode::ServiceUnavailable,
StorageError::DecommissionNotStarted => S3ErrorCode::InvalidRequest,
StorageError::DecommissionAlreadyRunning => S3ErrorCode::InvalidRequest,
StorageError::RebalanceAlreadyRunning => S3ErrorCode::InvalidRequest,
@@ -413,6 +414,16 @@ mod tests {
(StorageError::BucketExists("test".into()), S3ErrorCode::BucketAlreadyOwnedByYou),
(StorageError::StorageFull, S3ErrorCode::ServiceUnavailable),
(StorageError::SlowDown, S3ErrorCode::SlowDown),
(
StorageError::NamespaceLockQuorumUnavailable {
mode: "write",
bucket: "bucket".into(),
object: "object".into(),
required: 3,
achieved: 2,
},
S3ErrorCode::ServiceUnavailable,
),
(StorageError::DecommissionNotStarted, S3ErrorCode::InvalidRequest),
(StorageError::DecommissionAlreadyRunning, S3ErrorCode::InvalidRequest),
(StorageError::RebalanceAlreadyRunning, S3ErrorCode::InvalidRequest),
@@ -460,6 +471,23 @@ mod tests {
assert!(s3_error.source().is_some());
}
#[test]
fn test_namespace_lock_quorum_failure_maps_to_service_unavailable_status() {
let api_error: ApiError = StorageError::NamespaceLockQuorumUnavailable {
mode: "write",
bucket: "bucket".into(),
object: "object".into(),
required: 3,
achieved: 2,
}
.into();
let s3_error: S3Error = api_error.into();
assert_eq!(*s3_error.code(), S3ErrorCode::ServiceUnavailable);
assert_eq!(s3_error.status_code(), Some(http::StatusCode::SERVICE_UNAVAILABLE));
}
#[test]
fn test_api_error_to_s3_error_without_source() {
let api_error = ApiError {
rustfs/src/server/layer.rs
+3 -1
View File
@@ -642,7 +642,8 @@ where
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 health = health_check_state(storage_ready, iam_ready, probe);
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()
} else {
@@ -650,6 +651,7 @@ where
health,
storage_ready,
iam_ready,
lock_quorum_ready,
&readiness_report.degraded_reasons,
"rustfs-endpoint",
None,
rustfs/src/server/readiness.rs
+301 -12
View File
@@ -20,6 +20,8 @@ use http_body_util::{BodyExt, Full};
use hyper::body::Incoming;
use metrics::{counter, gauge};
use rustfs_common::GlobalReadiness;
use rustfs_ecstore::global::is_dist_erasure;
use rustfs_ecstore::global::{get_global_endpoints_opt, get_global_lock_clients};
use rustfs_ecstore::new_object_layer_fn;
use rustfs_ecstore::store_api::StorageAPI;
use rustfs_iam::get_global_iam_sys;
@@ -47,13 +49,18 @@ const METRIC_RUNTIME_READINESS_DEGRADED_TOTAL: &str = "rustfs_runtime_readiness_
pub struct DependencyReadiness {
pub storage_ready: bool,
pub iam_ready: bool,
pub lock_quorum_ready: bool,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub enum ReadinessDegradedReason {
StorageQuorumUnavailable,
IamNotReady,
LockQuorumUnavailable,
StorageAndIamUnavailable,
StorageAndLockUnavailable,
IamAndLockUnavailable,
StorageIamAndLockUnavailable,
}
impl ReadinessDegradedReason {
@@ -61,7 +68,11 @@ impl ReadinessDegradedReason {
match self {
ReadinessDegradedReason::StorageQuorumUnavailable => "storage_quorum_unavailable",
ReadinessDegradedReason::IamNotReady => "iam_not_ready",
ReadinessDegradedReason::LockQuorumUnavailable => "lock_quorum_unavailable",
ReadinessDegradedReason::StorageAndIamUnavailable => "storage_and_iam_unavailable",
ReadinessDegradedReason::StorageAndLockUnavailable => "storage_and_lock_unavailable",
ReadinessDegradedReason::IamAndLockUnavailable => "iam_and_lock_unavailable",
ReadinessDegradedReason::StorageIamAndLockUnavailable => "storage_iam_and_lock_unavailable",
}
}
}
@@ -211,6 +222,14 @@ struct StorageReadinessCacheEntry {
storage_ready: bool,
}
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
pub struct LockQuorumStatus {
pub ready: bool,
pub connected_clients: usize,
pub total_clients: usize,
pub required_quorum: usize,
}
const DISK_STATE_OK: &str = "ok";
const DISK_STATE_UNFORMATTED: &str = "unformatted";
const RUNTIME_STATE_RETURNING: &str = "returning";
@@ -346,17 +365,21 @@ fn storage_ready_from_runtime_state(info: &StorageInfo) -> bool {
})
}
fn degraded_reasons(storage_ready: bool, iam_ready_raw: bool) -> Vec<ReadinessDegradedReason> {
match (storage_ready, iam_ready_raw) {
(true, true) => Vec::new(),
(false, false) => vec![ReadinessDegradedReason::StorageAndIamUnavailable],
(false, true) => vec![ReadinessDegradedReason::StorageQuorumUnavailable],
(true, false) => vec![ReadinessDegradedReason::IamNotReady],
fn degraded_reasons(storage_ready: bool, iam_ready_raw: bool, lock_quorum_ready: bool) -> Vec<ReadinessDegradedReason> {
match (storage_ready, iam_ready_raw, lock_quorum_ready) {
(true, true, true) => Vec::new(),
(false, false, false) => vec![ReadinessDegradedReason::StorageIamAndLockUnavailable],
(false, false, true) => vec![ReadinessDegradedReason::StorageAndIamUnavailable],
(false, true, false) => vec![ReadinessDegradedReason::StorageAndLockUnavailable],
(true, false, false) => vec![ReadinessDegradedReason::IamAndLockUnavailable],
(false, true, true) => vec![ReadinessDegradedReason::StorageQuorumUnavailable],
(true, false, true) => vec![ReadinessDegradedReason::IamNotReady],
(true, true, false) => vec![ReadinessDegradedReason::LockQuorumUnavailable],
}
}
fn record_readiness_report(report: &DependencyReadinessReport) {
let ready = report.readiness.storage_ready && report.readiness.iam_ready;
let ready = report.readiness.storage_ready && report.readiness.iam_ready && report.readiness.lock_quorum_ready;
gauge!(METRIC_RUNTIME_READINESS_READY).set(if ready { 1.0 } else { 0.0 });
for reason in &report.degraded_reasons {
counter!(METRIC_RUNTIME_READINESS_DEGRADED_TOTAL, "reason" => reason.as_str()).increment(1);
@@ -376,13 +399,15 @@ 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 readiness = DependencyReadiness {
storage_ready,
iam_ready: iam_ready_raw,
lock_quorum_ready: lock_quorum_status.ready,
};
let report = DependencyReadinessReport {
degraded_reasons: degraded_reasons(readiness.storage_ready, iam_ready_raw),
degraded_reasons: degraded_reasons(readiness.storage_ready, iam_ready_raw, readiness.lock_quorum_ready),
readiness,
};
record_readiness_report(&report);
@@ -392,13 +417,15 @@ pub async fn collect_dependency_readiness_report() -> DependencyReadinessReport
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;
let lock_quorum_status = collect_lock_quorum_status_uncached().await;
let readiness = DependencyReadiness {
storage_ready,
iam_ready: iam_ready_raw,
lock_quorum_ready: lock_quorum_status.ready,
};
let report = DependencyReadinessReport {
degraded_reasons: degraded_reasons(readiness.storage_ready, iam_ready_raw),
degraded_reasons: degraded_reasons(readiness.storage_ready, iam_ready_raw, readiness.lock_quorum_ready),
readiness,
};
record_readiness_report(&report);
@@ -414,6 +441,115 @@ async fn collect_storage_readiness_uncached() -> bool {
}
}
fn set_lock_quorum_status(
online_hosts: &HashSet<String>,
set_endpoints: &[rustfs_ecstore::disk::endpoint::Endpoint],
) -> LockQuorumStatus {
let total_clients = set_endpoints
.iter()
.map(rustfs_ecstore::disk::endpoint::Endpoint::host_port)
.filter(|host| !host.is_empty())
.collect::<HashSet<_>>();
let total_clients_len = total_clients.len();
if total_clients_len == 0 {
return LockQuorumStatus::default();
}
let connected_clients = total_clients.iter().filter(|host| online_hosts.contains(*host)).count();
let required_quorum = if total_clients_len > 1 {
(total_clients_len / 2) + 1
} else {
1
};
LockQuorumStatus {
ready: connected_clients >= required_quorum,
connected_clients,
total_clients: total_clients_len,
required_quorum,
}
}
fn aggregate_lock_quorum_status(
pool_endpoints: &rustfs_ecstore::endpoints::EndpointServerPools,
online_hosts: &HashSet<String>,
) -> LockQuorumStatus {
let mut connected_clients = 0usize;
let mut total_clients = 0usize;
let mut required_quorum = 0usize;
for pool in pool_endpoints.as_ref() {
for set_idx in 0..pool.set_count {
let set_endpoints = pool
.endpoints
.as_ref()
.iter()
.filter(|endpoint| endpoint.set_idx == set_idx as i32)
.cloned()
.collect::<Vec<_>>();
let status = set_lock_quorum_status(online_hosts, &set_endpoints);
if status.total_clients == 0 {
return LockQuorumStatus::default();
}
connected_clients += status.connected_clients;
total_clients += status.total_clients;
required_quorum += status.required_quorum;
if !status.ready {
return LockQuorumStatus {
ready: false,
connected_clients,
total_clients,
required_quorum,
};
}
}
}
if total_clients == 0 {
LockQuorumStatus::default()
} else {
LockQuorumStatus {
ready: true,
connected_clients,
total_clients,
required_quorum,
}
}
}
async fn collect_lock_quorum_status_uncached() -> LockQuorumStatus {
if !is_dist_erasure().await {
return LockQuorumStatus {
ready: true,
connected_clients: 1,
total_clients: 1,
required_quorum: 1,
};
}
let Some(pool_endpoints) = get_global_endpoints_opt() else {
return LockQuorumStatus::default();
};
let Some(lock_clients) = get_global_lock_clients() else {
return LockQuorumStatus::default();
};
let online_hosts = futures::future::join_all(lock_clients.iter().map(|(host, client)| {
let host = host.clone();
let client = client.clone();
async move { (host, client.is_online().await) }
}))
.await
.into_iter()
.filter_map(|(host, online)| online.then_some(host))
.collect::<HashSet<_>>();
aggregate_lock_quorum_status(&pool_endpoints, &online_hosts)
}
pub async fn wait_for_runtime_readiness_with<F, Fut, ReadyFn>(
max_wait: Duration,
poll_interval: Duration,
@@ -429,17 +565,18 @@ where
loop {
let readiness = load_readiness().await;
if readiness.storage_ready && readiness.iam_ready {
if readiness.storage_ready && readiness.iam_ready && readiness.lock_quorum_ready {
on_ready(readiness);
return Ok(());
}
if tokio::time::Instant::now() >= startup_deadline {
let reason = format!(
"startup readiness timed out after {}s: storage_ready={}, iam_ready={}",
"startup readiness timed out after {}s: storage_ready={}, iam_ready={}, lock_quorum_ready={}",
max_wait.as_secs(),
readiness.storage_ready,
readiness.iam_ready
readiness.iam_ready,
readiness.lock_quorum_ready
);
return Err(std::io::Error::other(reason));
}
@@ -448,6 +585,7 @@ where
target: "rustfs::server::readiness",
storage_ready = readiness.storage_ready,
iam_ready = readiness.iam_ready,
lock_quorum_ready = readiness.lock_quorum_ready,
"Runtime readiness has not reached write quorum yet; delaying ready state publication"
);
tokio::time::sleep(poll_interval).await;
@@ -479,6 +617,7 @@ mod tests {
future::ready(DependencyReadiness {
storage_ready: false,
iam_ready: false,
lock_quorum_ready: false,
})
},
|_| {
@@ -494,6 +633,34 @@ mod tests {
assert_eq!(state_manager.current_state(), ServiceState::Starting);
}
#[tokio::test]
async fn wait_for_runtime_readiness_with_does_not_publish_ready_when_lock_quorum_is_not_reached() {
let readiness = GlobalReadiness::new();
let state_manager = ServiceStateManager::new();
let err = wait_for_runtime_readiness_with(
Duration::ZERO,
Duration::from_millis(1),
|| {
future::ready(DependencyReadiness {
storage_ready: true,
iam_ready: true,
lock_quorum_ready: false,
})
},
|_| {
readiness.mark_stage(rustfs_common::SystemStage::FullReady);
state_manager.update(ServiceState::Ready);
},
)
.await
.expect_err("lock quorum should block readiness publication");
assert!(err.to_string().contains("lock_quorum_ready=false"));
assert!(!readiness.is_ready());
assert_eq!(state_manager.current_state(), ServiceState::Starting);
}
#[test]
fn storage_ready_from_runtime_state_returns_false_when_all_disks_faulty() {
let info = StorageInfo {
@@ -605,4 +772,126 @@ mod tests {
"if any set fails write quorum, readiness must be false"
);
}
#[test]
fn aggregate_lock_quorum_status_requires_each_set_to_meet_quorum() {
use rustfs_ecstore::disk::endpoint::Endpoint;
use rustfs_ecstore::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints};
let endpoints = vec![
Endpoint {
url: url::Url::parse("http://node1:9000/data1").unwrap(),
is_local: false,
pool_idx: 0,
set_idx: 0,
disk_idx: 0,
},
Endpoint {
url: url::Url::parse("http://node2:9000/data2").unwrap(),
is_local: false,
pool_idx: 0,
set_idx: 0,
disk_idx: 1,
},
Endpoint {
url: url::Url::parse("http://node3:9000/data3").unwrap(),
is_local: false,
pool_idx: 0,
set_idx: 1,
disk_idx: 0,
},
Endpoint {
url: url::Url::parse("http://node4:9000/data4").unwrap(),
is_local: false,
pool_idx: 0,
set_idx: 1,
disk_idx: 1,
},
];
let pools = EndpointServerPools::from(vec![PoolEndpoints {
legacy: false,
set_count: 2,
drives_per_set: 2,
endpoints: Endpoints::from(endpoints),
cmd_line: String::new(),
platform: String::new(),
}]);
let status = aggregate_lock_quorum_status(
&pools,
&["node1:9000", "node2:9000", "node3:9000", "node4:9000"]
.into_iter()
.map(str::to_string)
.collect::<HashSet<_>>(),
);
assert!(status.ready);
assert_eq!(status.connected_clients, 4);
assert_eq!(status.total_clients, 4);
assert_eq!(status.required_quorum, 4);
}
#[test]
fn aggregate_lock_quorum_status_fails_when_any_set_loses_quorum() {
use rustfs_ecstore::disk::endpoint::Endpoint;
use rustfs_ecstore::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints};
let endpoints = vec![
Endpoint {
url: url::Url::parse("http://node1:9000/data1").unwrap(),
is_local: false,
pool_idx: 0,
set_idx: 0,
disk_idx: 0,
},
Endpoint {
url: url::Url::parse("http://node2:9000/data2").unwrap(),
is_local: false,
pool_idx: 0,
set_idx: 0,
disk_idx: 1,
},
Endpoint {
url: url::Url::parse("http://node3:9000/data3").unwrap(),
is_local: false,
pool_idx: 0,
set_idx: 1,
disk_idx: 0,
},
Endpoint {
url: url::Url::parse("http://node4:9000/data4").unwrap(),
is_local: false,
pool_idx: 0,
set_idx: 1,
disk_idx: 1,
},
];
let pools = EndpointServerPools::from(vec![PoolEndpoints {
legacy: false,
set_count: 2,
drives_per_set: 2,
endpoints: Endpoints::from(endpoints),
cmd_line: String::new(),
platform: String::new(),
}]);
let status =
aggregate_lock_quorum_status(&pools, &["node1:9000"].into_iter().map(str::to_string).collect::<HashSet<_>>());
assert!(!status.ready);
}
#[test]
fn degraded_reasons_include_lock_quorum_failures() {
assert_eq!(degraded_reasons(true, true, false), vec![ReadinessDegradedReason::LockQuorumUnavailable]);
assert_eq!(
degraded_reasons(false, true, false),
vec![ReadinessDegradedReason::StorageAndLockUnavailable]
);
assert_eq!(degraded_reasons(true, false, false), vec![ReadinessDegradedReason::IamAndLockUnavailable]);
assert_eq!(
degraded_reasons(false, false, false),
vec![ReadinessDegradedReason::StorageIamAndLockUnavailable]
);
}
}