diff --git a/rustfs/src/admin/console.rs b/rustfs/src/admin/console.rs index ed739c079..afb032a58 100644 --- a/rustfs/src/admin/console.rs +++ b/rustfs/src/admin/console.rs @@ -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::{ diff --git a/rustfs/src/admin/handlers/health.rs b/rustfs/src/admin/handlers/health.rs index cfd8c569e..207b3257e 100644 --- a/rustfs/src/admin/handlers/health.rs +++ b/rustfs/src/admin/handlers/health.rs @@ -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) -> 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) -> 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, -} - -#[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, - pub(crate) kms_ready: Option, - pub(crate) include_dependency_details: bool, -} - -pub(crate) async fn collect_probe_readiness(probe: HealthProbe) -> Option { - 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 { - 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, -) -> 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, - kms_ready: Option, -) -> 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: °raded_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, _params: Params<'_, '_>) -> S3Result> { diff --git a/rustfs/src/server/health.rs b/rustfs/src/server/health.rs new file mode 100644 index 000000000..72e451bbb --- /dev/null +++ b/rustfs/src/server/health.rs @@ -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, +} + +#[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, + pub(crate) kms_ready: Option, + pub(crate) include_dependency_details: bool, +} + +pub(crate) async fn collect_probe_readiness(probe: HealthProbe) -> Option { + 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 { + 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, +) -> 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, + kms_ready: Option, +) -> 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: °raded_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 +} diff --git a/rustfs/src/server/layer.rs b/rustfs/src/server/layer.rs index a5573d1a3..405cda3e3 100644 --- a/rustfs/src/server/layer.rs +++ b/rustfs/src/server/layer.rs @@ -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::{ diff --git a/rustfs/src/server/mod.rs b/rustfs/src/server/mod.rs index 3688ac862..3055af39e 100644 --- a/rustfs/src/server/mod.rs +++ b/rustfs/src/server/mod.rs @@ -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; diff --git a/rustfs/src/storage/concurrency/manager.rs b/rustfs/src/storage/concurrency/manager.rs index a577178b2..cd3b929ab 100644 --- a/rustfs/src/storage/concurrency/manager.rs +++ b/rustfs/src/storage/concurrency/manager.rs @@ -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; diff --git a/scripts/layer-dependency-baseline.txt b/scripts/layer-dependency-baseline.txt index 4539e6f75..9b41b7b12 100644 --- a/scripts/layer-dependency-baseline.txt +++ b/scripts/layer-dependency-baseline.txt @@ -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