mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-31 09:18:28 +00:00
fix(admin): harden health readiness and add health response controls (#2662)
This commit is contained in:
@@ -0,0 +1,28 @@
|
|||||||
|
// Copyright 2024 RustFS Team
|
||||||
|
//
|
||||||
|
// Licensed under the Apache License, Version 2.0 (the "License");
|
||||||
|
// you may not use this file except in compliance with the License.
|
||||||
|
// You may obtain a copy of the License at
|
||||||
|
//
|
||||||
|
// http://www.apache.org/licenses/LICENSE-2.0
|
||||||
|
//
|
||||||
|
// Unless required by applicable law or agreed to in writing, software
|
||||||
|
// distributed under the License is distributed on an "AS IS" BASIS,
|
||||||
|
// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
|
||||||
|
// See the License for the specific language governing permissions and
|
||||||
|
// limitations under the License.
|
||||||
|
|
||||||
|
/// Enable or disable public `/health` and `/health/ready` endpoints.
|
||||||
|
/// When disabled, the routes are not registered and return 404.
|
||||||
|
pub const ENV_HEALTH_ENDPOINT_ENABLE: &str = "RUSTFS_HEALTH_ENDPOINT_ENABLE";
|
||||||
|
pub const DEFAULT_HEALTH_ENDPOINT_ENABLE: bool = true;
|
||||||
|
|
||||||
|
/// Cache TTL for storage readiness runtime-state evaluation (milliseconds).
|
||||||
|
/// This reduces storage-layer pressure when probes are called at high frequency.
|
||||||
|
pub const ENV_HEALTH_READINESS_CACHE_TTL_MS: &str = "RUSTFS_HEALTH_READINESS_CACHE_TTL_MS";
|
||||||
|
pub const DEFAULT_HEALTH_READINESS_CACHE_TTL_MS: u64 = 1000;
|
||||||
|
|
||||||
|
/// Enable minimal health payload mode for GET `/health*` responses.
|
||||||
|
/// 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;
|
||||||
@@ -20,6 +20,7 @@ pub(crate) mod console;
|
|||||||
pub(crate) mod drive;
|
pub(crate) mod drive;
|
||||||
pub(crate) mod env;
|
pub(crate) mod env;
|
||||||
pub(crate) mod heal;
|
pub(crate) mod heal;
|
||||||
|
pub(crate) mod health;
|
||||||
pub(crate) mod object;
|
pub(crate) mod object;
|
||||||
pub(crate) mod oidc;
|
pub(crate) mod oidc;
|
||||||
pub(crate) mod profiler;
|
pub(crate) mod profiler;
|
||||||
|
|||||||
@@ -31,6 +31,8 @@ pub use constants::env::*;
|
|||||||
#[cfg(feature = "constants")]
|
#[cfg(feature = "constants")]
|
||||||
pub use constants::heal::*;
|
pub use constants::heal::*;
|
||||||
#[cfg(feature = "constants")]
|
#[cfg(feature = "constants")]
|
||||||
|
pub use constants::health::*;
|
||||||
|
#[cfg(feature = "constants")]
|
||||||
pub use constants::object::*;
|
pub use constants::object::*;
|
||||||
#[cfg(feature = "constants")]
|
#[cfg(feature = "constants")]
|
||||||
pub use constants::profiler::*;
|
pub use constants::profiler::*;
|
||||||
|
|||||||
+16
-19
@@ -12,9 +12,9 @@
|
|||||||
// See the License for the specific language governing permissions and
|
// See the License for the specific language governing permissions and
|
||||||
// limitations under the License.
|
// limitations under the License.
|
||||||
|
|
||||||
use crate::admin::handlers::health::{HealthProbe, build_component_details, collect_dependency_readiness, health_check_state};
|
use crate::admin::handlers::health::{HealthProbe, build_health_payload, collect_dependency_readiness, health_check_state};
|
||||||
use crate::license::get_license;
|
use crate::license::get_license;
|
||||||
use crate::server::{CONSOLE_PREFIX, FAVICON_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, RUSTFS_ADMIN_PREFIX};
|
use crate::server::{CONSOLE_PREFIX, FAVICON_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, LICENSE, RUSTFS_ADMIN_PREFIX, VERSION};
|
||||||
use crate::version::build;
|
use crate::version::build;
|
||||||
use axum::{
|
use axum::{
|
||||||
Router,
|
Router,
|
||||||
@@ -461,13 +461,17 @@ fn setup_console_middleware_stack(
|
|||||||
) -> Router {
|
) -> Router {
|
||||||
let mut app = Router::new()
|
let mut app = Router::new()
|
||||||
.route(FAVICON_PATH, get(static_handler))
|
.route(FAVICON_PATH, get(static_handler))
|
||||||
.route(&format!("{CONSOLE_PREFIX}/license"), get(license_handler))
|
.route(&format!("{CONSOLE_PREFIX}{LICENSE}"), get(license_handler))
|
||||||
.route(&format!("{CONSOLE_PREFIX}/version"), get(version_handler))
|
.route(&format!("{CONSOLE_PREFIX}{VERSION}"), get(version_handler))
|
||||||
.route(&format!("{CONSOLE_PREFIX}{HEALTH_PREFIX}"), get(health_check).head(health_check))
|
|
||||||
.route(&format!("{CONSOLE_PREFIX}{HEALTH_READY_PATH}"), get(health_check).head(health_check))
|
|
||||||
.nest(CONSOLE_PREFIX, Router::new().fallback_service(get(static_handler)))
|
.nest(CONSOLE_PREFIX, Router::new().fallback_service(get(static_handler)))
|
||||||
.fallback_service(get(static_handler));
|
.fallback_service(get(static_handler));
|
||||||
|
|
||||||
|
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}{HEALTH_READY_PATH}"), get(health_check).head(health_check));
|
||||||
|
}
|
||||||
|
|
||||||
// Add comprehensive middleware layers using tower-http features
|
// Add comprehensive middleware layers using tower-http features
|
||||||
app = app
|
app = app
|
||||||
.layer(CatchPanicLayer::new())
|
.layer(CatchPanicLayer::new())
|
||||||
@@ -511,7 +515,7 @@ async fn health_check(method: Method, uri: Uri) -> Response {
|
|||||||
} else {
|
} else {
|
||||||
HealthProbe::Liveness
|
HealthProbe::Liveness
|
||||||
};
|
};
|
||||||
let (storage_ready, iam_ready) = collect_dependency_readiness();
|
let (storage_ready, iam_ready) = collect_dependency_readiness().await;
|
||||||
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()
|
||||||
@@ -521,18 +525,11 @@ async fn health_check(method: Method, uri: Uri) -> Response {
|
|||||||
match method {
|
match method {
|
||||||
// GET: Returns complete JSON
|
// GET: Returns complete JSON
|
||||||
Method::GET => {
|
Method::GET => {
|
||||||
let body_json = json!({
|
let uptime = std::time::SystemTime::now()
|
||||||
"status": health.status,
|
.duration_since(std::time::UNIX_EPOCH)
|
||||||
"ready": health.ready,
|
.unwrap_or_default()
|
||||||
"service": "rustfs-console",
|
.as_secs();
|
||||||
"timestamp": jiff::Zoned::now().to_string(),
|
let body_json = build_health_payload(health, storage_ready, iam_ready, "rustfs-console", Some(uptime));
|
||||||
"version": env!("CARGO_PKG_VERSION"),
|
|
||||||
"details": build_component_details(storage_ready, iam_ready),
|
|
||||||
"uptime": std::time::SystemTime::now()
|
|
||||||
.duration_since(std::time::UNIX_EPOCH)
|
|
||||||
.unwrap_or_default()
|
|
||||||
.as_secs()
|
|
||||||
});
|
|
||||||
|
|
||||||
// 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| {
|
||||||
|
|||||||
@@ -24,11 +24,15 @@ use s3s::{Body, S3Request, S3Response, S3Result};
|
|||||||
use serde_json::{Value, json};
|
use serde_json::{Value, json};
|
||||||
|
|
||||||
pub fn register_health_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
|
pub fn register_health_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
|
||||||
// Health check endpoint for monitoring and orchestration
|
if rustfs_utils::get_env_bool(rustfs_config::ENV_HEALTH_ENDPOINT_ENABLE, rustfs_config::DEFAULT_HEALTH_ENDPOINT_ENABLE) {
|
||||||
r.insert(Method::GET, HEALTH_PREFIX, AdminOperation(&HealthCheckHandler {}))?;
|
// Health check endpoint for monitoring and orchestration
|
||||||
r.insert(Method::HEAD, HEALTH_PREFIX, AdminOperation(&HealthCheckHandler {}))?;
|
r.insert(Method::GET, HEALTH_PREFIX, AdminOperation(&HealthCheckHandler {}))?;
|
||||||
r.insert(Method::GET, HEALTH_READY_PATH, AdminOperation(&HealthCheckHandler {}))?;
|
r.insert(Method::HEAD, HEALTH_PREFIX, AdminOperation(&HealthCheckHandler {}))?;
|
||||||
r.insert(Method::HEAD, HEALTH_READY_PATH, AdminOperation(&HealthCheckHandler {}))?;
|
r.insert(Method::GET, HEALTH_READY_PATH, AdminOperation(&HealthCheckHandler {}))?;
|
||||||
|
r.insert(Method::HEAD, HEALTH_READY_PATH, AdminOperation(&HealthCheckHandler {}))?;
|
||||||
|
}
|
||||||
|
|
||||||
|
// Profiling routes are controlled separately and must not be affected by health endpoint toggles.
|
||||||
r.insert(Method::GET, PROFILE_CPU_PATH, AdminOperation(&TriggerProfileCPU {}))?;
|
r.insert(Method::GET, PROFILE_CPU_PATH, AdminOperation(&TriggerProfileCPU {}))?;
|
||||||
r.insert(Method::GET, PROFILE_MEMORY_PATH, AdminOperation(&TriggerProfileMemory {}))?;
|
r.insert(Method::GET, PROFILE_MEMORY_PATH, AdminOperation(&TriggerProfileMemory {}))?;
|
||||||
|
|
||||||
@@ -51,9 +55,9 @@ pub(crate) enum HealthProbe {
|
|||||||
Readiness,
|
Readiness,
|
||||||
}
|
}
|
||||||
|
|
||||||
pub(crate) fn collect_dependency_readiness() -> (bool, bool) {
|
pub(crate) async fn collect_dependency_readiness() -> (bool, bool) {
|
||||||
let usecase = DefaultAdminUsecase::from_global();
|
let usecase = DefaultAdminUsecase::from_global();
|
||||||
let readiness = usecase.execute_collect_dependency_readiness();
|
let readiness = usecase.execute_collect_dependency_readiness().await;
|
||||||
(readiness.storage_ready, readiness.iam_ready)
|
(readiness.storage_ready, readiness.iam_ready)
|
||||||
}
|
}
|
||||||
|
|
||||||
@@ -74,6 +78,13 @@ pub(crate) fn health_check_state(storage_ready: bool, iam_ready: bool, probe: He
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub(crate) fn health_minimal_response_enabled() -> bool {
|
||||||
|
rustfs_utils::get_env_bool(
|
||||||
|
rustfs_config::ENV_HEALTH_MINIMAL_RESPONSE_ENABLE,
|
||||||
|
rustfs_config::DEFAULT_HEALTH_MINIMAL_RESPONSE_ENABLE,
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
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) -> Value {
|
||||||
json!({
|
json!({
|
||||||
"storage": {
|
"storage": {
|
||||||
@@ -95,6 +106,36 @@ pub(crate) fn probe_from_path(path: &str) -> HealthProbe {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
pub(crate) fn build_health_payload(
|
||||||
|
health: HealthCheckState,
|
||||||
|
storage_ready: bool,
|
||||||
|
iam_ready: bool,
|
||||||
|
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),
|
||||||
|
});
|
||||||
|
|
||||||
|
if let Some(uptime) = uptime {
|
||||||
|
payload["uptime"] = json!(uptime);
|
||||||
|
}
|
||||||
|
|
||||||
|
payload
|
||||||
|
}
|
||||||
|
|
||||||
pub(crate) fn build_health_response(
|
pub(crate) fn build_health_response(
|
||||||
method: Method,
|
method: Method,
|
||||||
probe: HealthProbe,
|
probe: HealthProbe,
|
||||||
@@ -102,14 +143,7 @@ pub(crate) fn build_health_response(
|
|||||||
iam_ready: bool,
|
iam_ready: bool,
|
||||||
) -> 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 = json!({
|
let health_info = build_health_payload(health, storage_ready, iam_ready, "rustfs-endpoint", None);
|
||||||
"status": health.status,
|
|
||||||
"ready": health.ready,
|
|
||||||
"service": "rustfs-endpoint",
|
|
||||||
"timestamp": jiff::Zoned::now().to_string(),
|
|
||||||
"version": env!("CARGO_PKG_VERSION"),
|
|
||||||
"details": build_component_details(storage_ready, iam_ready)
|
|
||||||
});
|
|
||||||
|
|
||||||
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"));
|
||||||
@@ -131,7 +165,7 @@ impl Operation for HealthCheckHandler {
|
|||||||
let method = req.method;
|
let method = req.method;
|
||||||
|
|
||||||
// Only GET and HEAD are allowed
|
// Only GET and HEAD are allowed
|
||||||
if method != http::Method::GET && method != http::Method::HEAD {
|
if method != Method::GET && method != Method::HEAD {
|
||||||
// 405 Method Not Allowed
|
// 405 Method Not Allowed
|
||||||
let mut headers = HeaderMap::new();
|
let mut headers = HeaderMap::new();
|
||||||
headers.insert(http::header::ALLOW, HeaderValue::from_static("GET, HEAD"));
|
headers.insert(http::header::ALLOW, HeaderValue::from_static("GET, HEAD"));
|
||||||
@@ -142,7 +176,7 @@ 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();
|
let (storage_ready, iam_ready) = collect_dependency_readiness().await;
|
||||||
|
|
||||||
Ok(build_health_response(method, probe, storage_ready, iam_ready))
|
Ok(build_health_response(method, probe, storage_ready, iam_ready))
|
||||||
}
|
}
|
||||||
@@ -151,6 +185,7 @@ impl Operation for HealthCheckHandler {
|
|||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
use super::*;
|
use super::*;
|
||||||
|
use temp_env::with_var;
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_readiness_state_ready() {
|
fn test_readiness_state_ready() {
|
||||||
@@ -228,4 +263,18 @@ mod tests {
|
|||||||
let resp = build_health_response(Method::HEAD, HealthProbe::Readiness, false, false);
|
let resp = build_health_response(Method::HEAD, HealthProbe::Readiness, false, false);
|
||||||
assert_eq!(resp.output.0, StatusCode::SERVICE_UNAVAILABLE);
|
assert_eq!(resp.output.0, StatusCode::SERVICE_UNAVAILABLE);
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_build_health_payload_minimal_mode_returns_status_and_ready_only() {
|
||||||
|
let health = health_check_state(true, false, HealthProbe::Readiness);
|
||||||
|
with_var(rustfs_config::ENV_HEALTH_MINIMAL_RESPONSE_ENABLE, Some("true"), || {
|
||||||
|
let payload = build_health_payload(health, true, false, "rustfs-endpoint", Some(123));
|
||||||
|
assert_eq!(payload["status"], "degraded");
|
||||||
|
assert_eq!(payload["ready"], false);
|
||||||
|
assert!(payload.get("version").is_none());
|
||||||
|
assert!(payload.get("details").is_none());
|
||||||
|
assert!(payload.get("service").is_none());
|
||||||
|
assert!(payload.get("uptime").is_none());
|
||||||
|
});
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -21,6 +21,7 @@ use crate::admin::{
|
|||||||
};
|
};
|
||||||
use crate::server::{ADMIN_PREFIX, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_ADMIN_PREFIX, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH};
|
use crate::server::{ADMIN_PREFIX, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_ADMIN_PREFIX, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH};
|
||||||
use hyper::Method;
|
use hyper::Method;
|
||||||
|
use temp_env::with_var;
|
||||||
|
|
||||||
fn admin_path(path: &str) -> String {
|
fn admin_path(path: &str) -> String {
|
||||||
format!("{}{}", ADMIN_PREFIX, path)
|
format!("{}{}", ADMIN_PREFIX, path)
|
||||||
@@ -224,6 +225,39 @@ fn test_admin_alias_paths_match_existing_admin_routes() {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn test_health_routes_not_registered_when_disabled_by_env() {
|
||||||
|
with_var(rustfs_config::ENV_HEALTH_ENDPOINT_ENABLE, Some("false"), || {
|
||||||
|
let mut router: S3Router<AdminOperation> = S3Router::new(false);
|
||||||
|
health::register_health_route(&mut router).expect("register health route");
|
||||||
|
|
||||||
|
assert!(
|
||||||
|
!router.contains_route(Method::GET, HEALTH_PREFIX),
|
||||||
|
"GET /health must not be registered when health endpoint is disabled"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
!router.contains_route(Method::HEAD, HEALTH_PREFIX),
|
||||||
|
"HEAD /health must not be registered when health endpoint is disabled"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
!router.contains_route(Method::GET, HEALTH_READY_PATH),
|
||||||
|
"GET /health/ready must not be registered when health endpoint is disabled"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
!router.contains_route(Method::HEAD, HEALTH_READY_PATH),
|
||||||
|
"HEAD /health/ready must not be registered when health endpoint is disabled"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
router.contains_route(Method::GET, PROFILE_CPU_PATH),
|
||||||
|
"GET /profile/cpu must stay registered when health endpoint is disabled"
|
||||||
|
);
|
||||||
|
assert!(
|
||||||
|
router.contains_route(Method::GET, PROFILE_MEMORY_PATH),
|
||||||
|
"GET /profile/memory must stay registered when health endpoint is disabled"
|
||||||
|
);
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
#[test]
|
#[test]
|
||||||
fn test_phase5_admin_info_contract() {
|
fn test_phase5_admin_info_contract() {
|
||||||
let system_src = include_str!("handlers/system.rs");
|
let system_src = include_str!("handlers/system.rs");
|
||||||
|
|||||||
@@ -24,9 +24,12 @@ use rustfs_ecstore::endpoints::EndpointServerPools;
|
|||||||
use rustfs_ecstore::new_object_layer_fn;
|
use rustfs_ecstore::new_object_layer_fn;
|
||||||
use rustfs_ecstore::pools::{PoolStatus, get_total_usable_capacity, get_total_usable_capacity_free};
|
use rustfs_ecstore::pools::{PoolStatus, get_total_usable_capacity, get_total_usable_capacity_free};
|
||||||
use rustfs_ecstore::store_api::StorageAPI;
|
use rustfs_ecstore::store_api::StorageAPI;
|
||||||
use rustfs_madmin::{InfoMessage, StorageInfo};
|
use rustfs_madmin::{Disk, InfoMessage, StorageInfo};
|
||||||
use s3s::S3ErrorCode;
|
use s3s::S3ErrorCode;
|
||||||
use std::sync::Arc;
|
use std::collections::{HashMap, HashSet};
|
||||||
|
use std::sync::{Arc, OnceLock};
|
||||||
|
use std::time::{Duration, Instant};
|
||||||
|
use tokio::sync::Mutex;
|
||||||
use tracing::{debug, error, info, warn};
|
use tracing::{debug, error, info, warn};
|
||||||
|
|
||||||
pub type AdminUsecaseResult<T> = Result<T, ApiError>;
|
pub type AdminUsecaseResult<T> = Result<T, ApiError>;
|
||||||
@@ -63,7 +66,17 @@ pub struct DefaultAdminUsecase {
|
|||||||
context: Option<Arc<AppContext>>,
|
context: Option<Arc<AppContext>>,
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, Clone, Copy)]
|
||||||
|
struct StorageReadinessCacheEntry {
|
||||||
|
captured_at: Instant,
|
||||||
|
storage_ready: bool,
|
||||||
|
}
|
||||||
|
|
||||||
impl DefaultAdminUsecase {
|
impl DefaultAdminUsecase {
|
||||||
|
const DISK_STATE_OK: &'static str = "ok";
|
||||||
|
const DISK_STATE_UNFORMATTED: &'static str = "unformatted";
|
||||||
|
const RUNTIME_STATE_RETURNING: &'static str = "returning";
|
||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
pub fn without_context() -> Self {
|
pub fn without_context() -> Self {
|
||||||
Self { context: None }
|
Self { context: None }
|
||||||
@@ -233,12 +246,159 @@ impl DefaultAdminUsecase {
|
|||||||
store.status(idx).await.map_err(ApiError::from)
|
store.status(idx).await.map_err(ApiError::from)
|
||||||
}
|
}
|
||||||
|
|
||||||
pub fn execute_collect_dependency_readiness(&self) -> DependencyReadiness {
|
fn disk_is_online_for_readiness(disk: &Disk) -> bool {
|
||||||
|
let state_is_acceptable = disk.state.eq_ignore_ascii_case(Self::DISK_STATE_OK)
|
||||||
|
|| disk.state.eq_ignore_ascii_case(rustfs_madmin::ITEM_ONLINE)
|
||||||
|
|| disk.state.eq_ignore_ascii_case(Self::DISK_STATE_UNFORMATTED);
|
||||||
|
|
||||||
|
if let Some(runtime_state) = disk.runtime_state.as_deref() {
|
||||||
|
let runtime_state_is_acceptable = runtime_state.eq_ignore_ascii_case(rustfs_madmin::ITEM_ONLINE)
|
||||||
|
|| runtime_state.eq_ignore_ascii_case(Self::RUNTIME_STATE_RETURNING);
|
||||||
|
return runtime_state_is_acceptable && state_is_acceptable;
|
||||||
|
}
|
||||||
|
|
||||||
|
state_is_acceptable
|
||||||
|
}
|
||||||
|
|
||||||
|
fn health_readiness_cache_ttl() -> Duration {
|
||||||
|
Duration::from_millis(rustfs_utils::get_env_u64(
|
||||||
|
rustfs_config::ENV_HEALTH_READINESS_CACHE_TTL_MS,
|
||||||
|
rustfs_config::DEFAULT_HEALTH_READINESS_CACHE_TTL_MS,
|
||||||
|
))
|
||||||
|
}
|
||||||
|
|
||||||
|
fn storage_readiness_cache() -> &'static Mutex<Option<StorageReadinessCacheEntry>> {
|
||||||
|
static CACHE: OnceLock<Mutex<Option<StorageReadinessCacheEntry>>> = OnceLock::new();
|
||||||
|
CACHE.get_or_init(|| Mutex::new(None))
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn load_cached_storage_readiness() -> Option<bool> {
|
||||||
|
let ttl = Self::health_readiness_cache_ttl();
|
||||||
|
if ttl.is_zero() {
|
||||||
|
return None;
|
||||||
|
}
|
||||||
|
|
||||||
|
let cache = Self::storage_readiness_cache().lock().await;
|
||||||
|
let entry = cache.as_ref()?;
|
||||||
|
if entry.captured_at.elapsed() <= ttl {
|
||||||
|
return Some(entry.storage_ready);
|
||||||
|
}
|
||||||
|
|
||||||
|
None
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn update_storage_readiness_cache(storage_ready: bool) {
|
||||||
|
if Self::health_readiness_cache_ttl().is_zero() {
|
||||||
|
return;
|
||||||
|
}
|
||||||
|
|
||||||
|
let mut cache = Self::storage_readiness_cache().lock().await;
|
||||||
|
*cache = Some(StorageReadinessCacheEntry {
|
||||||
|
captured_at: Instant::now(),
|
||||||
|
storage_ready,
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
|
fn pool_write_quorum(info: &StorageInfo, pool_idx: usize, set_drive_count: usize) -> usize {
|
||||||
|
if set_drive_count == 0 {
|
||||||
|
return 1;
|
||||||
|
}
|
||||||
|
|
||||||
|
let data_drives = info
|
||||||
|
.backend
|
||||||
|
.standard_sc_data
|
||||||
|
.get(pool_idx)
|
||||||
|
.copied()
|
||||||
|
.filter(|count| *count > 0)
|
||||||
|
.unwrap_or_else(|| (set_drive_count / 2).max(1));
|
||||||
|
|
||||||
|
let parity_drives = if let Some(drives_per_set) = info.backend.drives_per_set.get(pool_idx).copied() {
|
||||||
|
drives_per_set.saturating_sub(data_drives)
|
||||||
|
} else if let Some(parity) = info.backend.standard_sc_parities.get(pool_idx).copied() {
|
||||||
|
parity
|
||||||
|
} else if let Some(parity) = info.backend.standard_sc_parity {
|
||||||
|
parity
|
||||||
|
} else {
|
||||||
|
set_drive_count.saturating_sub(data_drives)
|
||||||
|
};
|
||||||
|
|
||||||
|
let mut write_quorum = data_drives;
|
||||||
|
if data_drives == parity_drives {
|
||||||
|
write_quorum += 1;
|
||||||
|
}
|
||||||
|
write_quorum.max(1)
|
||||||
|
}
|
||||||
|
|
||||||
|
fn storage_ready_from_runtime_state(info: &StorageInfo) -> bool {
|
||||||
|
if info.disks.is_empty() {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
let mut total_online = 0usize;
|
||||||
|
let mut set_online_counts: HashMap<(usize, usize), usize> = HashMap::new();
|
||||||
|
let mut set_drive_counts: HashMap<(usize, usize), usize> = HashMap::new();
|
||||||
|
let mut seen_disks: HashSet<(String, String, i32, i32, i32)> = HashSet::new();
|
||||||
|
|
||||||
|
for disk in &info.disks {
|
||||||
|
if disk.pool_index < 0 || disk.set_index < 0 {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
let dedup_key = (
|
||||||
|
disk.endpoint.clone(),
|
||||||
|
disk.drive_path.clone(),
|
||||||
|
disk.pool_index,
|
||||||
|
disk.set_index,
|
||||||
|
disk.disk_index,
|
||||||
|
);
|
||||||
|
if !seen_disks.insert(dedup_key) {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
|
||||||
|
let pool_idx = disk.pool_index as usize;
|
||||||
|
let set_idx = disk.set_index as usize;
|
||||||
|
let key = (pool_idx, set_idx);
|
||||||
|
*set_drive_counts.entry(key).or_default() += 1;
|
||||||
|
|
||||||
|
if Self::disk_is_online_for_readiness(disk) {
|
||||||
|
total_online += 1;
|
||||||
|
*set_online_counts.entry(key).or_default() += 1;
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
if total_online == 0 {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
if set_drive_counts.is_empty() {
|
||||||
|
return false;
|
||||||
|
}
|
||||||
|
|
||||||
|
set_drive_counts.into_iter().all(|((pool_idx, set_idx), set_drive_count)| {
|
||||||
|
let online = set_online_counts.get(&(pool_idx, set_idx)).copied().unwrap_or_default();
|
||||||
|
let write_quorum = Self::pool_write_quorum(info, pool_idx, set_drive_count);
|
||||||
|
online >= write_quorum
|
||||||
|
})
|
||||||
|
}
|
||||||
|
|
||||||
|
pub async fn execute_collect_dependency_readiness(&self) -> DependencyReadiness {
|
||||||
let iam_ready = self.context.as_ref().map(|context| context.iam().is_ready()).unwrap_or(false);
|
let iam_ready = self.context.as_ref().map(|context| context.iam().is_ready()).unwrap_or(false);
|
||||||
|
let storage_ready = if let Some(cached) = Self::load_cached_storage_readiness().await {
|
||||||
|
cached
|
||||||
|
} else {
|
||||||
|
let computed = if let Some(store) = new_object_layer_fn() {
|
||||||
|
let storage_info = store.storage_info().await;
|
||||||
|
Self::storage_ready_from_runtime_state(&storage_info)
|
||||||
|
} else {
|
||||||
|
false
|
||||||
|
};
|
||||||
|
Self::update_storage_readiness_cache(computed).await;
|
||||||
|
computed
|
||||||
|
};
|
||||||
|
|
||||||
DependencyReadiness {
|
DependencyReadiness {
|
||||||
storage_ready: new_object_layer_fn().is_some(),
|
storage_ready,
|
||||||
iam_ready,
|
iam_ready: iam_ready && storage_ready,
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -263,12 +423,127 @@ mod tests {
|
|||||||
assert_eq!(err.code, S3ErrorCode::InternalError);
|
assert_eq!(err.code, S3ErrorCode::InternalError);
|
||||||
}
|
}
|
||||||
|
|
||||||
#[test]
|
#[tokio::test]
|
||||||
fn execute_collect_dependency_readiness_returns_state_flags() {
|
async fn execute_collect_dependency_readiness_returns_state_flags() {
|
||||||
let usecase = DefaultAdminUsecase::without_context();
|
let usecase = DefaultAdminUsecase::without_context();
|
||||||
|
|
||||||
let readiness = usecase.execute_collect_dependency_readiness();
|
let readiness = usecase.execute_collect_dependency_readiness().await;
|
||||||
let _ = readiness.storage_ready;
|
let _ = readiness.storage_ready;
|
||||||
let _ = readiness.iam_ready;
|
let _ = readiness.iam_ready;
|
||||||
}
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn storage_ready_from_runtime_state_returns_false_when_all_disks_faulty() {
|
||||||
|
let info = StorageInfo {
|
||||||
|
backend: rustfs_madmin::BackendInfo {
|
||||||
|
standard_sc_data: vec![1],
|
||||||
|
drives_per_set: vec![1],
|
||||||
|
..Default::default()
|
||||||
|
},
|
||||||
|
disks: vec![Disk {
|
||||||
|
pool_index: 0,
|
||||||
|
set_index: 0,
|
||||||
|
state: "offline".to_string(),
|
||||||
|
runtime_state: Some("offline".to_string()),
|
||||||
|
..Default::default()
|
||||||
|
}],
|
||||||
|
};
|
||||||
|
|
||||||
|
assert!(!DefaultAdminUsecase::storage_ready_from_runtime_state(&info));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn storage_ready_from_runtime_state_returns_true_when_set_meets_write_quorum() {
|
||||||
|
let info = StorageInfo {
|
||||||
|
backend: rustfs_madmin::BackendInfo {
|
||||||
|
standard_sc_data: vec![1],
|
||||||
|
drives_per_set: vec![1],
|
||||||
|
..Default::default()
|
||||||
|
},
|
||||||
|
disks: vec![Disk {
|
||||||
|
pool_index: 0,
|
||||||
|
set_index: 0,
|
||||||
|
state: "ok".to_string(),
|
||||||
|
runtime_state: Some("online".to_string()),
|
||||||
|
..Default::default()
|
||||||
|
}],
|
||||||
|
};
|
||||||
|
|
||||||
|
assert!(DefaultAdminUsecase::storage_ready_from_runtime_state(&info));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn storage_ready_from_runtime_state_deduplicates_duplicate_disk_rows() {
|
||||||
|
let duplicate_disk = Disk {
|
||||||
|
endpoint: "127.0.0.1:9000".to_string(),
|
||||||
|
drive_path: "/data0".to_string(),
|
||||||
|
pool_index: 0,
|
||||||
|
set_index: 0,
|
||||||
|
disk_index: 0,
|
||||||
|
state: "ok".to_string(),
|
||||||
|
runtime_state: Some("online".to_string()),
|
||||||
|
..Default::default()
|
||||||
|
};
|
||||||
|
let info = StorageInfo {
|
||||||
|
backend: rustfs_madmin::BackendInfo {
|
||||||
|
standard_sc_data: vec![2],
|
||||||
|
drives_per_set: vec![4],
|
||||||
|
..Default::default()
|
||||||
|
},
|
||||||
|
disks: vec![duplicate_disk.clone(), duplicate_disk],
|
||||||
|
};
|
||||||
|
|
||||||
|
assert!(
|
||||||
|
!DefaultAdminUsecase::storage_ready_from_runtime_state(&info),
|
||||||
|
"duplicate rows must not satisfy write quorum"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn disk_online_for_readiness_requires_runtime_and_state_both_acceptable() {
|
||||||
|
let disk = Disk {
|
||||||
|
state: "disk io error".to_string(),
|
||||||
|
runtime_state: Some("online".to_string()),
|
||||||
|
..Default::default()
|
||||||
|
};
|
||||||
|
assert!(!DefaultAdminUsecase::disk_is_online_for_readiness(&disk));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn storage_ready_from_runtime_state_requires_all_sets_meet_quorum() {
|
||||||
|
let info = StorageInfo {
|
||||||
|
backend: rustfs_madmin::BackendInfo {
|
||||||
|
standard_sc_data: vec![1],
|
||||||
|
drives_per_set: vec![2],
|
||||||
|
..Default::default()
|
||||||
|
},
|
||||||
|
disks: vec![
|
||||||
|
Disk {
|
||||||
|
endpoint: "127.0.0.1:9000".to_string(),
|
||||||
|
drive_path: "/set0d0".to_string(),
|
||||||
|
pool_index: 0,
|
||||||
|
set_index: 0,
|
||||||
|
disk_index: 0,
|
||||||
|
state: "ok".to_string(),
|
||||||
|
runtime_state: Some("online".to_string()),
|
||||||
|
..Default::default()
|
||||||
|
},
|
||||||
|
Disk {
|
||||||
|
endpoint: "127.0.0.1:9000".to_string(),
|
||||||
|
drive_path: "/set1d0".to_string(),
|
||||||
|
pool_index: 0,
|
||||||
|
set_index: 1,
|
||||||
|
disk_index: 0,
|
||||||
|
state: "offline".to_string(),
|
||||||
|
runtime_state: Some("offline".to_string()),
|
||||||
|
..Default::default()
|
||||||
|
},
|
||||||
|
],
|
||||||
|
};
|
||||||
|
|
||||||
|
assert!(
|
||||||
|
!DefaultAdminUsecase::storage_ready_from_runtime_state(&info),
|
||||||
|
"if any set fails write quorum, readiness must be false"
|
||||||
|
);
|
||||||
|
}
|
||||||
}
|
}
|
||||||
|
|||||||
@@ -39,8 +39,8 @@ pub use service_state::wait_for_shutdown;
|
|||||||
|
|
||||||
// Items only used within the library crate (admin handlers, server/http.rs, etc.).
|
// Items only used within the library crate (admin handlers, server/http.rs, etc.).
|
||||||
pub(crate) use prefix::{
|
pub(crate) use prefix::{
|
||||||
ADMIN_PREFIX, CONSOLE_PREFIX, FAVICON_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_ADMIN_PREFIX, MINIO_ADMIN_V3_PREFIX,
|
ADMIN_PREFIX, CONSOLE_PREFIX, FAVICON_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, LICENSE, MINIO_ADMIN_PREFIX,
|
||||||
PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, RPC_PREFIX, RUSTFS_ADMIN_PREFIX, TONIC_PREFIX,
|
MINIO_ADMIN_V3_PREFIX, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, RPC_PREFIX, RUSTFS_ADMIN_PREFIX, TONIC_PREFIX, VERSION,
|
||||||
};
|
};
|
||||||
pub(crate) use readiness::ReadinessGateLayer;
|
pub(crate) use readiness::ReadinessGateLayer;
|
||||||
|
|
||||||
|
|||||||
@@ -61,6 +61,12 @@ pub(crate) const RPC_PREFIX: &str = "/rustfs/rpc";
|
|||||||
/// For example, the full gRPC method path would be "/node_service.NodeService/MethodName".
|
/// For example, the full gRPC method path would be "/node_service.NodeService/MethodName".
|
||||||
pub(crate) const TONIC_PREFIX: &str = "/node_service.NodeService";
|
pub(crate) const TONIC_PREFIX: &str = "/node_service.NodeService";
|
||||||
|
|
||||||
|
/// version information path for RustFS server. This path is used to access version information about the RustFS server.
|
||||||
|
pub(crate) const VERSION: &str = "/version";
|
||||||
|
|
||||||
|
/// license information path for RustFS server. This path is used to access license information about the RustFS server.
|
||||||
|
pub(crate) const LICENSE: &str = "/license";
|
||||||
|
|
||||||
/// LOGO art for RustFS server.
|
/// LOGO art for RustFS server.
|
||||||
pub const LOGO: &str = r#"
|
pub const LOGO: &str = r#"
|
||||||
|
|
||||||
|
|||||||
Reference in New Issue
Block a user