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

refactor: decouple server health helpers (#3923)

Browse Source
This commit is contained in:
Zhengchao An
2026-06-26 22:40:22 +08:00
committed by GitHub
parent 15c39db42f
commit a22384708e
7 changed files with 292 additions and 265 deletions
Download Patch File Download Diff File Expand all files Collapse all files
rustfs/src/admin/console.rs
+2 -3
View File
@@ -12,14 +12,13 @@
// See the License for the specific language governing permissions and
// limitations under the License.
use crate::admin::handlers::health::{HealthProbe, build_health_response_parts, collect_probe_readiness};
use crate::admin::runtime_sources::{default_admin_usecase, resolve_oidc_handle};
use crate::admin::storage_api::access::RequestContext;
use crate::license::has_valid_license;
use crate::server::has_path_prefix;
use crate::server::{
CONSOLE_PREFIX, FAVICON_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, HeaderMapCarrier, LICENSE, RUSTFS_ADMIN_PREFIX,
RequestContextLayer, VERSION,
CONSOLE_PREFIX, FAVICON_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, HeaderMapCarrier, HealthProbe, LICENSE, RUSTFS_ADMIN_PREFIX,
RequestContextLayer, VERSION, build_health_response_parts, collect_probe_readiness,
};
use crate::version::build;
use axum::{
rustfs/src/admin/handlers/health.rs
+7 -255
View File
@@ -15,16 +15,19 @@
use super::profile::{TriggerProfileCPU, TriggerProfileMemory};
use crate::admin::router::{AdminOperation, Operation, S3Router};
use crate::server::{
HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_HEALTH_CLUSTER_PATH, MINIO_HEALTH_CLUSTER_READ_PATH, MINIO_HEALTH_READY_PATH,
PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, collect_cluster_read_health_report, collect_cluster_write_health_report,
collect_node_readiness_report,
HEALTH_PREFIX, HEALTH_READY_PATH, PROFILE_CPU_PATH, PROFILE_MEMORY_PATH, build_health_response_parts,
collect_probe_readiness, probe_from_path,
};
#[cfg(test)]
use crate::server::{
HealthPayloadContext, HealthProbe, HealthReadinessSource, MINIO_HEALTH_CLUSTER_PATH, MINIO_HEALTH_CLUSTER_READ_PATH,
MINIO_HEALTH_READY_PATH, build_component_details, build_health_payload, health_check_state, readiness_source_for_probe,
};
use http::{HeaderMap, HeaderValue};
use hyper::{Method, StatusCode};
use matchit::Params;
use s3s::header::CONTENT_TYPE;
use s3s::{Body, S3Request, S3Response, S3Result};
use serde_json::{Value, json};
pub fn register_health_route(r: &mut S3Router<AdminOperation>) -> std::io::Result<()> {
if rustfs_utils::get_env_bool(rustfs_config::ENV_HEALTH_ENDPOINT_ENABLE, rustfs_config::DEFAULT_HEALTH_ENDPOINT_ENABLE) {
@@ -45,257 +48,6 @@ pub fn register_health_route(r: &mut S3Router<AdminOperation>) -> std::io::Resul
/// Health check handler for endpoint monitoring
pub struct HealthCheckHandler {}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct HealthCheckState {
pub(crate) status_code: StatusCode,
pub(crate) status: &'static str,
pub(crate) ready: bool,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum HealthProbe {
Liveness,
Readiness,
ClusterWrite,
ClusterRead,
}
impl HealthProbe {
const fn requires_lock_quorum(self) -> bool {
matches!(self, Self::ClusterWrite | Self::ClusterRead)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
enum HealthReadinessSource {
Node,
ClusterWrite,
ClusterRead,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct HealthResponseParts {
pub(crate) status_code: StatusCode,
pub(crate) payload: Option<Value>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct HealthPayloadContext<'a> {
pub(crate) health: HealthCheckState,
pub(crate) storage_ready: bool,
pub(crate) iam_ready: bool,
pub(crate) lock_quorum_ready: bool,
pub(crate) degraded_reasons: &'a [crate::server::ReadinessDegradedReason],
pub(crate) service: &'a str,
pub(crate) uptime: Option<u64>,
pub(crate) kms_ready: Option<bool>,
pub(crate) include_dependency_details: bool,
}
pub(crate) async fn collect_probe_readiness(probe: HealthProbe) -> Option<crate::server::DependencyReadinessReport> {
match readiness_source_for_probe(probe)? {
HealthReadinessSource::Node => Some(collect_node_readiness_report().await),
HealthReadinessSource::ClusterWrite => Some(collect_cluster_write_health_report().await),
HealthReadinessSource::ClusterRead => Some(collect_cluster_read_health_report().await),
}
}
fn readiness_source_for_probe(probe: HealthProbe) -> Option<HealthReadinessSource> {
match probe {
HealthProbe::Liveness => None,
HealthProbe::Readiness => Some(HealthReadinessSource::Node),
HealthProbe::ClusterWrite => Some(HealthReadinessSource::ClusterWrite),
HealthProbe::ClusterRead => Some(HealthReadinessSource::ClusterRead),
}
}
pub(crate) fn health_check_state(
storage_ready: bool,
iam_ready: bool,
lock_quorum_ready: bool,
probe: HealthProbe,
) -> HealthCheckState {
if probe == HealthProbe::Liveness {
return HealthCheckState {
status_code: StatusCode::OK,
status: "ok",
ready: true,
};
}
let ready = storage_ready && iam_ready && (!probe.requires_lock_quorum() || lock_quorum_ready);
let status = if ready { "ok" } else { "degraded" };
let status_code = if ready {
StatusCode::OK
} else {
StatusCode::SERVICE_UNAVAILABLE
};
HealthCheckState {
status_code,
status,
ready,
}
}
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,
lock_quorum_ready: bool,
kms_ready: Option<bool>,
) -> Value {
let mut details = json!({
"storage": {
"status": if storage_ready { "connected" } else { "disconnected" },
"ready": storage_ready,
},
"iam": {
"status": if iam_ready { "connected" } else { "disconnected" },
"ready": iam_ready,
},
"lock": {
"status": if lock_quorum_ready { "connected" } else { "disconnected" },
"ready": lock_quorum_ready,
}
});
if let Some(kms_ready) = kms_ready {
details["kms"] = json!({
"status": if kms_ready { "connected" } else { "disconnected" },
"ready": kms_ready,
});
}
details
}
pub(crate) fn build_degraded_reasons(reasons: &[crate::server::ReadinessDegradedReason]) -> Value {
Value::Array(
reasons
.iter()
.map(|reason| Value::String(reason.as_str().to_string()))
.collect(),
)
}
pub(crate) fn probe_from_path(path: &str) -> HealthProbe {
match path {
HEALTH_READY_PATH | MINIO_HEALTH_READY_PATH => HealthProbe::Readiness,
MINIO_HEALTH_CLUSTER_PATH => HealthProbe::ClusterWrite,
MINIO_HEALTH_CLUSTER_READ_PATH => HealthProbe::ClusterRead,
_ => HealthProbe::Liveness,
}
}
pub(crate) fn build_health_response_parts(
method: Method,
probe: HealthProbe,
readiness_report: Option<&crate::server::DependencyReadinessReport>,
service: &str,
uptime: Option<u64>,
kms_ready: Option<bool>,
) -> HealthResponseParts {
let (storage_ready, iam_ready, lock_quorum_ready, mut health, mut degraded_reasons, include_dependency_details) =
match (probe, readiness_report) {
(probe @ (HealthProbe::Readiness | HealthProbe::ClusterWrite | HealthProbe::ClusterRead), Some(readiness_report)) => {
let storage_ready = readiness_report.readiness.storage_ready;
let iam_ready = readiness_report.readiness.iam_ready;
let lock_quorum_ready = readiness_report.readiness.lock_quorum_ready;
(
storage_ready,
iam_ready,
lock_quorum_ready,
health_check_state(storage_ready, iam_ready, lock_quorum_ready, probe),
readiness_report.degraded_reasons.clone(),
true,
)
}
(HealthProbe::Readiness | HealthProbe::ClusterWrite | HealthProbe::ClusterRead, None) => (
false,
false,
false,
HealthCheckState {
status_code: StatusCode::SERVICE_UNAVAILABLE,
status: "degraded",
ready: false,
},
vec![crate::server::ReadinessDegradedReason::StorageIamAndLockUnavailable],
true,
),
(HealthProbe::Liveness, _) => {
(false, false, false, health_check_state(false, false, false, probe), Vec::new(), false)
}
};
if probe == HealthProbe::Readiness && matches!(kms_ready, Some(false)) {
health = HealthCheckState {
status_code: StatusCode::SERVICE_UNAVAILABLE,
status: "degraded",
ready: false,
};
if !degraded_reasons.contains(&crate::server::ReadinessDegradedReason::KmsNotReady) {
degraded_reasons.push(crate::server::ReadinessDegradedReason::KmsNotReady);
}
}
let payload = if method == Method::HEAD {
None
} else {
Some(build_health_payload(HealthPayloadContext {
health,
storage_ready,
iam_ready,
lock_quorum_ready,
degraded_reasons: &degraded_reasons,
service,
uptime,
kms_ready,
include_dependency_details,
}))
};
HealthResponseParts {
status_code: health.status_code,
payload,
}
}
pub(crate) fn build_health_payload(ctx: HealthPayloadContext<'_>) -> Value {
if health_minimal_response_enabled() {
return json!({
"status": ctx.health.status,
"ready": ctx.health.ready,
});
}
let mut payload = json!({
"status": ctx.health.status,
"ready": ctx.health.ready,
"service": ctx.service,
"timestamp": jiff::Zoned::now().to_string(),
"version": env!("CARGO_PKG_VERSION"),
});
if ctx.include_dependency_details {
payload["details"] = build_component_details(ctx.storage_ready, ctx.iam_ready, ctx.lock_quorum_ready, ctx.kms_ready);
payload["degradedReasons"] = build_degraded_reasons(ctx.degraded_reasons);
}
if let Some(uptime) = ctx.uptime {
payload["uptime"] = json!(uptime);
}
payload
}
#[async_trait::async_trait]
impl Operation for HealthCheckHandler {
async fn call(&self, req: S3Request<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
rustfs/src/server/health.rs
+272
View File
@@ -0,0 +1,272 @@
// 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.
use super::readiness::{DependencyReadinessReport, ReadinessDegradedReason};
use super::{
HEALTH_READY_PATH, MINIO_HEALTH_CLUSTER_PATH, MINIO_HEALTH_CLUSTER_READ_PATH, MINIO_HEALTH_READY_PATH,
collect_cluster_read_health_report, collect_cluster_write_health_report, collect_node_readiness_report,
};
use http::{Method, StatusCode};
use serde_json::{Value, json};
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct HealthCheckState {
pub(crate) status_code: StatusCode,
pub(crate) status: &'static str,
pub(crate) ready: bool,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum HealthProbe {
Liveness,
Readiness,
ClusterWrite,
ClusterRead,
}
impl HealthProbe {
const fn requires_lock_quorum(self) -> bool {
matches!(self, Self::ClusterWrite | Self::ClusterRead)
}
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) enum HealthReadinessSource {
Node,
ClusterWrite,
ClusterRead,
}
#[derive(Debug, Clone, PartialEq, Eq)]
pub(crate) struct HealthResponseParts {
pub(crate) status_code: StatusCode,
pub(crate) payload: Option<Value>,
}
#[derive(Debug, Clone, Copy, PartialEq, Eq)]
pub(crate) struct HealthPayloadContext<'a> {
pub(crate) health: HealthCheckState,
pub(crate) storage_ready: bool,
pub(crate) iam_ready: bool,
pub(crate) lock_quorum_ready: bool,
pub(crate) degraded_reasons: &'a [ReadinessDegradedReason],
pub(crate) service: &'a str,
pub(crate) uptime: Option<u64>,
pub(crate) kms_ready: Option<bool>,
pub(crate) include_dependency_details: bool,
}
pub(crate) async fn collect_probe_readiness(probe: HealthProbe) -> Option<DependencyReadinessReport> {
match readiness_source_for_probe(probe)? {
HealthReadinessSource::Node => Some(collect_node_readiness_report().await),
HealthReadinessSource::ClusterWrite => Some(collect_cluster_write_health_report().await),
HealthReadinessSource::ClusterRead => Some(collect_cluster_read_health_report().await),
}
}
pub(crate) fn readiness_source_for_probe(probe: HealthProbe) -> Option<HealthReadinessSource> {
match probe {
HealthProbe::Liveness => None,
HealthProbe::Readiness => Some(HealthReadinessSource::Node),
HealthProbe::ClusterWrite => Some(HealthReadinessSource::ClusterWrite),
HealthProbe::ClusterRead => Some(HealthReadinessSource::ClusterRead),
}
}
pub(crate) fn health_check_state(
storage_ready: bool,
iam_ready: bool,
lock_quorum_ready: bool,
probe: HealthProbe,
) -> HealthCheckState {
if probe == HealthProbe::Liveness {
return HealthCheckState {
status_code: StatusCode::OK,
status: "ok",
ready: true,
};
}
let ready = storage_ready && iam_ready && (!probe.requires_lock_quorum() || lock_quorum_ready);
let status = if ready { "ok" } else { "degraded" };
let status_code = if ready {
StatusCode::OK
} else {
StatusCode::SERVICE_UNAVAILABLE
};
HealthCheckState {
status_code,
status,
ready,
}
}
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,
lock_quorum_ready: bool,
kms_ready: Option<bool>,
) -> Value {
let mut details = json!({
"storage": {
"status": if storage_ready { "connected" } else { "disconnected" },
"ready": storage_ready,
},
"iam": {
"status": if iam_ready { "connected" } else { "disconnected" },
"ready": iam_ready,
},
"lock": {
"status": if lock_quorum_ready { "connected" } else { "disconnected" },
"ready": lock_quorum_ready,
}
});
if let Some(kms_ready) = kms_ready {
details["kms"] = json!({
"status": if kms_ready { "connected" } else { "disconnected" },
"ready": kms_ready,
});
}
details
}
pub(crate) fn build_degraded_reasons(reasons: &[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 {
match path {
HEALTH_READY_PATH | MINIO_HEALTH_READY_PATH => HealthProbe::Readiness,
MINIO_HEALTH_CLUSTER_PATH => HealthProbe::ClusterWrite,
MINIO_HEALTH_CLUSTER_READ_PATH => HealthProbe::ClusterRead,
_ => HealthProbe::Liveness,
}
}
pub(crate) fn build_health_response_parts(
method: Method,
probe: HealthProbe,
readiness_report: Option<&DependencyReadinessReport>,
service: &str,
uptime: Option<u64>,
kms_ready: Option<bool>,
) -> HealthResponseParts {
let (storage_ready, iam_ready, lock_quorum_ready, mut health, mut degraded_reasons, include_dependency_details) =
match (probe, readiness_report) {
(probe @ (HealthProbe::Readiness | HealthProbe::ClusterWrite | HealthProbe::ClusterRead), Some(readiness_report)) => {
let storage_ready = readiness_report.readiness.storage_ready;
let iam_ready = readiness_report.readiness.iam_ready;
let lock_quorum_ready = readiness_report.readiness.lock_quorum_ready;
(
storage_ready,
iam_ready,
lock_quorum_ready,
health_check_state(storage_ready, iam_ready, lock_quorum_ready, probe),
readiness_report.degraded_reasons.clone(),
true,
)
}
(HealthProbe::Readiness | HealthProbe::ClusterWrite | HealthProbe::ClusterRead, None) => (
false,
false,
false,
HealthCheckState {
status_code: StatusCode::SERVICE_UNAVAILABLE,
status: "degraded",
ready: false,
},
vec![ReadinessDegradedReason::StorageIamAndLockUnavailable],
true,
),
(HealthProbe::Liveness, _) => {
(false, false, false, health_check_state(false, false, false, probe), Vec::new(), false)
}
};
if probe == HealthProbe::Readiness && matches!(kms_ready, Some(false)) {
health = HealthCheckState {
status_code: StatusCode::SERVICE_UNAVAILABLE,
status: "degraded",
ready: false,
};
if !degraded_reasons.contains(&ReadinessDegradedReason::KmsNotReady) {
degraded_reasons.push(ReadinessDegradedReason::KmsNotReady);
}
}
let payload = if method == Method::HEAD {
None
} else {
Some(build_health_payload(HealthPayloadContext {
health,
storage_ready,
iam_ready,
lock_quorum_ready,
degraded_reasons: &degraded_reasons,
service,
uptime,
kms_ready,
include_dependency_details,
}))
};
HealthResponseParts {
status_code: health.status_code,
payload,
}
}
pub(crate) fn build_health_payload(ctx: HealthPayloadContext<'_>) -> Value {
if health_minimal_response_enabled() {
return json!({
"status": ctx.health.status,
"ready": ctx.health.ready,
});
}
let mut payload = json!({
"status": ctx.health.status,
"ready": ctx.health.ready,
"service": ctx.service,
"timestamp": jiff::Zoned::now().to_string(),
"version": env!("CARGO_PKG_VERSION"),
});
if ctx.include_dependency_details {
payload["details"] = build_component_details(ctx.storage_ready, ctx.iam_ready, ctx.lock_quorum_ready, ctx.kms_ready);
payload["degradedReasons"] = build_degraded_reasons(ctx.degraded_reasons);
}
if let Some(uptime) = ctx.uptime {
payload["uptime"] = json!(uptime);
}
payload
}
rustfs/src/server/layer.rs
+3 -4
View File
@@ -14,16 +14,15 @@
use super::runtime_sources;
use crate::admin::console::is_console_path;
use crate::admin::handlers::health::{HealthProbe, build_health_response_parts, collect_probe_readiness};
use crate::error::ApiError;
use crate::server::RemoteAddr;
use crate::server::cors;
use crate::server::hybrid::HybridBody;
use crate::server::{
ADMIN_PREFIX, CONSOLE_PREFIX, HEALTH_COMPAT_LIVE_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, MINIO_ADMIN_PREFIX,
ADMIN_PREFIX, CONSOLE_PREFIX, HEALTH_COMPAT_LIVE_PATH, HEALTH_PREFIX, HEALTH_READY_PATH, HealthProbe, MINIO_ADMIN_PREFIX,
MINIO_ADMIN_V3_PREFIX, MINIO_HEALTH_CLUSTER_PATH, MINIO_HEALTH_CLUSTER_READ_PATH, MINIO_HEALTH_LIVE_PATH,
MINIO_HEALTH_READY_PATH, RPC_PREFIX, RUSTFS_ADMIN_PREFIX, active_http_requests, has_path_prefix, is_admin_path,
is_table_catalog_path,
MINIO_HEALTH_READY_PATH, RPC_PREFIX, RUSTFS_ADMIN_PREFIX, active_http_requests, build_health_response_parts,
collect_probe_readiness, has_path_prefix, is_admin_path, is_table_catalog_path,
};
use crate::storage_api::server::layer::apply_cors_headers;
use crate::storage_api::server::layer::request_context::{
rustfs/src/server/mod.rs
+7
View File
@@ -16,6 +16,7 @@ mod audit;
mod compress;
pub mod cors;
mod event;
mod health;
mod http;
mod hybrid;
mod layer;
@@ -43,6 +44,12 @@ pub use service_state::wait_for_shutdown;
// Items only used within the library crate (admin handlers, server/http.rs, etc.).
pub(crate) use event::convert_ecstore_object_info;
#[cfg(test)]
pub(crate) use health::{
HealthPayloadContext, HealthReadinessSource, build_component_details, build_health_payload, health_check_state,
readiness_source_for_probe,
};
pub(crate) use health::{HealthProbe, build_health_response_parts, collect_probe_readiness, probe_from_path};
pub(crate) use http::HeaderMapCarrier;
pub(crate) use http::active_http_requests;
pub(crate) use layer::RequestContextLayer;
rustfs/src/storage/concurrency/manager.rs
+1 -1
View File
@@ -667,7 +667,7 @@ impl Default for ConcurrencyManager {
#[allow(unused_imports)]
mod integration_tests {
use super::super::io_schedule::{IoLoadLevel, IoPriority};
use super::super::request_guard::GetObjectGuard;
use super::super::request_guard::{GetObjectGuard, PutObjectGuard};
use super::ConcurrencyManager;
use rustfs_concurrency::{AdmissionState, WorkloadAdmissionSnapshotProvider, WorkloadClass};
use rustfs_config::MI_B;
scripts/layer-dependency-baseline.txt
-2
View File
@@ -25,8 +25,6 @@ dep|rustfs/src/init.rs|infra->interface|crate::admin
dep|rustfs/src/runtime_sources.rs|infra->app|crate::app::context
dep|rustfs/src/server/http.rs|infra->interface|crate::admin
dep|rustfs/src/server/layer.rs|infra->interface|crate::admin::console::is_console_path
dep|rustfs/src/server/layer.rs|infra->interface|crate::admin::handlers::health::HealthProbe
dep|rustfs/src/server/layer.rs|infra->interface|crate::admin::handlers::health::build_health_response_parts
dep|rustfs/src/storage/ecfs_extend.rs|infra->interface|crate::storage::ecfs::ListObjectUnorderedQuery
dep|rustfs/src/storage/ecfs_test.rs|infra->interface|crate::storage::ecfs::FS
dep|rustfs/src/storage/ecfs_test.rs|infra->interface|crate::storage::ecfs::validate_object_lock_configuration_input