mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-26 05:56:50 +00:00
feat: improve degraded readiness reporting and shutdown handling (#3089)
* fix(server): unify runtime readiness and shutdown flow * refactor(server): decouple readiness shared types * refactor(server): keep readiness types crate-private * refactor(server): await protocol shutdown handles * fix(server): bypass cached startup readiness * feat(health): expose degraded readiness reasons * fix(health): preserve raw IAM readiness in degraded reports
This commit is contained in:
Generated
+319
-338
File diff suppressed because it is too large
Load Diff
+8
-8
@@ -117,11 +117,11 @@ rustfs-zip = { path = "./crates/zip", version = "1.0.0-beta.4" }
|
|||||||
|
|
||||||
# Async Runtime and Networking
|
# Async Runtime and Networking
|
||||||
async-channel = "2.5.0"
|
async-channel = "2.5.0"
|
||||||
mysql_async = { version = "0.36.1", default-features = false, features = ["default-rustls", "tracing"], git = "https://github.com/blackbeam/mysql_async", rev = "2bad388283bc3ce48801fc2ffcd22445eb6f3d24" }
|
mysql_async = { version = "0.37", default-features = false, features = ["default-rustls", "tracing"] }
|
||||||
async-compression = { version = "0.4.42" }
|
async-compression = { version = "0.4.42" }
|
||||||
async-recursion = "1.1.1"
|
async-recursion = "1.1.1"
|
||||||
async-trait = "0.1.89"
|
async-trait = "0.1.89"
|
||||||
async-nats = "0.48.0"
|
async-nats = "0.49.0"
|
||||||
axum = "0.8.9"
|
axum = "0.8.9"
|
||||||
futures = "0.3.32"
|
futures = "0.3.32"
|
||||||
futures-core = "0.3.32"
|
futures-core = "0.3.32"
|
||||||
@@ -132,10 +132,10 @@ lapin = { version = "4.10.0", default-features = false, features = ["tokio", "ru
|
|||||||
hyper = { version = "1.9.0", features = ["http2", "http1", "server"] }
|
hyper = { version = "1.9.0", features = ["http2", "http1", "server"] }
|
||||||
hyper-rustls = { version = "0.27.9", default-features = false, features = ["native-tokio", "http1", "tls12", "logging", "http2", "aws-lc-rs", "webpki-roots"] }
|
hyper-rustls = { version = "0.27.9", default-features = false, features = ["native-tokio", "http1", "tls12", "logging", "http2", "aws-lc-rs", "webpki-roots"] }
|
||||||
hyper-util = { version = "0.1.20", features = ["tokio", "server-auto", "server-graceful", "tracing"] }
|
hyper-util = { version = "0.1.20", features = ["tokio", "server-auto", "server-graceful", "tracing"] }
|
||||||
http = "1.4.0"
|
http = "1.4.1"
|
||||||
http-body = "1.0.1"
|
http-body = "1.0.1"
|
||||||
http-body-util = "0.1.3"
|
http-body-util = "0.1.3"
|
||||||
reqwest = { version = "0.13.3", default-features = false, features = ["rustls", "charset", "http2", "system-proxy", "stream", "json", "blocking", "query", "form"] }
|
reqwest = { version = "0.13.4", default-features = false, features = ["rustls", "charset", "http2", "system-proxy", "stream", "json", "blocking", "query", "form"] }
|
||||||
rustfs-kafka-async = { version = "1.2.0" }
|
rustfs-kafka-async = { version = "1.2.0" }
|
||||||
socket2 = { version = "0.6.3", features = ["all"] }
|
socket2 = { version = "0.6.3", features = ["all"] }
|
||||||
tokio = { version = "1.52.3", features = ["fs", "rt-multi-thread"] }
|
tokio = { version = "1.52.3", features = ["fs", "rt-multi-thread"] }
|
||||||
@@ -164,7 +164,7 @@ serde_json = { version = "1.0.150", features = ["raw_value"] }
|
|||||||
serde_urlencoded = "0.7.1"
|
serde_urlencoded = "0.7.1"
|
||||||
|
|
||||||
# Cryptography and Security
|
# Cryptography and Security
|
||||||
aes-gcm = { version = "0.11.0-rc.3", features = ["rand_core"] }
|
aes-gcm = { version = "0.11.0-rc.4", features = ["rand_core"] }
|
||||||
argon2 = { version = "0.6.0-rc.8" }
|
argon2 = { version = "0.6.0-rc.8" }
|
||||||
blake2 = "0.11.0-rc.6"
|
blake2 = "0.11.0-rc.6"
|
||||||
chacha20poly1305 = { version = "0.11.0-rc.3" }
|
chacha20poly1305 = { version = "0.11.0-rc.3" }
|
||||||
@@ -185,7 +185,7 @@ zeroize = { version = "1.8.2", features = ["derive"] }
|
|||||||
# Time and Date
|
# Time and Date
|
||||||
chrono = { version = "0.4.44", features = ["serde"] }
|
chrono = { version = "0.4.44", features = ["serde"] }
|
||||||
humantime = "2.3.0"
|
humantime = "2.3.0"
|
||||||
jiff = { version = "0.2.24", features = ["serde"] }
|
jiff = { version = "0.2.27", features = ["serde"] }
|
||||||
time = { version = "0.3.47", features = ["std", "parsing", "formatting", "macros", "serde"] }
|
time = { version = "0.3.47", features = ["std", "parsing", "formatting", "macros", "serde"] }
|
||||||
|
|
||||||
# Database
|
# Database
|
||||||
@@ -298,7 +298,7 @@ dial9-tokio-telemetry = "0.3"
|
|||||||
opentelemetry = { version = "0.32.0" }
|
opentelemetry = { version = "0.32.0" }
|
||||||
opentelemetry-appender-tracing = { version = "0.32.0", features = ["experimental_span_attributes", "experimental_metadata_attributes"] }
|
opentelemetry-appender-tracing = { version = "0.32.0", features = ["experimental_span_attributes", "experimental_metadata_attributes"] }
|
||||||
opentelemetry-otlp = { version = "0.32.0", features = ["gzip-http", "reqwest-rustls"] }
|
opentelemetry-otlp = { version = "0.32.0", features = ["gzip-http", "reqwest-rustls"] }
|
||||||
opentelemetry_sdk = { version = "0.32.0", features = ["rt-tokio"] }
|
opentelemetry_sdk = { version = "0.32.1", features = ["rt-tokio"] }
|
||||||
opentelemetry-semantic-conventions = { version = "0.32.0", features = ["semconv_experimental"] }
|
opentelemetry-semantic-conventions = { version = "0.32.0", features = ["semconv_experimental"] }
|
||||||
opentelemetry-stdout = { version = "0.32.0" }
|
opentelemetry-stdout = { version = "0.32.0" }
|
||||||
pyroscope = { version = "2.0.4", features = ["backend-pprof-rs"] }
|
pyroscope = { version = "2.0.4", features = ["backend-pprof-rs"] }
|
||||||
@@ -308,7 +308,7 @@ libunftp = { version = "0.23.0", features = ["experimental"] }
|
|||||||
unftp-core = "0.1.0"
|
unftp-core = "0.1.0"
|
||||||
suppaftp = { version = "8.0.3", features = ["tokio", "tokio-rustls-aws-lc-rs"] }
|
suppaftp = { version = "8.0.3", features = ["tokio", "tokio-rustls-aws-lc-rs"] }
|
||||||
rcgen = "0.14.8"
|
rcgen = "0.14.8"
|
||||||
russh = { version = "0.60.3", git = "https://github.com/Eugeny/russh", rev = "fc6e3ab4cd4338e94ae64e17aeed2acee9335e6b" }
|
russh = { version = "0.61.1",features = ["serde"] }
|
||||||
russh-sftp = "2.3.0"
|
russh-sftp = "2.3.0"
|
||||||
|
|
||||||
# WebDAV
|
# WebDAV
|
||||||
|
|||||||
@@ -48,8 +48,8 @@ use rustfs_common::heal_channel::rep_has_active_rules;
|
|||||||
use rustfs_common::metrics::{IlmAction, Metrics};
|
use rustfs_common::metrics::{IlmAction, Metrics};
|
||||||
use rustfs_config::{
|
use rustfs_config::{
|
||||||
DEFAULT_TRANSITION_QUEUE_CAPACITY, DEFAULT_TRANSITION_QUEUE_SEND_TIMEOUT_MS, DEFAULT_TRANSITION_WORKERS_ABSOLUTE_MAX,
|
DEFAULT_TRANSITION_QUEUE_CAPACITY, DEFAULT_TRANSITION_QUEUE_SEND_TIMEOUT_MS, DEFAULT_TRANSITION_WORKERS_ABSOLUTE_MAX,
|
||||||
DEFAULT_TRANSITION_WORKERS_CAP, ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT, ENV_TRANSITION_QUEUE_CAPACITY,
|
DEFAULT_TRANSITION_WORKERS_CAP, ENV_TRANSITION_QUEUE_CAPACITY, ENV_TRANSITION_QUEUE_SEND_TIMEOUT_MS, ENV_TRANSITION_WORKERS,
|
||||||
ENV_TRANSITION_QUEUE_SEND_TIMEOUT_MS, ENV_TRANSITION_WORKERS, ENV_TRANSITION_WORKERS_ABSOLUTE_MAX,
|
ENV_TRANSITION_WORKERS_ABSOLUTE_MAX,
|
||||||
};
|
};
|
||||||
use rustfs_data_usage::TierStats;
|
use rustfs_data_usage::TierStats;
|
||||||
use rustfs_filemeta::{
|
use rustfs_filemeta::{
|
||||||
@@ -145,7 +145,7 @@ fn is_immediate_transition_source(src: &LcEventSrc) -> bool {
|
|||||||
|
|
||||||
#[cfg(any(test, debug_assertions))]
|
#[cfg(any(test, debug_assertions))]
|
||||||
fn should_force_immediate_transition_enqueue_timeout() -> bool {
|
fn should_force_immediate_transition_enqueue_timeout() -> bool {
|
||||||
env::var(ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT)
|
env::var(rustfs_config::ENV_TEST_FORCE_IMMEDIATE_TRANSITION_ENQUEUE_TIMEOUT)
|
||||||
.ok()
|
.ok()
|
||||||
.is_some_and(|value| value == "1")
|
.is_some_and(|value| value == "1")
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -523,7 +523,9 @@ async fn health_check(method: Method, uri: Uri) -> Response {
|
|||||||
} else {
|
} else {
|
||||||
HealthProbe::Liveness
|
HealthProbe::Liveness
|
||||||
};
|
};
|
||||||
let (storage_ready, iam_ready) = collect_dependency_readiness().await;
|
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 health = health_check_state(storage_ready, iam_ready, probe);
|
||||||
|
|
||||||
let builder = Response::builder()
|
let builder = Response::builder()
|
||||||
@@ -537,7 +539,14 @@ async fn health_check(method: Method, uri: Uri) -> Response {
|
|||||||
.duration_since(std::time::UNIX_EPOCH)
|
.duration_since(std::time::UNIX_EPOCH)
|
||||||
.unwrap_or_default()
|
.unwrap_or_default()
|
||||||
.as_secs();
|
.as_secs();
|
||||||
let body_json = build_health_payload(health, storage_ready, iam_ready, "rustfs-console", Some(uptime));
|
let body_json = build_health_payload(
|
||||||
|
health,
|
||||||
|
storage_ready,
|
||||||
|
iam_ready,
|
||||||
|
&readiness_report.degraded_reasons,
|
||||||
|
"rustfs-console",
|
||||||
|
Some(uptime),
|
||||||
|
);
|
||||||
|
|
||||||
// Return a minimal JSON when serialization fails to avoid panic
|
// Return a minimal JSON when serialization fails to avoid panic
|
||||||
let body_str = serde_json::to_string(&body_json).unwrap_or_else(|e| {
|
let body_str = serde_json::to_string(&body_json).unwrap_or_else(|e| {
|
||||||
|
|||||||
@@ -16,7 +16,7 @@ use super::profile::{TriggerProfileCPU, TriggerProfileMemory};
|
|||||||
use crate::admin::router::{AdminOperation, Operation, S3Router};
|
use crate::admin::router::{AdminOperation, Operation, S3Router};
|
||||||
use crate::server::{
|
use crate::server::{
|
||||||
HEALTH_PREFIX, HEALTH_READY_PATH, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH,
|
HEALTH_PREFIX, HEALTH_READY_PATH, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH,
|
||||||
collect_dependency_readiness as collect_runtime_dependency_readiness,
|
collect_dependency_readiness_report as collect_runtime_dependency_readiness_report,
|
||||||
};
|
};
|
||||||
use http::{HeaderMap, HeaderValue};
|
use http::{HeaderMap, HeaderValue};
|
||||||
use hyper::{Method, StatusCode};
|
use hyper::{Method, StatusCode};
|
||||||
@@ -57,9 +57,8 @@ pub(crate) enum HealthProbe {
|
|||||||
Readiness,
|
Readiness,
|
||||||
}
|
}
|
||||||
|
|
||||||
pub(crate) async fn collect_dependency_readiness() -> (bool, bool) {
|
pub(crate) async fn collect_dependency_readiness() -> crate::server::DependencyReadinessReport {
|
||||||
let readiness = collect_runtime_dependency_readiness().await;
|
collect_runtime_dependency_readiness_report().await
|
||||||
(readiness.storage_ready, readiness.iam_ready)
|
|
||||||
}
|
}
|
||||||
|
|
||||||
pub(crate) fn health_check_state(storage_ready: bool, iam_ready: bool, probe: HealthProbe) -> HealthCheckState {
|
pub(crate) fn health_check_state(storage_ready: bool, iam_ready: bool, probe: HealthProbe) -> HealthCheckState {
|
||||||
@@ -99,6 +98,15 @@ pub(crate) fn build_component_details(storage_ready: bool, iam_ready: bool) -> V
|
|||||||
})
|
})
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub(crate) fn build_degraded_reasons(reasons: &[crate::server::ReadinessDegradedReason]) -> Value {
|
||||||
|
Value::Array(
|
||||||
|
reasons
|
||||||
|
.iter()
|
||||||
|
.map(|reason| Value::String(reason.as_str().to_string()))
|
||||||
|
.collect(),
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
pub(crate) fn probe_from_path(path: &str) -> HealthProbe {
|
pub(crate) fn probe_from_path(path: &str) -> HealthProbe {
|
||||||
if path == HEALTH_READY_PATH {
|
if path == HEALTH_READY_PATH {
|
||||||
HealthProbe::Readiness
|
HealthProbe::Readiness
|
||||||
@@ -111,6 +119,7 @@ pub(crate) fn build_health_payload(
|
|||||||
health: HealthCheckState,
|
health: HealthCheckState,
|
||||||
storage_ready: bool,
|
storage_ready: bool,
|
||||||
iam_ready: bool,
|
iam_ready: bool,
|
||||||
|
degraded_reasons: &[crate::server::ReadinessDegradedReason],
|
||||||
service: &str,
|
service: &str,
|
||||||
uptime: Option<u64>,
|
uptime: Option<u64>,
|
||||||
) -> Value {
|
) -> Value {
|
||||||
@@ -128,6 +137,7 @@ pub(crate) fn build_health_payload(
|
|||||||
"timestamp": jiff::Zoned::now().to_string(),
|
"timestamp": jiff::Zoned::now().to_string(),
|
||||||
"version": env!("CARGO_PKG_VERSION"),
|
"version": env!("CARGO_PKG_VERSION"),
|
||||||
"details": build_component_details(storage_ready, iam_ready),
|
"details": build_component_details(storage_ready, iam_ready),
|
||||||
|
"degradedReasons": build_degraded_reasons(degraded_reasons),
|
||||||
});
|
});
|
||||||
|
|
||||||
if let Some(uptime) = uptime {
|
if let Some(uptime) = uptime {
|
||||||
@@ -142,9 +152,10 @@ pub(crate) fn build_health_response(
|
|||||||
probe: HealthProbe,
|
probe: HealthProbe,
|
||||||
storage_ready: bool,
|
storage_ready: bool,
|
||||||
iam_ready: bool,
|
iam_ready: bool,
|
||||||
|
degraded_reasons: &[crate::server::ReadinessDegradedReason],
|
||||||
) -> S3Response<(StatusCode, Body)> {
|
) -> S3Response<(StatusCode, Body)> {
|
||||||
let health = health_check_state(storage_ready, iam_ready, probe);
|
let health = health_check_state(storage_ready, iam_ready, probe);
|
||||||
let health_info = build_health_payload(health, storage_ready, iam_ready, "rustfs-endpoint", None);
|
let health_info = build_health_payload(health, storage_ready, iam_ready, degraded_reasons, "rustfs-endpoint", None);
|
||||||
|
|
||||||
let mut headers = HeaderMap::new();
|
let mut headers = HeaderMap::new();
|
||||||
headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
|
headers.insert(CONTENT_TYPE, HeaderValue::from_static("application/json"));
|
||||||
@@ -177,9 +188,15 @@ impl Operation for HealthCheckHandler {
|
|||||||
}
|
}
|
||||||
|
|
||||||
let probe = probe_from_path(req.uri.path());
|
let probe = probe_from_path(req.uri.path());
|
||||||
let (storage_ready, iam_ready) = collect_dependency_readiness().await;
|
let readiness_report = collect_dependency_readiness().await;
|
||||||
|
|
||||||
Ok(build_health_response(method, probe, storage_ready, iam_ready))
|
Ok(build_health_response(
|
||||||
|
method,
|
||||||
|
probe,
|
||||||
|
readiness_report.readiness.storage_ready,
|
||||||
|
readiness_report.readiness.iam_ready,
|
||||||
|
&readiness_report.degraded_reasons,
|
||||||
|
))
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -243,25 +260,43 @@ mod tests {
|
|||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_build_health_response_readiness_returns_503_when_deps_not_ready() {
|
fn test_build_health_response_readiness_returns_503_when_deps_not_ready() {
|
||||||
let resp = build_health_response(Method::GET, HealthProbe::Readiness, false, true);
|
let resp = build_health_response(
|
||||||
|
Method::GET,
|
||||||
|
HealthProbe::Readiness,
|
||||||
|
false,
|
||||||
|
true,
|
||||||
|
&[crate::server::ReadinessDegradedReason::StorageQuorumUnavailable],
|
||||||
|
);
|
||||||
assert_eq!(resp.output.0, StatusCode::SERVICE_UNAVAILABLE);
|
assert_eq!(resp.output.0, StatusCode::SERVICE_UNAVAILABLE);
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_build_health_response_readiness_returns_200_when_deps_ready() {
|
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, &[]);
|
||||||
assert_eq!(resp.output.0, StatusCode::OK);
|
assert_eq!(resp.output.0, StatusCode::OK);
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_build_health_response_liveness_returns_200_when_deps_not_ready() {
|
fn test_build_health_response_liveness_returns_200_when_deps_not_ready() {
|
||||||
let resp = build_health_response(Method::GET, HealthProbe::Liveness, false, false);
|
let resp = build_health_response(
|
||||||
|
Method::GET,
|
||||||
|
HealthProbe::Liveness,
|
||||||
|
false,
|
||||||
|
false,
|
||||||
|
&[crate::server::ReadinessDegradedReason::StorageAndIamUnavailable],
|
||||||
|
);
|
||||||
assert_eq!(resp.output.0, StatusCode::OK);
|
assert_eq!(resp.output.0, StatusCode::OK);
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_build_health_response_head_returns_empty_body() {
|
fn test_build_health_response_head_returns_empty_body() {
|
||||||
let resp = build_health_response(Method::HEAD, HealthProbe::Readiness, false, false);
|
let resp = build_health_response(
|
||||||
|
Method::HEAD,
|
||||||
|
HealthProbe::Readiness,
|
||||||
|
false,
|
||||||
|
false,
|
||||||
|
&[crate::server::ReadinessDegradedReason::StorageAndIamUnavailable],
|
||||||
|
);
|
||||||
assert_eq!(resp.output.0, StatusCode::SERVICE_UNAVAILABLE);
|
assert_eq!(resp.output.0, StatusCode::SERVICE_UNAVAILABLE);
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -269,7 +304,14 @@ mod tests {
|
|||||||
fn test_build_health_payload_minimal_mode_returns_status_and_ready_only() {
|
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, HealthProbe::Readiness);
|
||||||
with_var(rustfs_config::ENV_HEALTH_MINIMAL_RESPONSE_ENABLE, Some("true"), || {
|
with_var(rustfs_config::ENV_HEALTH_MINIMAL_RESPONSE_ENABLE, Some("true"), || {
|
||||||
let payload = build_health_payload(health, true, false, "rustfs-endpoint", Some(123));
|
let payload = build_health_payload(
|
||||||
|
health,
|
||||||
|
true,
|
||||||
|
false,
|
||||||
|
&[crate::server::ReadinessDegradedReason::IamNotReady],
|
||||||
|
"rustfs-endpoint",
|
||||||
|
Some(123),
|
||||||
|
);
|
||||||
assert_eq!(payload["status"], "degraded");
|
assert_eq!(payload["status"], "degraded");
|
||||||
assert_eq!(payload["ready"], false);
|
assert_eq!(payload["ready"], false);
|
||||||
assert!(payload.get("version").is_none());
|
assert!(payload.get("version").is_none());
|
||||||
@@ -278,4 +320,18 @@ mod tests {
|
|||||||
assert!(payload.get("uptime").is_none());
|
assert!(payload.get("uptime").is_none());
|
||||||
});
|
});
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_build_health_payload_includes_degraded_reasons() {
|
||||||
|
let health = health_check_state(false, false, HealthProbe::Readiness);
|
||||||
|
let payload = build_health_payload(
|
||||||
|
health,
|
||||||
|
false,
|
||||||
|
false,
|
||||||
|
&[crate::server::ReadinessDegradedReason::StorageAndIamUnavailable],
|
||||||
|
"rustfs-endpoint",
|
||||||
|
None,
|
||||||
|
);
|
||||||
|
assert_eq!(payload["degradedReasons"][0], "storage_and_iam_unavailable");
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -13,13 +13,13 @@
|
|||||||
// limitations under the License.
|
// limitations under the License.
|
||||||
|
|
||||||
use crate::admin::console::is_console_path;
|
use crate::admin::console::is_console_path;
|
||||||
use crate::admin::handlers::health::{build_health_payload, collect_dependency_readiness, health_check_state, probe_from_path};
|
use crate::admin::handlers::health::{build_health_payload, health_check_state, probe_from_path};
|
||||||
use crate::error::ApiError;
|
use crate::error::ApiError;
|
||||||
use crate::server::cors;
|
use crate::server::cors;
|
||||||
use crate::server::hybrid::HybridBody;
|
use crate::server::hybrid::HybridBody;
|
||||||
use crate::server::{
|
use crate::server::{
|
||||||
ADMIN_PREFIX, CONSOLE_PREFIX, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_ADMIN_PREFIX, MINIO_ADMIN_V3_PREFIX, RPC_PREFIX,
|
ADMIN_PREFIX, CONSOLE_PREFIX, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_ADMIN_PREFIX, MINIO_ADMIN_V3_PREFIX, RPC_PREFIX,
|
||||||
RUSTFS_ADMIN_PREFIX,
|
RUSTFS_ADMIN_PREFIX, collect_dependency_readiness_report,
|
||||||
};
|
};
|
||||||
use crate::storage::apply_cors_headers;
|
use crate::storage::apply_cors_headers;
|
||||||
use crate::storage::request_context::{RequestContext, extract_request_id_from_headers};
|
use crate::storage::request_context::{RequestContext, extract_request_id_from_headers};
|
||||||
@@ -639,12 +639,21 @@ where
|
|||||||
RestBody: From<Bytes>,
|
RestBody: From<Bytes>,
|
||||||
{
|
{
|
||||||
let probe = probe_from_path(&path);
|
let probe = probe_from_path(&path);
|
||||||
let (storage_ready, iam_ready) = collect_dependency_readiness().await;
|
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 health = health_check_state(storage_ready, iam_ready, probe);
|
||||||
let body = if method == Method::HEAD {
|
let body = if method == Method::HEAD {
|
||||||
Bytes::new()
|
Bytes::new()
|
||||||
} else {
|
} else {
|
||||||
let payload = build_health_payload(health, storage_ready, iam_ready, "rustfs-endpoint", None);
|
let payload = build_health_payload(
|
||||||
|
health,
|
||||||
|
storage_ready,
|
||||||
|
iam_ready,
|
||||||
|
&readiness_report.degraded_reasons,
|
||||||
|
"rustfs-endpoint",
|
||||||
|
None,
|
||||||
|
);
|
||||||
Bytes::from(serde_json::to_vec(&payload).unwrap_or_else(|_| b"{}".to_vec()))
|
Bytes::from(serde_json::to_vec(&payload).unwrap_or_else(|_| b"{}".to_vec()))
|
||||||
};
|
};
|
||||||
|
|
||||||
|
|||||||
@@ -51,8 +51,11 @@ pub(crate) use prefix::{
|
|||||||
MINIO_ADMIN_V3_PREFIX, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, RPC_PREFIX, RUSTFS_ADMIN_PREFIX, TONIC_PREFIX, VERSION,
|
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::DependencyReadiness;
|
||||||
|
pub(crate) use readiness::DependencyReadinessReport;
|
||||||
|
pub(crate) use readiness::ReadinessDegradedReason;
|
||||||
pub(crate) use readiness::ReadinessGateLayer;
|
pub(crate) use readiness::ReadinessGateLayer;
|
||||||
pub(crate) use readiness::collect_dependency_readiness;
|
pub(crate) use readiness::collect_dependency_readiness;
|
||||||
|
pub(crate) use readiness::collect_dependency_readiness_report;
|
||||||
pub use readiness::publish_ready_when_runtime_ready;
|
pub use readiness::publish_ready_when_runtime_ready;
|
||||||
|
|
||||||
#[derive(Clone, Copy, Debug)]
|
#[derive(Clone, Copy, Debug)]
|
||||||
|
|||||||
@@ -18,6 +18,7 @@ use http::{Request as HttpRequest, Response, StatusCode};
|
|||||||
use http_body::Body;
|
use http_body::Body;
|
||||||
use http_body_util::{BodyExt, Full};
|
use http_body_util::{BodyExt, Full};
|
||||||
use hyper::body::Incoming;
|
use hyper::body::Incoming;
|
||||||
|
use metrics::{counter, gauge};
|
||||||
use rustfs_common::GlobalReadiness;
|
use rustfs_common::GlobalReadiness;
|
||||||
use rustfs_ecstore::new_object_layer_fn;
|
use rustfs_ecstore::new_object_layer_fn;
|
||||||
use rustfs_ecstore::store_api::StorageAPI;
|
use rustfs_ecstore::store_api::StorageAPI;
|
||||||
@@ -39,6 +40,8 @@ use tracing::{debug, info};
|
|||||||
|
|
||||||
pub const STARTUP_RUNTIME_READINESS_MAX_WAIT: Duration = Duration::from_secs(30);
|
pub const STARTUP_RUNTIME_READINESS_MAX_WAIT: Duration = Duration::from_secs(30);
|
||||||
pub const STARTUP_RUNTIME_READINESS_POLL_INTERVAL: Duration = Duration::from_secs(1);
|
pub const STARTUP_RUNTIME_READINESS_POLL_INTERVAL: Duration = Duration::from_secs(1);
|
||||||
|
const METRIC_RUNTIME_READINESS_READY: &str = "rustfs_runtime_readiness_ready";
|
||||||
|
const METRIC_RUNTIME_READINESS_DEGRADED_TOTAL: &str = "rustfs_runtime_readiness_degraded_total";
|
||||||
|
|
||||||
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
|
#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)]
|
||||||
pub struct DependencyReadiness {
|
pub struct DependencyReadiness {
|
||||||
@@ -46,6 +49,29 @@ pub struct DependencyReadiness {
|
|||||||
pub iam_ready: bool,
|
pub iam_ready: bool,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
|
||||||
|
pub enum ReadinessDegradedReason {
|
||||||
|
StorageQuorumUnavailable,
|
||||||
|
IamNotReady,
|
||||||
|
StorageAndIamUnavailable,
|
||||||
|
}
|
||||||
|
|
||||||
|
impl ReadinessDegradedReason {
|
||||||
|
pub fn as_str(&self) -> &'static str {
|
||||||
|
match self {
|
||||||
|
ReadinessDegradedReason::StorageQuorumUnavailable => "storage_quorum_unavailable",
|
||||||
|
ReadinessDegradedReason::IamNotReady => "iam_not_ready",
|
||||||
|
ReadinessDegradedReason::StorageAndIamUnavailable => "storage_and_iam_unavailable",
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone, Default, PartialEq, Eq)]
|
||||||
|
pub struct DependencyReadinessReport {
|
||||||
|
pub readiness: DependencyReadiness,
|
||||||
|
pub degraded_reasons: Vec<ReadinessDegradedReason>,
|
||||||
|
}
|
||||||
|
|
||||||
/// ReadinessGateLayer ensures that the system components (IAM, Storage)
|
/// ReadinessGateLayer ensures that the system components (IAM, Storage)
|
||||||
/// are fully initialized before allowing any request to proceed.
|
/// are fully initialized before allowing any request to proceed.
|
||||||
#[derive(Clone)]
|
#[derive(Clone)]
|
||||||
@@ -320,8 +346,29 @@ 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 record_readiness_report(report: &DependencyReadinessReport) {
|
||||||
|
let ready = report.readiness.storage_ready && report.readiness.iam_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);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
pub async fn collect_dependency_readiness() -> DependencyReadiness {
|
pub async fn collect_dependency_readiness() -> DependencyReadiness {
|
||||||
let iam_ready = get_global_iam_sys().is_some_and(|sys| sys.is_ready());
|
collect_dependency_readiness_report().await.readiness
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn collect_dependency_readiness_report() -> DependencyReadinessReport {
|
||||||
|
let iam_ready_raw = get_global_iam_sys().is_some_and(|sys| sys.is_ready());
|
||||||
let storage_ready = if let Some(cached) = load_cached_storage_readiness().await {
|
let storage_ready = if let Some(cached) = load_cached_storage_readiness().await {
|
||||||
cached
|
cached
|
||||||
} else {
|
} else {
|
||||||
@@ -330,20 +377,32 @@ pub async fn collect_dependency_readiness() -> DependencyReadiness {
|
|||||||
computed
|
computed
|
||||||
};
|
};
|
||||||
|
|
||||||
DependencyReadiness {
|
let readiness = DependencyReadiness {
|
||||||
storage_ready,
|
storage_ready,
|
||||||
iam_ready: iam_ready && storage_ready,
|
iam_ready: iam_ready_raw,
|
||||||
}
|
};
|
||||||
|
let report = DependencyReadinessReport {
|
||||||
|
degraded_reasons: degraded_reasons(readiness.storage_ready, iam_ready_raw),
|
||||||
|
readiness,
|
||||||
|
};
|
||||||
|
record_readiness_report(&report);
|
||||||
|
report
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn collect_dependency_readiness_uncached() -> DependencyReadiness {
|
async fn collect_dependency_readiness_uncached() -> DependencyReadiness {
|
||||||
let iam_ready = get_global_iam_sys().is_some_and(|sys| sys.is_ready());
|
let iam_ready_raw = get_global_iam_sys().is_some_and(|sys| sys.is_ready());
|
||||||
let storage_ready = collect_storage_readiness_uncached().await;
|
let storage_ready = collect_storage_readiness_uncached().await;
|
||||||
|
|
||||||
DependencyReadiness {
|
let readiness = DependencyReadiness {
|
||||||
storage_ready,
|
storage_ready,
|
||||||
iam_ready: iam_ready && storage_ready,
|
iam_ready: iam_ready_raw,
|
||||||
}
|
};
|
||||||
|
let report = DependencyReadinessReport {
|
||||||
|
degraded_reasons: degraded_reasons(readiness.storage_ready, iam_ready_raw),
|
||||||
|
readiness,
|
||||||
|
};
|
||||||
|
record_readiness_report(&report);
|
||||||
|
report.readiness
|
||||||
}
|
}
|
||||||
|
|
||||||
async fn collect_storage_readiness_uncached() -> bool {
|
async fn collect_storage_readiness_uncached() -> bool {
|
||||||
|
|||||||
Reference in New Issue
Block a user