diff --git a/rustfs/src/admin/console.rs b/rustfs/src/admin/console.rs index 65af8f52d..0e9a0aaec 100644 --- a/rustfs/src/admin/console.rs +++ b/rustfs/src/admin/console.rs @@ -13,6 +13,7 @@ // limitations under the License. use crate::admin::handlers::health::{HealthProbe, build_health_response_parts, collect_dependency_readiness}; +use crate::app::admin_usecase::DefaultAdminUsecase; use crate::app::context::resolve_oidc_handle; use crate::license::has_valid_license; use crate::server::has_path_prefix; @@ -151,6 +152,7 @@ impl Config { port, api: Api { base_url: build_console_api_base_url(&format!("{http_prefix}{local_ip}:{port}")), + discovery: console_api_discovery(), }, s3: S3 { endpoint: format!("{http_prefix}{local_ip}:{port}"), @@ -208,6 +210,26 @@ fn build_console_api_base_url(base_url: &str) -> String { struct Api { #[serde(rename = "baseURL")] base_url: String, + discovery: ApiDiscovery, +} + +#[derive(Debug, Serialize, Clone)] +struct ApiDiscovery { + #[serde(rename = "runtimeCapabilities")] + runtime_capabilities: String, + #[serde(rename = "clusterSnapshot")] + cluster_snapshot: String, + #[serde(rename = "extensionsCatalog")] + extensions_catalog: String, +} + +fn console_api_discovery() -> ApiDiscovery { + let usecase = DefaultAdminUsecase::from_global(); + ApiDiscovery { + runtime_capabilities: usecase.runtime_capabilities_route().to_string(), + cluster_snapshot: usecase.cluster_snapshot_route().to_string(), + extensions_catalog: usecase.extensions_catalog_route().to_string(), + } } #[derive(Debug, Serialize, Clone)] @@ -791,6 +813,33 @@ mod tests { ); } + #[test] + fn console_config_exposes_admin_discovery_paths() { + let cfg = Config::new(IpAddr::V4(Ipv4Addr::LOCALHOST), 9001, "test", "2026-03-16T00:00:00Z"); + + assert_eq!(cfg.api.discovery.runtime_capabilities, "/rustfs/admin/v4/runtime/capabilities"); + assert_eq!(cfg.api.discovery.cluster_snapshot, "/rustfs/admin/v4/cluster/snapshot"); + assert_eq!(cfg.api.discovery.extensions_catalog, "/rustfs/admin/v4/extensions/catalog"); + } + + #[tokio::test] + async fn console_config_handler_serializes_admin_discovery_paths() { + init_console_cfg(IpAddr::V4(Ipv4Addr::LOCALHOST), 9001); + + let response = config_handler(Uri::from_static("http://127.0.0.1:9001/rustfs/console/api/v1/config"), HeaderMap::new()) + .await + .into_response(); + + assert_eq!(response.status(), StatusCode::OK); + let body = response.into_body(); + let bytes = body.collect().await.expect("collect console config body").to_bytes(); + let value: serde_json::Value = serde_json::from_slice(&bytes).expect("console config JSON should deserialize"); + + assert_eq!(value["api"]["discovery"]["runtimeCapabilities"], "/rustfs/admin/v4/runtime/capabilities"); + assert_eq!(value["api"]["discovery"]["clusterSnapshot"], "/rustfs/admin/v4/cluster/snapshot"); + assert_eq!(value["api"]["discovery"]["extensionsCatalog"], "/rustfs/admin/v4/extensions/catalog"); + } + #[test] fn external_admin_paths_are_not_console_paths() { assert!(is_console_path("/rustfs/console/")); diff --git a/rustfs/src/admin/handlers/cluster_snapshot.rs b/rustfs/src/admin/handlers/cluster_snapshot.rs new file mode 100644 index 000000000..953269941 --- /dev/null +++ b/rustfs/src/admin/handlers/cluster_snapshot.rs @@ -0,0 +1,756 @@ +// 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 crate::admin::{ + auth::validate_admin_request, + router::{AdminOperation, Operation, S3Router}, + system, +}; +use crate::app::admin_usecase::DefaultAdminUsecase; +use crate::auth::{check_key_valid, get_session_token}; +use crate::cluster_snapshot::{ + ClusterReadOnlySnapshot, ClusterRuntimeReadinessState, ClusterRuntimeStatusSnapshot, cluster_has_actionable_pressure, +}; +use crate::server::{ADMIN_PREFIX, RemoteAddr}; +use http::{HeaderMap, HeaderValue, StatusCode}; +use hyper::Method; +use matchit::Params; +use rustfs_concurrency::AdmissionState as WorkloadAdmissionState; +use rustfs_concurrency::{AdmissionState, WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshot, WorkloadClass}; +use rustfs_ecstore::api::cluster::{ + ClusterDriveMembership, ClusterEndpointType, ClusterLocalNodeStorage, ClusterLocalNodeStorageSnapshot, + ClusterMembershipSnapshot, ClusterNodeMembership, ClusterPeerHealth, ClusterPeerHealthSnapshot, ClusterPoolState, + ClusterPoolStateSnapshot, +}; +use rustfs_policy::policy::action::{Action, AdminAction}; +use rustfs_storage_api::{CapabilityState, CapabilityStatus, ObservabilitySnapshot, TopologySnapshot}; +use s3s::header::CONTENT_TYPE; +use s3s::{Body, S3Request, S3Response, S3Result, s3_error}; +use serde::Serialize; + +pub fn register_cluster_snapshot_route(r: &mut S3Router) -> std::io::Result<()> { + r.insert( + Method::GET, + format!("{}{}", ADMIN_PREFIX, "/v4/cluster/snapshot").as_str(), + AdminOperation(&GetClusterSnapshotHandler {}), + )?; + + Ok(()) +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct ClusterSnapshotResponse { + pub snapshot: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct ClusterSnapshotDiscoveryResponse { + pub path: String, + pub summary: Option, + pub topology: Option, + pub peer_health: Option, + pub workload_admission: Option, + pub runtime: Option, +} + +async fn authorize_cluster_snapshot_request(req: &S3Request) -> S3Result<()> { + let Some(input_cred) = &req.credentials else { + return Err(s3_error!(InvalidRequest, "authentication required")); + }; + + let (cred, owner) = + check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?; + + validate_admin_request( + &req.headers, + &cred, + owner, + false, + vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)], + req.extensions.get::>().and_then(|opt| opt.map(|a| a.0)), + ) + .await +} + +fn build_json_response( + status: StatusCode, + body: &impl Serialize, + request_id: Option<&HeaderValue>, +) -> S3Result> { + let data = serde_json::to_vec(body).map_err(|e| s3_error!(InternalError, "failed to serialize response: {}", e))?; + let mut header = HeaderMap::new(); + header.insert(CONTENT_TYPE, HeaderValue::from_static("application/json")); + if let Some(value) = request_id { + header.insert("x-request-id", value.clone()); + } + Ok(S3Response::with_headers((status, Body::from(data)), header)) +} + +pub struct GetClusterSnapshotHandler {} + +#[async_trait::async_trait] +impl Operation for GetClusterSnapshotHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + authorize_cluster_snapshot_request(&req).await?; + let snapshot = DefaultAdminUsecase::from_global() + .execute_collect_cluster_read_only_snapshot() + .await + .map(ClusterSnapshotView::from); + build_json_response(StatusCode::OK, &ClusterSnapshotResponse { snapshot }, req.headers.get("x-request-id")) + } +} + +pub(crate) async fn build_cluster_snapshot_discovery_response() -> ClusterSnapshotDiscoveryResponse { + let usecase = DefaultAdminUsecase::from_global(); + let path = usecase.cluster_snapshot_route().to_string(); + let snapshot = usecase.execute_collect_cluster_read_only_snapshot().await; + + match snapshot { + Some(snapshot) => { + let summary = ClusterSnapshotSummary::from(&snapshot); + ClusterSnapshotDiscoveryResponse { + path, + summary: Some(summary.actionable_pressure.clone()), + topology: Some(summary.topology), + peer_health: Some(summary.peer_health), + workload_admission: Some(summary.workload_admission), + runtime: Some(summary.runtime), + } + } + None => ClusterSnapshotDiscoveryResponse { + path, + summary: None, + topology: None, + peer_health: None, + workload_admission: None, + runtime: None, + }, + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct ClusterSnapshotView { + pub summary: ClusterSnapshotSummary, + pub runtime_capabilities_path: String, + pub extensions_catalog_path: String, + pub topology: TopologySnapshot, + pub membership: ClusterMembershipView, + pub pool_state: ClusterPoolStateView, + pub local_storage: ClusterLocalStorageView, + pub peer_health: ClusterPeerHealthView, + pub observability: ObservabilitySnapshot, + pub workload_admission: Vec, + pub runtime_status: ClusterRuntimeStatusView, + pub actionable_pressure: bool, +} + +impl From for ClusterSnapshotView { + fn from(snapshot: ClusterReadOnlySnapshot) -> Self { + let actionable_pressure = cluster_has_actionable_pressure(&snapshot); + Self { + summary: ClusterSnapshotSummary::from(&snapshot), + runtime_capabilities_path: format!("{}{}", ADMIN_PREFIX, system::RUNTIME_CAPABILITIES_ROUTE_SUFFIX), + extensions_catalog_path: format!("{}{}", ADMIN_PREFIX, "/v4/extensions/catalog"), + topology: snapshot.topology, + membership: ClusterMembershipView::from(snapshot.membership), + pool_state: ClusterPoolStateView::from(snapshot.pool_state), + local_storage: ClusterLocalStorageView::from(snapshot.local_storage), + peer_health: ClusterPeerHealthView::from(snapshot.peer_health), + observability: snapshot.observability, + workload_admission: workload_admission_views(snapshot.workload_admission), + runtime_status: ClusterRuntimeStatusView::from(snapshot.runtime_status), + actionable_pressure, + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct ClusterSnapshotSummary { + pub runtime: CapabilityStatus, + pub topology: CapabilityStatus, + pub membership: CapabilityStatus, + pub peer_health: CapabilityStatus, + pub observability: CapabilityStatus, + pub workload_admission: CapabilityStatus, + pub actionable_pressure: CapabilityStatus, +} + +impl From<&ClusterReadOnlySnapshot> for ClusterSnapshotSummary { + fn from(snapshot: &ClusterReadOnlySnapshot) -> Self { + let topology = summarize_topology(snapshot); + let membership = summarize_membership(snapshot); + let peer_health = summarize_peer_health(snapshot); + let observability = summarize_observability(snapshot); + let workload_admission = summarize_workload_admission(snapshot); + let actionable_pressure = if cluster_has_actionable_pressure(snapshot) { + CapabilityStatus::supported().with_reason("cluster snapshot reports degraded runtime or non-open admission") + } else { + CapabilityStatus::disabled().with_reason("cluster snapshot reports no actionable pressure") + }; + + Self { + runtime: summarize_runtime(snapshot), + topology, + membership, + peer_health, + observability, + workload_admission, + actionable_pressure, + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct ClusterMembershipView { + pub nodes: Vec, + pub drives: Vec, +} + +impl From for ClusterMembershipView { + fn from(snapshot: ClusterMembershipSnapshot) -> Self { + Self { + nodes: snapshot.nodes.into_iter().map(ClusterNodeMembershipView::from).collect(), + drives: snapshot.drives.into_iter().map(ClusterDriveMembershipView::from).collect(), + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct ClusterNodeMembershipView { + pub node_id: String, + pub grid_host: String, + pub is_local: bool, + pub pools: Vec, +} + +impl From for ClusterNodeMembershipView { + fn from(node: ClusterNodeMembership) -> Self { + Self { + node_id: node.node_id, + grid_host: node.grid_host, + is_local: node.is_local, + pools: node.pools, + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct ClusterDriveMembershipView { + pub pool_index: usize, + pub set_index: usize, + pub disk_index: usize, + pub node_id: String, + pub is_local: bool, + pub endpoint_type: &'static str, +} + +impl From for ClusterDriveMembershipView { + fn from(drive: ClusterDriveMembership) -> Self { + Self { + pool_index: drive.pool_index, + set_index: drive.set_index, + disk_index: drive.disk_index, + node_id: drive.node_id, + is_local: drive.is_local, + endpoint_type: endpoint_type_label(drive.endpoint_type), + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct ClusterPoolStateView { + pub pools: Vec, +} + +impl From for ClusterPoolStateView { + fn from(snapshot: ClusterPoolStateSnapshot) -> Self { + Self { + pools: snapshot.pools.into_iter().map(ClusterPoolStateItemView::from).collect(), + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct ClusterPoolStateItemView { + pub pool_index: usize, + pub set_count: usize, + pub drives_per_set: usize, + pub endpoint_count: usize, + pub local_drive_count: usize, + pub remote_drive_count: usize, + pub legacy: bool, + pub endpoint_types: Vec<&'static str>, +} + +impl From for ClusterPoolStateItemView { + fn from(pool: ClusterPoolState) -> Self { + Self { + pool_index: pool.pool_index, + set_count: pool.set_count, + drives_per_set: pool.drives_per_set, + endpoint_count: pool.endpoint_count, + local_drive_count: pool.local_drive_count, + remote_drive_count: pool.remote_drive_count, + legacy: pool.legacy, + endpoint_types: pool.endpoint_types.into_iter().map(endpoint_type_label).collect(), + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct ClusterLocalStorageView { + pub nodes: Vec, +} + +impl From for ClusterLocalStorageView { + fn from(snapshot: ClusterLocalNodeStorageSnapshot) -> Self { + Self { + nodes: snapshot.nodes.into_iter().map(ClusterLocalNodeStorageView::from).collect(), + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct ClusterLocalNodeStorageView { + pub node_id: String, + pub pools: Vec, + pub drive_count: usize, + pub path_drive_count: usize, + pub url_drive_count: usize, +} + +impl From for ClusterLocalNodeStorageView { + fn from(node: ClusterLocalNodeStorage) -> Self { + Self { + node_id: node.node_id, + pools: node.pools, + drive_count: node.drive_count, + path_drive_count: node.path_drive_count, + url_drive_count: node.url_drive_count, + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct ClusterPeerHealthView { + pub peers: Vec, +} + +impl From for ClusterPeerHealthView { + fn from(snapshot: ClusterPeerHealthSnapshot) -> Self { + Self { + peers: snapshot.peers.into_iter().map(ClusterPeerHealthItemView::from).collect(), + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct ClusterPeerHealthItemView { + pub node_id: String, + pub is_local: bool, + pub status: CapabilityStatus, +} + +impl From for ClusterPeerHealthItemView { + fn from(peer: ClusterPeerHealth) -> Self { + Self { + node_id: peer.node_id, + is_local: peer.is_local, + status: peer.status, + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct WorkloadAdmissionView { + pub class: &'static str, + pub state: &'static str, + pub active: Option, + pub queued: Option, + pub limit: Option, + pub reason: Option, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct ClusterRuntimeStatusView { + pub state: &'static str, + pub storage_ready: bool, + pub iam_ready: bool, + pub lock_quorum_ready: bool, + pub degraded_reasons: Vec<&'static str>, +} + +impl From for ClusterRuntimeStatusView { + fn from(runtime: ClusterRuntimeStatusSnapshot) -> Self { + Self { + state: runtime_readiness_state_label(runtime.state), + storage_ready: runtime.readiness.storage_ready, + iam_ready: runtime.readiness.iam_ready, + lock_quorum_ready: runtime.readiness.lock_quorum_ready, + degraded_reasons: runtime.degraded_reasons.into_iter().map(|reason| reason.as_str()).collect(), + } + } +} + +fn workload_admission_views(snapshot: WorkloadAdmissionRegistrySnapshot) -> Vec { + snapshot.entries().iter().cloned().map(WorkloadAdmissionView::from).collect() +} + +impl From for WorkloadAdmissionView { + fn from(snapshot: WorkloadAdmissionSnapshot) -> Self { + Self { + class: workload_class_label(snapshot.class), + state: admission_state_label(snapshot.state), + active: snapshot.active, + queued: snapshot.queued, + limit: snapshot.limit, + reason: snapshot.reason, + } + } +} + +fn endpoint_type_label(endpoint_type: ClusterEndpointType) -> &'static str { + match endpoint_type { + ClusterEndpointType::Path => "path", + ClusterEndpointType::Url => "url", + } +} + +fn workload_class_label(class: WorkloadClass) -> &'static str { + class.as_str() +} + +fn admission_state_label(state: AdmissionState) -> &'static str { + match state { + AdmissionState::Open => "open", + AdmissionState::Throttled => "throttled", + AdmissionState::Saturated => "saturated", + AdmissionState::Disabled => "disabled", + AdmissionState::Unknown => "unknown", + } +} + +fn runtime_readiness_state_label(state: ClusterRuntimeReadinessState) -> &'static str { + match state { + ClusterRuntimeReadinessState::Ready => "ready", + ClusterRuntimeReadinessState::Degraded => "degraded", + ClusterRuntimeReadinessState::Unknown => "unknown", + } +} + +fn summarize_runtime(snapshot: &ClusterReadOnlySnapshot) -> CapabilityStatus { + match snapshot.runtime_status.state { + ClusterRuntimeReadinessState::Ready => CapabilityStatus::supported().with_reason("runtime readiness reports ready"), + ClusterRuntimeReadinessState::Degraded => { + let reasons = snapshot + .runtime_status + .degraded_reasons + .iter() + .map(|reason| reason.as_str()) + .collect::>() + .join(", "); + CapabilityStatus::unknown().with_reason(format!("runtime readiness degraded: {reasons}")) + } + ClusterRuntimeReadinessState::Unknown => CapabilityStatus::unknown().with_reason("runtime readiness status unknown"), + } +} + +fn summarize_topology(snapshot: &ClusterReadOnlySnapshot) -> CapabilityStatus { + summarize_named_capability_statuses( + [ + ("profiling", &snapshot.topology.capabilities.profiling), + ("numa", &snapshot.topology.capabilities.numa), + ("failure_domain_labels", &snapshot.topology.capabilities.failure_domain_labels), + ("media_labels", &snapshot.topology.capabilities.media_labels), + ], + "topology capability", + ) + .with_reason("topology summary resolved from cluster snapshot") +} + +fn summarize_membership(snapshot: &ClusterReadOnlySnapshot) -> CapabilityStatus { + if snapshot.membership.nodes.is_empty() { + CapabilityStatus::unknown().with_reason("cluster membership has no nodes") + } else { + CapabilityStatus::supported().with_reason(format!( + "cluster membership reports {} nodes and {} drives", + snapshot.membership.nodes.len(), + snapshot.membership.drives.len() + )) + } +} + +fn summarize_peer_health(snapshot: &ClusterReadOnlySnapshot) -> CapabilityStatus { + if snapshot.peer_health.peers.is_empty() { + return CapabilityStatus::unknown().with_reason("cluster peer health has no peers"); + } + + let unknown = snapshot + .peer_health + .peers + .iter() + .filter(|peer| peer.status.state == CapabilityState::Unknown) + .count(); + if unknown > 0 { + CapabilityStatus::unknown().with_reason(format!("cluster peer health has {unknown} unresolved peers")) + } else { + CapabilityStatus::supported().with_reason("cluster peer health resolved for all peers") + } +} + +fn summarize_workload_admission(snapshot: &ClusterReadOnlySnapshot) -> CapabilityStatus { + let entries = snapshot.workload_admission.entries(); + if entries.is_empty() { + return CapabilityStatus::unknown().with_reason("workload admission snapshot is empty"); + } + + let non_open = entries + .iter() + .filter(|entry| entry.state != WorkloadAdmissionState::Open) + .map(|entry| entry.class.as_str()) + .collect::>(); + if non_open.is_empty() { + CapabilityStatus::supported().with_reason("all workload admission classes are open") + } else { + CapabilityStatus::unknown().with_reason(format!("workload admission has non-open classes: {}", non_open.join(", "))) + } +} + +fn summarize_observability(snapshot: &ClusterReadOnlySnapshot) -> CapabilityStatus { + let userspace_profiling = summarize_named_capability_statuses( + [ + ("cpu", &snapshot.observability.userspace_profiling.cpu), + ("memory", &snapshot.observability.userspace_profiling.memory), + ("continuous_cpu", &snapshot.observability.userspace_profiling.continuous_cpu), + ("periodic_cpu", &snapshot.observability.userspace_profiling.periodic_cpu), + ], + "userspace profiling", + ); + let memory_sampling = summarize_named_capability_statuses( + [ + ("process", &snapshot.observability.memory_sampling.process), + ("system", &snapshot.observability.memory_sampling.system), + ("cgroup", &snapshot.observability.memory_sampling.cgroup), + ], + "memory sampling", + ); + let platform = summarize_named_capability_statuses( + [ + ("allocator", &snapshot.observability.platform.allocator), + ("ebpf", &snapshot.observability.platform.ebpf), + ("numa", &snapshot.observability.platform.numa), + ], + "platform support", + ); + + if [userspace_profiling.state, memory_sampling.state, platform.state] + .into_iter() + .any(|state| state == CapabilityState::Unknown) + { + CapabilityStatus::unknown().with_reason("observability summary resolved from cluster snapshot") + } else { + CapabilityStatus::supported().with_reason("observability summary resolved from cluster snapshot") + } +} + +fn summarize_named_capability_statuses( + statuses: [(&'static str, &CapabilityStatus); N], + subject: &'static str, +) -> CapabilityStatus { + let unknown = statuses + .iter() + .filter_map(|(name, status)| (status.state == CapabilityState::Unknown).then_some(*name)) + .collect::>(); + if !unknown.is_empty() { + return CapabilityStatus::unknown().with_reason(format!("{subject} unresolved fields: {}", unknown.join(", "))); + } + + let supported_or_disabled = statuses + .iter() + .any(|(_, status)| matches!(status.state, CapabilityState::Supported | CapabilityState::Disabled)); + if supported_or_disabled { + return CapabilityStatus::supported(); + } + + let unsupported = statuses.iter().map(|(name, _)| *name).collect::>(); + CapabilityStatus::unsupported().with_reason(format!("{subject} unsupported fields: {}", unsupported.join(", "))) +} + +#[cfg(test)] +mod tests { + use super::{ClusterSnapshotResponse, ClusterSnapshotSummary, ClusterSnapshotView}; + use crate::cluster_snapshot::{ClusterReadOnlySnapshot, ClusterRuntimeReadinessState, ClusterRuntimeStatusSnapshot}; + use crate::server::{DependencyReadiness, ReadinessDegradedReason}; + use rustfs_concurrency::{AdmissionState, WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshot, WorkloadClass}; + use rustfs_ecstore::api::cluster::{ + ClusterDriveMembership, ClusterEndpointType, ClusterLocalNodeStorage, ClusterLocalNodeStorageSnapshot, + ClusterMembershipSnapshot, ClusterNodeMembership, ClusterPeerHealth, ClusterPeerHealthSnapshot, ClusterPoolState, + ClusterPoolStateSnapshot, + }; + use rustfs_storage_api::CapabilityState; + use rustfs_storage_api::{CapabilityStatus, ObservabilitySnapshot, TopologySnapshot}; + + #[test] + fn cluster_snapshot_handler_requires_server_info_admin_permission() { + let src = include_str!("cluster_snapshot.rs"); + let handler_block = extract_block_between_markers(src, "impl Operation for GetClusterSnapshotHandler", "#[cfg(test)]"); + let auth_block = + extract_block_between_markers(src, "async fn authorize_cluster_snapshot_request", "fn build_json_response"); + + assert!( + handler_block.contains("authorize_cluster_snapshot_request(&req).await?;"), + "cluster snapshot handler should require admin authorization" + ); + assert!( + auth_block.contains("AdminAction::ServerInfoAdminAction"), + "cluster snapshot should require server info admin permission" + ); + } + + #[test] + fn cluster_snapshot_response_serializes_none_snapshot() { + let value = serde_json::to_value(ClusterSnapshotResponse { snapshot: None }).expect("serialize response"); + assert_eq!(value, serde_json::json!({ "snapshot": null })); + } + + #[tokio::test] + async fn cluster_snapshot_discovery_reports_path_without_snapshot() { + let response = super::build_cluster_snapshot_discovery_response().await; + + assert_eq!(response.path, "/rustfs/admin/v4/cluster/snapshot"); + assert_eq!(response.summary, None); + assert_eq!(response.topology, None); + assert_eq!(response.peer_health, None); + assert_eq!(response.workload_admission, None); + assert_eq!(response.runtime, None); + } + + #[test] + fn cluster_snapshot_view_serializes_machine_readable_sections() { + let snapshot = ClusterReadOnlySnapshot { + topology: TopologySnapshot::default(), + membership: ClusterMembershipSnapshot { + nodes: vec![ClusterNodeMembership { + node_id: "node-a".to_string(), + grid_host: "node-a:9000".to_string(), + is_local: true, + pools: vec![0], + }], + drives: vec![ClusterDriveMembership { + pool_index: 0, + set_index: 1, + disk_index: 2, + node_id: "node-a".to_string(), + is_local: true, + endpoint_type: ClusterEndpointType::Url, + }], + }, + pool_state: ClusterPoolStateSnapshot { + pools: vec![ClusterPoolState { + pool_index: 0, + set_count: 1, + drives_per_set: 4, + endpoint_count: 4, + local_drive_count: 2, + remote_drive_count: 2, + legacy: false, + endpoint_types: vec![ClusterEndpointType::Url], + }], + }, + local_storage: ClusterLocalNodeStorageSnapshot { + nodes: vec![ClusterLocalNodeStorage { + node_id: "node-a".to_string(), + pools: vec![0], + drive_count: 2, + path_drive_count: 0, + url_drive_count: 2, + }], + }, + peer_health: ClusterPeerHealthSnapshot { + peers: vec![ClusterPeerHealth { + node_id: "node-b".to_string(), + is_local: false, + status: CapabilityStatus::unknown().with_reason("peer state unavailable"), + }], + }, + observability: ObservabilitySnapshot::default(), + workload_admission: WorkloadAdmissionRegistrySnapshot::new(vec![ + WorkloadAdmissionSnapshot::new(WorkloadClass::Repair, AdmissionState::Unknown) + .with_counts(Some(1), Some(2), Some(8)) + .with_reason("repair backlog not yet normalized"), + ]), + runtime_status: ClusterRuntimeStatusSnapshot { + readiness: DependencyReadiness { + storage_ready: false, + iam_ready: true, + lock_quorum_ready: false, + }, + state: ClusterRuntimeReadinessState::Degraded, + degraded_reasons: vec![ReadinessDegradedReason::StorageAndLockUnavailable], + }, + }; + + let value = serde_json::to_value(ClusterSnapshotView::from(snapshot)).expect("serialize view"); + assert_eq!(value["runtime_capabilities_path"], "/rustfs/admin/v4/runtime/capabilities"); + assert_eq!(value["extensions_catalog_path"], "/rustfs/admin/v4/extensions/catalog"); + assert_eq!(value["membership"]["drives"][0]["endpoint_type"], "url"); + assert_eq!(value["workload_admission"][0]["class"], "repair"); + assert_eq!(value["workload_admission"][0]["state"], "unknown"); + assert_eq!(value["runtime_status"]["state"], "degraded"); + assert_eq!(value["summary"]["runtime"]["state"], "unknown"); + assert_eq!(value["runtime_status"]["degraded_reasons"][0], "storage_and_lock_unavailable"); + assert_eq!(value["actionable_pressure"], true); + } + + #[test] + fn cluster_snapshot_summary_reports_cross_surface_status() { + let snapshot = ClusterReadOnlySnapshot { + topology: TopologySnapshot::default(), + membership: ClusterMembershipSnapshot::default(), + pool_state: ClusterPoolStateSnapshot::default(), + local_storage: ClusterLocalNodeStorageSnapshot::default(), + peer_health: ClusterPeerHealthSnapshot::default(), + observability: ObservabilitySnapshot::default(), + workload_admission: WorkloadAdmissionRegistrySnapshot::new(vec![WorkloadAdmissionSnapshot::new( + WorkloadClass::ForegroundRead, + AdmissionState::Open, + )]), + runtime_status: ClusterRuntimeStatusSnapshot { + readiness: DependencyReadiness { + storage_ready: true, + iam_ready: true, + lock_quorum_ready: true, + }, + state: ClusterRuntimeReadinessState::Ready, + degraded_reasons: Vec::new(), + }, + }; + + let summary = ClusterSnapshotSummary::from(&snapshot); + assert_eq!(summary.runtime.state, CapabilityState::Supported); + assert_eq!(summary.membership.state, CapabilityState::Unknown); + assert_eq!(summary.peer_health.state, CapabilityState::Unknown); + assert_eq!(summary.workload_admission.state, CapabilityState::Supported); + assert_eq!(summary.actionable_pressure.state, CapabilityState::Disabled); + } + + fn extract_block_between_markers(src: &str, start: &str, end: &str) -> String { + let start_index = src.find(start).expect("start marker should exist"); + let end_index = src[start_index..] + .find(end) + .map(|index| start_index + index) + .unwrap_or(src.len()); + src[start_index..end_index].to_string() + } +} diff --git a/rustfs/src/admin/handlers/extensions.rs b/rustfs/src/admin/handlers/extensions.rs index d6a41c329..35635f2d9 100644 --- a/rustfs/src/admin/handlers/extensions.rs +++ b/rustfs/src/admin/handlers/extensions.rs @@ -14,7 +14,7 @@ use crate::admin::{ auth::validate_admin_request, - handlers::{plugins_instances, system}, + handlers::{cluster_snapshot, plugins_instances, system}, plugin_contract::{ PluginContractDomain, PluginInstanceDiagnosticCode, PluginInstanceDiagnosticCount, PluginInstanceEntry, PluginInstanceSource, PluginOperationalStateContract, @@ -62,6 +62,7 @@ pub fn register_extension_route(r: &mut S3Router) -> std::io::Re pub(crate) struct ExtensionCatalogResponse { pub extensions: Vec, pub runtime_capabilities: ExtensionRuntimeCapabilitiesResponse, + pub cluster_snapshot: cluster_snapshot::ClusterSnapshotDiscoveryResponse, pub external_plugin_flow: TargetPluginExternalFlowGateStatus, } @@ -123,10 +124,12 @@ async fn build_extension_catalog_response() -> Result for ModuleSwitchesResponse { fn from(value: ModuleSwitchSnapshot) -> Self { + let usecase = DefaultAdminUsecase::from_global(); Self { notify_enabled: value.notify_enabled, audit_enabled: value.audit_enabled, @@ -73,6 +86,11 @@ impl From for ModuleSwitchesResponse { persisted_audit_enabled: value.persisted_audit_enabled, notify_source: value.notify_source, audit_source: value.audit_source, + admin_discovery: ModuleSwitchDiscovery { + runtime_capabilities: usecase.runtime_capabilities_route(), + cluster_snapshot: usecase.cluster_snapshot_route(), + extensions_catalog: usecase.extensions_catalog_route(), + }, } } } @@ -195,6 +213,8 @@ impl Operation for UpdateModuleSwitchesHandler { #[cfg(test)] mod tests { + use super::{ModuleSwitchDiscovery, ModuleSwitchSource, ModuleSwitchesResponse}; + #[test] fn module_switch_handlers_require_admin_authorization_contract() { let src = include_str!("module_switch.rs"); @@ -215,6 +235,28 @@ mod tests { ); } + #[test] + fn module_switch_response_exposes_admin_discovery_paths() { + let response = ModuleSwitchesResponse { + notify_enabled: true, + audit_enabled: false, + persisted_notify_enabled: true, + persisted_audit_enabled: false, + notify_source: ModuleSwitchSource::Console, + audit_source: ModuleSwitchSource::Console, + admin_discovery: ModuleSwitchDiscovery { + runtime_capabilities: "/rustfs/admin/v4/runtime/capabilities", + cluster_snapshot: "/rustfs/admin/v4/cluster/snapshot", + extensions_catalog: "/rustfs/admin/v4/extensions/catalog", + }, + }; + + let value = serde_json::to_value(response).expect("module switch response should serialize"); + assert_eq!(value["admin_discovery"]["runtimeCapabilities"], "/rustfs/admin/v4/runtime/capabilities"); + assert_eq!(value["admin_discovery"]["clusterSnapshot"], "/rustfs/admin/v4/cluster/snapshot"); + assert_eq!(value["admin_discovery"]["extensionsCatalog"], "/rustfs/admin/v4/extensions/catalog"); + } + fn extract_block_between_markers<'a>(src: &'a str, start_marker: &str, end_marker: &str) -> &'a str { let start = src .find(start_marker) diff --git a/rustfs/src/admin/handlers/plugins_catalog.rs b/rustfs/src/admin/handlers/plugins_catalog.rs index 5a1fb7890..43215b66f 100644 --- a/rustfs/src/admin/handlers/plugins_catalog.rs +++ b/rustfs/src/admin/handlers/plugins_catalog.rs @@ -15,11 +15,12 @@ use crate::admin::{ auth::validate_admin_request, plugin_contract::{ - PluginCatalogDomainEntry, PluginCatalogEntry, PluginCatalogResponse, PluginContractDomain, PluginContractEntrypointKind, - PluginContractPackaging, PluginDistributionContract, PluginRuntimeContract, + PluginCatalogAdminDiscovery, PluginCatalogDomainEntry, PluginCatalogEntry, PluginCatalogResponse, PluginContractDomain, + PluginContractEntrypointKind, PluginContractPackaging, PluginDistributionContract, PluginRuntimeContract, }, router::{AdminOperation, Operation, S3Router}, }; +use crate::app::admin_usecase::DefaultAdminUsecase; use crate::auth::{check_key_valid, get_session_token}; use crate::server::{ADMIN_PREFIX, RemoteAddr}; use http::{HeaderMap, HeaderValue, StatusCode}; @@ -57,6 +58,7 @@ fn target_domain_name_from_subsystem(subsystem: &str) -> PluginContractDomain { } fn build_catalog_response() -> PluginCatalogResponse { + let usecase = DefaultAdminUsecase::from_global(); let mut plugins: HashMap<&'static str, PluginCatalogEntry> = HashMap::new(); for descriptor in builtin_notify_target_admin_descriptors() @@ -74,7 +76,14 @@ fn build_catalog_response() -> PluginCatalogResponse { plugin.domain_configs.sort_by_key(|a| a.domain); } - PluginCatalogResponse { plugins } + PluginCatalogResponse { + plugins, + admin_discovery: PluginCatalogAdminDiscovery { + runtime_capabilities: usecase.runtime_capabilities_route().to_string(), + cluster_snapshot: usecase.cluster_snapshot_route().to_string(), + extensions_catalog: usecase.extensions_catalog_route().to_string(), + }, + } } fn example_external_webhook_plugin_entry() -> PluginCatalogEntry { @@ -200,6 +209,10 @@ mod tests { fn plugin_catalog_contains_representative_builtin_targets() { let response = build_catalog_response(); + assert_eq!(response.admin_discovery.runtime_capabilities, "/rustfs/admin/v4/runtime/capabilities"); + assert_eq!(response.admin_discovery.cluster_snapshot, "/rustfs/admin/v4/cluster/snapshot"); + assert_eq!(response.admin_discovery.extensions_catalog, "/rustfs/admin/v4/extensions/catalog"); + let webhook = response .plugins .iter() diff --git a/rustfs/src/admin/handlers/pools.rs b/rustfs/src/admin/handlers/pools.rs index bf0d379c7..e24fe263c 100644 --- a/rustfs/src/admin/handlers/pools.rs +++ b/rustfs/src/admin/handlers/pools.rs @@ -40,6 +40,22 @@ use hyper::Method; use std::collections::HashSet; +#[derive(Debug, Clone, serde::Serialize)] +struct PoolAdminDiscovery { + #[serde(rename = "runtimeCapabilities")] + runtime_capabilities: String, + #[serde(rename = "clusterSnapshot")] + cluster_snapshot: String, + #[serde(rename = "extensionsCatalog")] + extensions_catalog: String, +} + +#[derive(Debug, Clone, serde::Serialize)] +struct PoolStatusResponse { + pool: crate::app::admin_usecase::AdminPoolStatus, + admin_discovery: PoolAdminDiscovery, +} + const LOG_COMPONENT_ADMIN_API: &str = "admin_api"; const LOG_SUBSYSTEM_POOL_ADMIN: &str = "pool_admin"; const EVENT_ADMIN_REQUEST_STATE: &str = "admin_request_state"; @@ -396,6 +412,15 @@ fn parse_pool_idx_by_id(pool: &str, endpoint_count: usize) -> Option { (idx < endpoint_count).then_some(idx) } +fn pool_admin_discovery() -> PoolAdminDiscovery { + let usecase = DefaultAdminUsecase::from_global(); + PoolAdminDiscovery { + runtime_capabilities: usecase.runtime_capabilities_route().to_string(), + cluster_snapshot: usecase.cluster_snapshot_route().to_string(), + extensions_catalog: usecase.extensions_catalog_route().to_string(), + } +} + fn has_duplicate_indices(indices: &[usize]) -> bool { let mut seen = HashSet::with_capacity(indices.len()); for idx in indices { @@ -580,8 +605,12 @@ impl Operation for StatusPool { }) .await .map_err(S3Error::from)?; + let response = PoolStatusResponse { + pool: pools_status, + admin_discovery: pool_admin_discovery(), + }; - let data = serde_json::to_vec(&pools_status).map_err(|e| { + let data = serde_json::to_vec(&response).map_err(|e| { log_pool_request_failed!("query_pool_status", "serialize_pool_status_failed", e); S3Error::with_message(S3ErrorCode::InternalError, "parse accountInfo failed") })?; @@ -1320,4 +1349,33 @@ mod pools_handler_tests { assert!(!has_duplicate_indices(&empty)); assert!(!has_duplicate_indices(&[0, 2, 1, 3])); } + + #[test] + fn test_pool_status_response_exposes_admin_discovery_paths() { + let response = super::PoolStatusResponse { + pool: crate::app::admin_usecase::AdminPoolStatus { + id: 0, + cmd_line: "pool-0".to_string(), + last_update: time::OffsetDateTime::UNIX_EPOCH, + total_size: 0, + current_size: 0, + used_size: 0, + used: 0.0, + status: "active".to_string(), + decommission_status: "none".to_string(), + rebalance_status: "none".to_string(), + decommission: None, + }, + admin_discovery: super::PoolAdminDiscovery { + runtime_capabilities: "/rustfs/admin/v4/runtime/capabilities".to_string(), + cluster_snapshot: "/rustfs/admin/v4/cluster/snapshot".to_string(), + extensions_catalog: "/rustfs/admin/v4/extensions/catalog".to_string(), + }, + }; + + let value = serde_json::to_value(response).expect("pool status response should serialize"); + assert_eq!(value["admin_discovery"]["runtimeCapabilities"], "/rustfs/admin/v4/runtime/capabilities"); + assert_eq!(value["admin_discovery"]["clusterSnapshot"], "/rustfs/admin/v4/cluster/snapshot"); + assert_eq!(value["admin_discovery"]["extensionsCatalog"], "/rustfs/admin/v4/extensions/catalog"); + } } diff --git a/rustfs/src/admin/handlers/system.rs b/rustfs/src/admin/handlers/system.rs index 765909cb6..1cbae5ce3 100644 --- a/rustfs/src/admin/handlers/system.rs +++ b/rustfs/src/admin/handlers/system.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use super::metrics; +use super::{cluster_snapshot, metrics}; use crate::admin::auth::validate_admin_request; use crate::admin::router::{AdminOperation, Operation, S3Router}; use crate::app::admin_usecase::{DefaultAdminUsecase, QueryServerInfoRequest}; @@ -25,6 +25,7 @@ use http::{HeaderMap, HeaderValue}; use hyper::{Method, StatusCode}; use matchit::Params; use rustfs_concurrency::WorkloadAdmissionRegistrySnapshot; +use rustfs_madmin::{InfoMessage, StorageInfo}; use rustfs_policy::policy::action::{Action, AdminAction, S3Action}; use rustfs_storage_api::{ CapabilityState, CapabilityStatus, ObservabilitySnapshotProvider, TopologySnapshot, TopologySnapshotProvider, @@ -149,6 +150,36 @@ impl Operation for ServiceHandle { pub struct ServerInfoHandler {} +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +struct SystemAdminDiscovery { + #[serde(rename = "runtimeCapabilities")] + runtime_capabilities: String, + #[serde(rename = "clusterSnapshot")] + cluster_snapshot: String, + #[serde(rename = "extensionsCatalog")] + extensions_catalog: String, +} + +#[derive(Serialize)] +struct ServerInfoResponse { + info: InfoMessage, + admin_discovery: SystemAdminDiscovery, +} + +#[derive(Serialize)] +struct StorageInfoResponse { + info: StorageInfo, + admin_discovery: SystemAdminDiscovery, +} + +fn system_admin_discovery(usecase: &DefaultAdminUsecase) -> SystemAdminDiscovery { + SystemAdminDiscovery { + runtime_capabilities: usecase.runtime_capabilities_route().to_string(), + cluster_snapshot: usecase.cluster_snapshot_route().to_string(), + extensions_catalog: usecase.extensions_catalog_route().to_string(), + } +} + #[async_trait::async_trait] impl Operation for ServerInfoHandler { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { @@ -177,8 +208,12 @@ impl Operation for ServerInfoHandler { .await .map_err(S3Error::from)? .info; + let response = ServerInfoResponse { + info, + admin_discovery: system_admin_discovery(&usecase), + }; - let data = serde_json::to_vec(&info).map_err(|e| { + let data = serde_json::to_vec(&response).map_err(|e| { log_system_request_failed!("query_server_info", "serialize_server_info_failed", e); S3Error::with_message(S3ErrorCode::InternalError, "parse serverInfo failed") })?; @@ -227,8 +262,12 @@ impl Operation for StorageInfoHandler { let usecase = DefaultAdminUsecase::from_global(); let info = usecase.execute_query_storage_info().await.map_err(S3Error::from)?; + let response = StorageInfoResponse { + info, + admin_discovery: system_admin_discovery(&usecase), + }; - let data = serde_json::to_vec(&info).map_err(|e| { + let data = serde_json::to_vec(&response).map_err(|e| { log_system_request_failed!("query_storage_info", "serialize_storage_info_failed", e); S3Error::with_message(S3ErrorCode::InternalError, "failed to serialize storage info") })?; @@ -250,11 +289,14 @@ pub struct RuntimeCapabilitiesSummary { pub memory_sampling: CapabilityStatus, pub platform: CapabilityStatus, pub topology: CapabilityStatus, + pub cluster_snapshot: CapabilityStatus, } #[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub struct RuntimeCapabilitiesResponse { pub summary: RuntimeCapabilitiesSummary, + pub cluster_snapshot_path: String, + pub cluster_snapshot_summary: Option, pub observability: rustfs_storage_api::ObservabilitySnapshot, pub workload_admission: WorkloadAdmissionRegistrySnapshot, pub topology: Option, @@ -265,9 +307,11 @@ pub struct RuntimeCapabilitiesHandler {} pub(crate) async fn build_runtime_capabilities_response() -> Result { + let usecase = DefaultAdminUsecase::from_global(); let observability_provider = RustFsObservabilitySnapshotProvider; let observability = observability_provider.observability_snapshot().await?; let workload_admission = workload_admission_registry_snapshot(); + let cluster_snapshot_discovery = cluster_snapshot::build_cluster_snapshot_discovery_response().await; let (topology, topology_status) = if let Some(endpoint_pools) = resolve_endpoints_handle() { let topology_provider = EndpointTopologySnapshotProvider::new(endpoint_pools); @@ -276,10 +320,17 @@ pub(crate) async fn build_runtime_capabilities_response() } else { (None, CapabilityStatus::unknown().with_reason(TOPOLOGY_SNAPSHOT_NOT_AVAILABLE)) }; - let summary = build_runtime_capabilities_summary(&observability, topology.as_ref(), &topology_status); + let summary = build_runtime_capabilities_summary( + &observability, + topology.as_ref(), + &topology_status, + cluster_snapshot_discovery.summary.as_ref(), + ); Ok(RuntimeCapabilitiesResponse { summary, + cluster_snapshot_path: usecase.cluster_snapshot_route().to_string(), + cluster_snapshot_summary: cluster_snapshot_discovery.summary, observability, workload_admission, topology, @@ -291,6 +342,7 @@ fn build_runtime_capabilities_summary( observability: &rustfs_storage_api::ObservabilitySnapshot, topology: Option<&TopologySnapshot>, topology_status: &CapabilityStatus, + cluster_snapshot_summary: Option<&CapabilityStatus>, ) -> RuntimeCapabilitiesSummary { let userspace_profiling = summarize_named_capability_statuses( [ @@ -347,6 +399,9 @@ fn build_runtime_capabilities_summary( memory_sampling, platform, topology: topology_summary, + cluster_snapshot: cluster_snapshot_summary.cloned().unwrap_or_else(|| { + CapabilityStatus::unknown().with_reason("cluster snapshot is not available before storage endpoint pools initialize") + }), } } @@ -457,10 +512,12 @@ impl Operation for DataUsageInfoHandler { #[cfg(test)] mod tests { use super::{ - OBSERVABILITY_SUMMARY_RESOLVED, TOPOLOGY_SNAPSHOT_NOT_AVAILABLE, TOPOLOGY_SUMMARY_RESOLVED, - build_runtime_capabilities_response, build_runtime_capabilities_summary, + OBSERVABILITY_SUMMARY_RESOLVED, ServerInfoResponse, TOPOLOGY_SNAPSHOT_NOT_AVAILABLE, TOPOLOGY_SUMMARY_RESOLVED, + build_runtime_capabilities_response, build_runtime_capabilities_summary, system_admin_discovery, }; + use crate::app::admin_usecase::DefaultAdminUsecase; use rustfs_concurrency::{AdmissionState, WorkloadClass}; + use rustfs_madmin::{InfoMessage, StorageInfo}; use rustfs_storage_api::{ CapabilityState, CapabilityStatus, MemorySamplingState, ObservabilitySnapshot, PlatformSupport, TopologyCapabilities, TopologySnapshot, UserspaceProfilingCapability, @@ -477,6 +534,9 @@ mod tests { assert_eq!(response.topology_status.reason.as_deref(), Some(TOPOLOGY_SNAPSHOT_NOT_AVAILABLE)); assert_eq!(response.summary.topology.state, CapabilityState::Unknown); assert_eq!(response.summary.topology.reason.as_deref(), Some(TOPOLOGY_SNAPSHOT_NOT_AVAILABLE)); + assert_eq!(response.cluster_snapshot_path, "/rustfs/admin/v4/cluster/snapshot"); + assert_eq!(response.cluster_snapshot_summary, None); + assert_eq!(response.summary.cluster_snapshot.state, CapabilityState::Unknown); assert_eq!(response.observability.platform.os.as_deref(), Some(std::env::consts::OS)); assert_eq!( response @@ -495,6 +555,49 @@ mod tests { assert_eq!(response.workload_admission.entries().len(), WorkloadClass::REQUIRED.len()); } + #[test] + fn server_info_response_exposes_admin_discovery_paths() { + let usecase = DefaultAdminUsecase::without_context(); + let response = ServerInfoResponse { + info: InfoMessage { + mode: None, + domain: None, + region: None, + sqs_arn: None, + deployment_id: None, + buckets: None, + objects: None, + versions: None, + delete_markers: None, + usage: None, + services: None, + backend: None, + servers: None, + pools: None, + }, + admin_discovery: system_admin_discovery(&usecase), + }; + + let value = serde_json::to_value(response).expect("server info response should serialize"); + assert_eq!(value["admin_discovery"]["runtimeCapabilities"], "/rustfs/admin/v4/runtime/capabilities"); + assert_eq!(value["admin_discovery"]["clusterSnapshot"], "/rustfs/admin/v4/cluster/snapshot"); + assert_eq!(value["admin_discovery"]["extensionsCatalog"], "/rustfs/admin/v4/extensions/catalog"); + } + + #[test] + fn storage_info_response_exposes_admin_discovery_paths() { + let usecase = DefaultAdminUsecase::without_context(); + let response = super::StorageInfoResponse { + info: StorageInfo::default(), + admin_discovery: system_admin_discovery(&usecase), + }; + + let value = serde_json::to_value(response).expect("storage info response should serialize"); + assert_eq!(value["admin_discovery"]["runtimeCapabilities"], "/rustfs/admin/v4/runtime/capabilities"); + assert_eq!(value["admin_discovery"]["clusterSnapshot"], "/rustfs/admin/v4/cluster/snapshot"); + assert_eq!(value["admin_discovery"]["extensionsCatalog"], "/rustfs/admin/v4/extensions/catalog"); + } + #[test] fn runtime_capabilities_summary_marks_observability_supported_when_groups_are_resolved() { let observability = ObservabilitySnapshot { @@ -529,7 +632,12 @@ mod tests { }, }; - let summary = build_runtime_capabilities_summary(&observability, Some(&topology), &CapabilityStatus::supported()); + let summary = build_runtime_capabilities_summary( + &observability, + Some(&topology), + &CapabilityStatus::supported(), + Some(&CapabilityStatus::supported().with_reason("cluster snapshot is available")), + ); assert_eq!(summary.observability.state, CapabilityState::Supported); assert_eq!(summary.observability.reason.as_deref(), Some(OBSERVABILITY_SUMMARY_RESOLVED)); @@ -538,6 +646,7 @@ mod tests { assert_eq!(summary.platform.state, CapabilityState::Supported); assert_eq!(summary.topology.state, CapabilityState::Supported); assert_eq!(summary.topology.reason.as_deref(), Some(TOPOLOGY_SUMMARY_RESOLVED)); + assert_eq!(summary.cluster_snapshot.state, CapabilityState::Supported); } #[test] @@ -574,13 +683,19 @@ mod tests { }, }; - let summary = build_runtime_capabilities_summary(&observability, Some(&topology), &CapabilityStatus::supported()); + let summary = build_runtime_capabilities_summary( + &observability, + Some(&topology), + &CapabilityStatus::supported(), + Some(&CapabilityStatus::unknown().with_reason("cluster snapshot unresolved")), + ); assert_eq!(summary.observability.state, CapabilityState::Unknown); assert_eq!(summary.userspace_profiling.state, CapabilityState::Unknown); assert_eq!(summary.memory_sampling.state, CapabilityState::Unknown); assert_eq!(summary.platform.state, CapabilityState::Unknown); assert_eq!(summary.topology.state, CapabilityState::Unknown); + assert_eq!(summary.cluster_snapshot.state, CapabilityState::Unknown); assert!( summary .userspace_profiling diff --git a/rustfs/src/admin/handlers/table_catalog.rs b/rustfs/src/admin/handlers/table_catalog.rs index 6e8cb743a..569e916af 100644 --- a/rustfs/src/admin/handlers/table_catalog.rs +++ b/rustfs/src/admin/handlers/table_catalog.rs @@ -17,6 +17,7 @@ use crate::admin::{ auth::{AdminResourceScope, validate_admin_request, validate_admin_request_with_bucket_object}, router::{AdminOperation, Operation, S3Router}, }; +use crate::app::admin_usecase::DefaultAdminUsecase; use crate::app::context::{resolve_object_store_handle, resolve_token_signing_key}; use crate::auth::{check_key_valid, get_session_token}; use crate::server::{RemoteAddr, TABLE_CATALOG_COMPAT_PREFIX, TABLE_CATALOG_PREFIX}; @@ -190,6 +191,17 @@ struct CatalogConfigResponse { defaults: BTreeMap<&'static str, &'static str>, overrides: BTreeMap<&'static str, &'static str>, endpoints: Vec<&'static str>, + admin_discovery: CatalogAdminDiscovery, +} + +#[derive(Debug, Serialize)] +struct CatalogAdminDiscovery { + #[serde(rename = "runtimeCapabilities")] + runtime_capabilities: &'static str, + #[serde(rename = "clusterSnapshot")] + cluster_snapshot: &'static str, + #[serde(rename = "extensionsCatalog")] + extensions_catalog: &'static str, } #[derive(Debug, Deserialize)] @@ -975,6 +987,7 @@ fn register_table_catalog_prefix_routes(r: &mut S3Router, prefix } fn catalog_config_response() -> CatalogConfigResponse { + let usecase = DefaultAdminUsecase::from_global(); CatalogConfigResponse { defaults: BTreeMap::from([ (WAREHOUSE_PROPERTY, DEFAULT_WAREHOUSE_ID), @@ -983,6 +996,11 @@ fn catalog_config_response() -> CatalogConfigResponse { ]), overrides: BTreeMap::new(), endpoints: TABLE_CATALOG_ENDPOINTS.to_vec(), + admin_discovery: CatalogAdminDiscovery { + runtime_capabilities: usecase.runtime_capabilities_route(), + cluster_snapshot: usecase.cluster_snapshot_route(), + extensions_catalog: usecase.extensions_catalog_route(), + }, } } @@ -5345,6 +5363,9 @@ mod tests { Some(&TABLE_CATALOG_COMPAT_PREFIX) ); assert!(response.overrides.is_empty()); + assert_eq!(response.admin_discovery.runtime_capabilities, "/rustfs/admin/v4/runtime/capabilities"); + assert_eq!(response.admin_discovery.cluster_snapshot, "/rustfs/admin/v4/cluster/snapshot"); + assert_eq!(response.admin_discovery.extensions_catalog, "/rustfs/admin/v4/extensions/catalog"); assert!(response.endpoints.contains(&"GET /v1/{prefix}/namespaces")); assert!(response.endpoints.contains(&"HEAD /v1/{prefix}/namespaces/{namespace}")); assert!( diff --git a/rustfs/src/admin/mod.rs b/rustfs/src/admin/mod.rs index fb7ccb1a5..bb95f4501 100644 --- a/rustfs/src/admin/mod.rs +++ b/rustfs/src/admin/mod.rs @@ -31,9 +31,9 @@ mod console_test; mod route_registration_test; use handlers::{ - audit, bucket_meta, config_admin, extensions, heal, health, kms, module_switch, object_zip_download, oidc, plugins_catalog, - plugins_instances, pools, profile_admin, quota as quota_handler, rebalance, replication as replication_handler, scanner, - site_replication, sts, system, table_catalog, tier, tls_debug, user, + audit, bucket_meta, cluster_snapshot, config_admin, extensions, heal, health, kms, module_switch, object_zip_download, oidc, + plugins_catalog, plugins_instances, pools, profile_admin, quota as quota_handler, rebalance, + replication as replication_handler, scanner, site_replication, sts, system, table_catalog, tier, tls_debug, user, }; use router::{AdminOperation, S3Router}; use s3s::route::S3Route; @@ -70,6 +70,7 @@ fn register_admin_routes(r: &mut S3Router) -> std::io::Result<() scanner::register_scanner_route(r)?; audit::register_audit_target_route(r)?; module_switch::register_module_switch_route(r)?; + cluster_snapshot::register_cluster_snapshot_route(r)?; extensions::register_extension_route(r)?; object_zip_download::register_object_zip_download_route(r)?; plugins_catalog::register_plugin_catalog_route(r)?; diff --git a/rustfs/src/admin/plugin_contract.rs b/rustfs/src/admin/plugin_contract.rs index 6eb2a69ad..b308e260a 100644 --- a/rustfs/src/admin/plugin_contract.rs +++ b/rustfs/src/admin/plugin_contract.rs @@ -300,6 +300,17 @@ pub(crate) struct PluginCatalogEntry { #[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub(crate) struct PluginCatalogResponse { pub plugins: Vec, + pub admin_discovery: PluginCatalogAdminDiscovery, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct PluginCatalogAdminDiscovery { + #[serde(rename = "runtimeCapabilities")] + pub runtime_capabilities: String, + #[serde(rename = "clusterSnapshot")] + pub cluster_snapshot: String, + #[serde(rename = "extensionsCatalog")] + pub extensions_catalog: String, } #[derive(Debug, Clone, PartialEq, Eq, Serialize)] @@ -366,10 +377,10 @@ pub(crate) struct PluginInstancesResponse { #[cfg(test)] mod tests { use super::{ - PluginArtifactContract, PluginCatalogDomainEntry, PluginCatalogEntry, PluginCatalogResponse, PluginContractDomain, - PluginContractEntrypointKind, PluginContractPackaging, PluginDistributionContract, PluginInstanceDetail, - PluginInstanceDiagnostic, PluginInstanceDiagnosticCode, PluginInstanceDiagnosticCount, PluginInstanceEntry, - PluginInstanceSource, PluginInstancesResponse, PluginRuntimeContract, PluginRuntimeTransport, + PluginArtifactContract, PluginCatalogAdminDiscovery, PluginCatalogDomainEntry, PluginCatalogEntry, PluginCatalogResponse, + PluginContractDomain, PluginContractEntrypointKind, PluginContractPackaging, PluginDistributionContract, + PluginInstanceDetail, PluginInstanceDiagnostic, PluginInstanceDiagnosticCode, PluginInstanceDiagnosticCount, + PluginInstanceEntry, PluginInstanceSource, PluginInstancesResponse, PluginRuntimeContract, PluginRuntimeTransport, }; use serde_json::json; use std::collections::HashMap; @@ -400,6 +411,11 @@ mod tests { }], installation: None, }], + admin_discovery: PluginCatalogAdminDiscovery { + runtime_capabilities: "/rustfs/admin/v4/runtime/capabilities".to_string(), + cluster_snapshot: "/rustfs/admin/v4/cluster/snapshot".to_string(), + extensions_catalog: "/rustfs/admin/v4/extensions/catalog".to_string(), + }, }; let value = serde_json::to_value(response).expect("catalog response should serialize"); @@ -427,7 +443,12 @@ mod tests { "subsystem": "notify_webhook", "valid_fields": ["endpoint", "auth_token"] }] - }] + }], + "admin_discovery": { + "runtimeCapabilities": "/rustfs/admin/v4/runtime/capabilities", + "clusterSnapshot": "/rustfs/admin/v4/cluster/snapshot", + "extensionsCatalog": "/rustfs/admin/v4/extensions/catalog" + } }) ); } diff --git a/rustfs/src/admin/route_policy.rs b/rustfs/src/admin/route_policy.rs index 870254e98..e13039926 100644 --- a/rustfs/src/admin/route_policy.rs +++ b/rustfs/src/admin/route_policy.rs @@ -344,6 +344,12 @@ pub const ADMIN_ROUTE_POLICY_SPECS: &[AdminRouteSpec] = &[ RouteRiskLevel::Sensitive, ), admin(HttpMethod::Put, "/rustfs/admin/v3/module-switches", CONFIG_UPDATE, RouteRiskLevel::High), + admin( + HttpMethod::Get, + "/rustfs/admin/v4/cluster/snapshot", + SERVER_INFO, + RouteRiskLevel::Sensitive, + ), admin( HttpMethod::Get, "/rustfs/admin/v4/extensions/catalog", diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index 6fd9d591d..ede96e166 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -225,6 +225,7 @@ fn expected_admin_route_matrix() -> Vec { ), admin_route(Method::GET, "/v3/module-switches"), admin_route(Method::PUT, "/v3/module-switches"), + admin_route(Method::GET, "/v4/cluster/snapshot"), admin_route(Method::GET, "/v4/extensions/catalog"), admin_route(Method::GET, "/v4/extensions/instances"), admin_route(Method::GET, "/v4/runtime/capabilities"), @@ -724,6 +725,7 @@ fn test_register_routes_cover_representative_admin_paths() { assert_route(&router, Method::GET, &admin_path("/v3/audit/target/list")); assert_route(&router, Method::GET, &admin_path("/v3/module-switches")); assert_route(&router, Method::PUT, &admin_path("/v3/module-switches")); + assert_route(&router, Method::GET, &admin_path("/v4/cluster/snapshot")); assert_route(&router, Method::GET, &admin_path("/v4/extensions/catalog")); assert_route(&router, Method::GET, &admin_path("/v4/extensions/instances")); assert_route(&router, Method::GET, &admin_path("/v4/runtime/capabilities")); diff --git a/rustfs/src/app/admin_usecase.rs b/rustfs/src/app/admin_usecase.rs index a54938797..0524b9280 100644 --- a/rustfs/src/app/admin_usecase.rs +++ b/rustfs/src/app/admin_usecase.rs @@ -19,8 +19,11 @@ use super::EndpointServerPools; use super::get_server_info; use super::{PoolDecommissionInfo, PoolStatus, RebalStatus, get_total_usable_capacity, get_total_usable_capacity_free}; use super::{apply_bucket_usage_memory_overlay, load_data_usage_from_backend}; -use crate::app::context::{AppContext, get_global_app_context, resolve_object_store_handle_for_context}; +use crate::app::context::{ + AppContext, get_global_app_context, resolve_endpoints_handle, resolve_object_store_handle_for_context, +}; use crate::capacity::resolve_admin_used_capacity; +use crate::cluster_snapshot::{ClusterReadOnlySnapshot, collect_cluster_read_only_snapshot}; use crate::error::ApiError; use crate::server::{DependencyReadiness, collect_dependency_readiness as collect_runtime_dependency_readiness}; use rustfs_data_usage::DataUsageInfo; @@ -31,6 +34,9 @@ use std::sync::Arc; use tracing::{debug, error, info, warn}; pub type AdminUsecaseResult = Result; +pub const ADMIN_CLUSTER_SNAPSHOT_ROUTE: &str = "/rustfs/admin/v4/cluster/snapshot"; +pub const ADMIN_EXTENSIONS_CATALOG_ROUTE: &str = "/rustfs/admin/v4/extensions/catalog"; +pub const ADMIN_RUNTIME_CAPABILITIES_ROUTE: &str = "/rustfs/admin/v4/runtime/capabilities"; #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] pub struct QueryServerInfoRequest { @@ -524,6 +530,23 @@ impl DefaultAdminUsecase { pub async fn execute_collect_dependency_readiness(&self) -> DependencyReadiness { collect_runtime_dependency_readiness().await } + + pub async fn execute_collect_cluster_read_only_snapshot(&self) -> Option { + let endpoint_pools = resolve_endpoints_handle()?; + collect_cluster_read_only_snapshot(&endpoint_pools).await + } + + pub fn cluster_snapshot_route(&self) -> &'static str { + ADMIN_CLUSTER_SNAPSHOT_ROUTE + } + + pub fn runtime_capabilities_route(&self) -> &'static str { + ADMIN_RUNTIME_CAPABILITIES_ROUTE + } + + pub fn extensions_catalog_route(&self) -> &'static str { + ADMIN_EXTENSIONS_CATALOG_ROUTE + } } #[cfg(test)] @@ -557,6 +580,24 @@ mod tests { let _ = readiness.iam_ready; } + #[tokio::test] + async fn execute_collect_cluster_read_only_snapshot_returns_none_without_context() { + let usecase = DefaultAdminUsecase::without_context(); + + let snapshot = usecase.execute_collect_cluster_read_only_snapshot().await; + + assert!(snapshot.is_none()); + } + + #[test] + fn admin_usecase_exposes_stable_discovery_routes() { + let usecase = DefaultAdminUsecase::without_context(); + + assert_eq!(usecase.cluster_snapshot_route(), "/rustfs/admin/v4/cluster/snapshot"); + assert_eq!(usecase.runtime_capabilities_route(), "/rustfs/admin/v4/runtime/capabilities"); + assert_eq!(usecase.extensions_catalog_route(), "/rustfs/admin/v4/extensions/catalog"); + } + #[test] fn admin_query_pool_status_by_id_rejects_non_numeric_index() { assert_eq!(DefaultAdminUsecase::parse_pool_idx_by_id("pool-a", 4), None); diff --git a/rustfs/src/cluster_snapshot.rs b/rustfs/src/cluster_snapshot.rs new file mode 100644 index 000000000..e2abd1a90 --- /dev/null +++ b/rustfs/src/cluster_snapshot.rs @@ -0,0 +1,304 @@ +// 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 crate::runtime_capabilities::runtime_observability_snapshot; +use crate::server::{ + DependencyReadiness, DependencyReadinessReport, ReadinessDegradedReason, snapshot_dependency_readiness_report, +}; +use crate::storage::EndpointServerPools; +use crate::workload_admission::workload_admission_registry_snapshot; +use rustfs_concurrency::{AdmissionState, WorkloadAdmissionRegistrySnapshot}; +use rustfs_ecstore::api::cluster::{ + ClusterControlPlane, ClusterControlPlaneSnapshot, ClusterLocalNodeStorageSnapshot, ClusterMembershipSnapshot, + ClusterPeerHealthSnapshot, ClusterPoolStateSnapshot, +}; +use rustfs_storage_api::{ObservabilitySnapshot, TopologySnapshot}; + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ClusterReadOnlySnapshot { + pub topology: TopologySnapshot, + pub membership: ClusterMembershipSnapshot, + pub pool_state: ClusterPoolStateSnapshot, + pub local_storage: ClusterLocalNodeStorageSnapshot, + pub peer_health: ClusterPeerHealthSnapshot, + pub observability: ObservabilitySnapshot, + pub workload_admission: WorkloadAdmissionRegistrySnapshot, + pub runtime_status: ClusterRuntimeStatusSnapshot, +} + +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ClusterRuntimeStatusSnapshot { + pub readiness: DependencyReadiness, + pub state: ClusterRuntimeReadinessState, + pub degraded_reasons: Vec, +} + +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub enum ClusterRuntimeReadinessState { + Ready, + Degraded, + Unknown, +} + +impl ClusterRuntimeReadinessState { + fn from_report(report: &DependencyReadinessReport) -> Self { + if report.readiness.storage_ready && report.readiness.iam_ready && report.readiness.lock_quorum_ready { + Self::Ready + } else if report.degraded_reasons.is_empty() { + Self::Unknown + } else { + Self::Degraded + } + } +} + +impl ClusterRuntimeStatusSnapshot { + pub fn from_readiness_report(report: DependencyReadinessReport) -> Self { + let state = ClusterRuntimeReadinessState::from_report(&report); + Self { + readiness: report.readiness, + state, + degraded_reasons: report.degraded_reasons, + } + } +} + +pub fn cluster_read_only_snapshot_from_endpoint_pools( + endpoint_pools: &EndpointServerPools, + runtime_status: ClusterRuntimeStatusSnapshot, +) -> ClusterReadOnlySnapshot { + let control_plane = ClusterControlPlane::new(endpoint_pools.clone()); + let control_plane_snapshot = control_plane.read_snapshot(); + cluster_read_only_snapshot_from_control_plane(control_plane_snapshot, runtime_status) +} + +pub fn cluster_read_only_snapshot_from_control_plane( + control_plane: ClusterControlPlaneSnapshot, + runtime_status: ClusterRuntimeStatusSnapshot, +) -> ClusterReadOnlySnapshot { + ClusterReadOnlySnapshot { + topology: control_plane.topology, + membership: control_plane.membership, + pool_state: control_plane.pool_state, + local_storage: control_plane.local_storage, + peer_health: control_plane.peer_health, + observability: runtime_observability_snapshot(), + workload_admission: workload_admission_registry_snapshot(), + runtime_status, + } +} + +pub async fn collect_cluster_read_only_snapshot(endpoint_pools: &EndpointServerPools) -> Option { + let runtime_status = ClusterRuntimeStatusSnapshot::from_readiness_report(snapshot_dependency_readiness_report().await); + Some(cluster_read_only_snapshot_from_endpoint_pools(endpoint_pools, runtime_status)) +} + +pub fn cluster_has_actionable_pressure(snapshot: &ClusterReadOnlySnapshot) -> bool { + snapshot.runtime_status.state == ClusterRuntimeReadinessState::Degraded + || snapshot + .workload_admission + .entries() + .iter() + .any(|entry| entry.state != AdmissionState::Open) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::storage::{Endpoint, Endpoints, PoolEndpoints}; + use rustfs_concurrency::{WorkloadAdmissionSnapshot, WorkloadClass}; + use rustfs_storage_api::{CapabilityState, CapabilityStatus, DiskCapabilities, TopologyCapabilities}; + + #[test] + fn runtime_status_snapshot_maps_ready_and_degraded_reports() { + let ready = ClusterRuntimeStatusSnapshot::from_readiness_report(DependencyReadinessReport { + readiness: DependencyReadiness { + storage_ready: true, + iam_ready: true, + lock_quorum_ready: true, + }, + degraded_reasons: Vec::new(), + }); + assert_eq!(ready.state, ClusterRuntimeReadinessState::Ready); + + let degraded = ClusterRuntimeStatusSnapshot::from_readiness_report(DependencyReadinessReport { + readiness: DependencyReadiness { + storage_ready: false, + iam_ready: true, + lock_quorum_ready: true, + }, + degraded_reasons: vec![ReadinessDegradedReason::StorageQuorumUnavailable], + }); + assert_eq!(degraded.state, ClusterRuntimeReadinessState::Degraded); + assert_eq!(degraded.degraded_reasons, vec![ReadinessDegradedReason::StorageQuorumUnavailable]); + } + + #[test] + fn cluster_snapshot_preserves_read_only_sections_and_partial_states() { + let endpoint_pools = sample_endpoint_pools(); + let runtime_status = ClusterRuntimeStatusSnapshot::from_readiness_report(DependencyReadinessReport { + readiness: DependencyReadiness { + storage_ready: false, + iam_ready: true, + lock_quorum_ready: false, + }, + degraded_reasons: vec![ReadinessDegradedReason::StorageAndLockUnavailable], + }); + + let snapshot = cluster_read_only_snapshot_from_endpoint_pools(&endpoint_pools, runtime_status); + + assert_eq!(snapshot.topology.pools.len(), 1); + assert_eq!(snapshot.membership.nodes.len(), 2); + assert_eq!(snapshot.pool_state.pools[0].endpoint_count, 4); + assert_eq!(snapshot.local_storage.nodes.len(), 1); + assert_eq!(snapshot.peer_health.peers.len(), 2); + assert_eq!(snapshot.runtime_status.state, ClusterRuntimeReadinessState::Degraded); + assert_eq!( + snapshot.runtime_status.degraded_reasons, + vec![ReadinessDegradedReason::StorageAndLockUnavailable] + ); + assert_eq!(snapshot.peer_health.peers[0].status.state, CapabilityState::Unknown); + if cfg!(target_os = "linux") { + assert_eq!(snapshot.observability.platform.numa.state, CapabilityState::Unknown); + } else { + assert_eq!(snapshot.observability.platform.numa.state, CapabilityState::Unsupported); + } + assert!( + snapshot + .workload_admission + .entries() + .iter() + .any(|entry| entry.state == AdmissionState::Unknown) + ); + } + + #[test] + fn cluster_pressure_helper_detects_degraded_runtime_or_non_open_admission() { + let no_pressure = ClusterReadOnlySnapshot { + topology: TopologySnapshot::default(), + membership: ClusterMembershipSnapshot::default(), + pool_state: ClusterPoolStateSnapshot::default(), + local_storage: ClusterLocalNodeStorageSnapshot::default(), + peer_health: ClusterPeerHealthSnapshot::default(), + observability: ObservabilitySnapshot::default(), + workload_admission: WorkloadAdmissionRegistrySnapshot::new(vec![WorkloadAdmissionSnapshot::new( + WorkloadClass::ForegroundRead, + AdmissionState::Open, + )]), + runtime_status: ClusterRuntimeStatusSnapshot { + readiness: DependencyReadiness { + storage_ready: true, + iam_ready: true, + lock_quorum_ready: true, + }, + state: ClusterRuntimeReadinessState::Ready, + degraded_reasons: Vec::new(), + }, + }; + assert!(!cluster_has_actionable_pressure(&no_pressure)); + + let runtime_pressure = ClusterReadOnlySnapshot { + runtime_status: ClusterRuntimeStatusSnapshot { + readiness: DependencyReadiness { + storage_ready: false, + iam_ready: true, + lock_quorum_ready: true, + }, + state: ClusterRuntimeReadinessState::Degraded, + degraded_reasons: vec![ReadinessDegradedReason::StorageQuorumUnavailable], + }, + ..no_pressure.clone() + }; + assert!(cluster_has_actionable_pressure(&runtime_pressure)); + + let admission_pressure = ClusterReadOnlySnapshot { + workload_admission: WorkloadAdmissionRegistrySnapshot::new(vec![WorkloadAdmissionSnapshot::new( + WorkloadClass::Repair, + AdmissionState::Unknown, + )]), + ..no_pressure + }; + assert!(cluster_has_actionable_pressure(&admission_pressure)); + } + + #[test] + fn topology_capability_states_can_remain_unknown_disabled_and_unsupported() { + let endpoint_pools = sample_endpoint_pools(); + let mut control_plane = ClusterControlPlane::new(endpoint_pools).read_snapshot(); + control_plane.topology.capabilities = TopologyCapabilities { + profiling: CapabilityStatus::disabled().with_reason("disabled for test"), + numa: CapabilityStatus::unsupported().with_reason("unsupported for test"), + failure_domain_labels: CapabilityStatus::unknown().with_reason("unknown for test"), + media_labels: CapabilityStatus::supported(), + }; + + for pool in &mut control_plane.topology.pools { + for set in &mut pool.sets { + for disk in &mut set.disks { + disk.capabilities = DiskCapabilities { + media_type: CapabilityStatus::unknown().with_reason("media unknown"), + failure_domain: CapabilityStatus::unknown().with_reason("fd unknown"), + numa: CapabilityStatus::unsupported().with_reason("numa unsupported"), + profiling: CapabilityStatus::disabled().with_reason("profiling disabled"), + }; + } + } + } + + let snapshot = cluster_read_only_snapshot_from_control_plane( + control_plane, + ClusterRuntimeStatusSnapshot { + readiness: DependencyReadiness { + storage_ready: true, + iam_ready: true, + lock_quorum_ready: true, + }, + state: ClusterRuntimeReadinessState::Ready, + degraded_reasons: Vec::new(), + }, + ); + + assert_eq!(snapshot.topology.capabilities.profiling.state, CapabilityState::Disabled); + assert_eq!(snapshot.topology.capabilities.numa.state, CapabilityState::Unsupported); + assert_eq!(snapshot.topology.capabilities.failure_domain_labels.state, CapabilityState::Unknown); + assert_eq!( + snapshot.topology.pools[0].sets[0].disks[0].capabilities.profiling.state, + CapabilityState::Disabled + ); + } + + fn sample_endpoint_pools() -> EndpointServerPools { + let endpoints = (0..4) + .map(|index| { + let host = if index < 2 { "node1.example" } else { "node2.example" }; + let mut endpoint = + Endpoint::try_from(format!("http://{host}:9000/export{index}").as_str()).expect("url endpoint"); + endpoint.set_pool_index(0); + endpoint.set_set_index(index / 2); + endpoint.set_disk_index(index % 2); + endpoint.is_local = index < 2; + endpoint + }) + .collect::>(); + + EndpointServerPools::from(vec![PoolEndpoints { + legacy: false, + set_count: 2, + drives_per_set: 2, + endpoints: Endpoints::from(endpoints), + cmd_line: "http://node{1...2}.example:9000/export{0...3}".to_owned(), + platform: "OS: test | Arch: test".to_owned(), + }]) + } +} diff --git a/rustfs/src/lib.rs b/rustfs/src/lib.rs index 100f7ac90..d111356c8 100644 --- a/rustfs/src/lib.rs +++ b/rustfs/src/lib.rs @@ -56,6 +56,7 @@ pub mod app; pub mod auth; pub mod auth_keystone; pub mod capacity; +pub mod cluster_snapshot; pub mod config; pub mod delete_tail_activity; pub mod embedded; diff --git a/rustfs/src/server/mod.rs b/rustfs/src/server/mod.rs index efe0f75b8..fe0765749 100644 --- a/rustfs/src/server/mod.rs +++ b/rustfs/src/server/mod.rs @@ -62,6 +62,7 @@ pub(crate) use readiness::ReadinessGateLayer; pub(crate) use readiness::collect_dependency_readiness; pub(crate) use readiness::collect_dependency_readiness_report; pub use readiness::publish_ready_when_runtime_ready; +pub(crate) use readiness::snapshot_dependency_readiness_report; #[derive(Clone, Copy, Debug)] pub struct RemoteAddr(pub std::net::SocketAddr); diff --git a/rustfs/src/server/readiness.rs b/rustfs/src/server/readiness.rs index fbbf36cfc..e9b7f8fff 100644 --- a/rustfs/src/server/readiness.rs +++ b/rustfs/src/server/readiness.rs @@ -433,6 +433,13 @@ fn record_readiness_report(report: &DependencyReadinessReport) { } } +fn dependency_readiness_report_from_readiness(readiness: DependencyReadiness) -> DependencyReadinessReport { + DependencyReadinessReport { + degraded_reasons: degraded_reasons(readiness.storage_ready, readiness.iam_ready, readiness.lock_quorum_ready), + readiness, + } +} + pub async fn collect_dependency_readiness() -> DependencyReadiness { collect_dependency_readiness_report().await.readiness } @@ -453,14 +460,21 @@ pub async fn collect_dependency_readiness_report() -> DependencyReadinessReport iam_ready: iam_ready_raw, lock_quorum_ready: lock_quorum_status.ready, }; - let report = DependencyReadinessReport { - degraded_reasons: degraded_reasons(readiness.storage_ready, iam_ready_raw, readiness.lock_quorum_ready), - readiness, - }; + let report = dependency_readiness_report_from_readiness(readiness); record_readiness_report(&report); report } +pub(crate) async fn snapshot_dependency_readiness_report() -> DependencyReadinessReport { + let readiness = DependencyReadiness { + storage_ready: collect_storage_readiness_uncached().await, + iam_ready: resolve_iam_ready(), + lock_quorum_ready: collect_lock_quorum_status_uncached().await.ready, + }; + + dependency_readiness_report_from_readiness(readiness) +} + async fn collect_lock_quorum_status() -> LockQuorumStatus { if let Some(cached) = load_cached_lock_quorum_status().await { cached @@ -481,10 +495,7 @@ async fn collect_dependency_readiness_uncached() -> DependencyReadiness { iam_ready: iam_ready_raw, lock_quorum_ready: lock_quorum_status.ready, }; - let report = DependencyReadinessReport { - degraded_reasons: degraded_reasons(readiness.storage_ready, iam_ready_raw, readiness.lock_quorum_ready), - readiness, - }; + let report = dependency_readiness_report_from_readiness(readiness); record_readiness_report(&report); report.readiness }