feat(admin): expose cluster snapshot view (#3803)

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/"));

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<AdminOperation>) -> 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<ClusterSnapshotView>,
+}
+
+#[derive(Debug, Clone, PartialEq, Eq, Serialize)]
+pub(crate) struct ClusterSnapshotDiscoveryResponse {
+    pub path: String,
+    pub summary: Option<CapabilityStatus>,
+    pub topology: Option<CapabilityStatus>,
+    pub peer_health: Option<CapabilityStatus>,
+    pub workload_admission: Option<CapabilityStatus>,
+    pub runtime: Option<CapabilityStatus>,
+}
+
+async fn authorize_cluster_snapshot_request(req: &S3Request<Body>) -> 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::<Option<RemoteAddr>>().and_then(|opt| opt.map(|a| a.0)),
+    )
+    .await
+}
+
+fn build_json_response(
+    status: StatusCode,
+    body: &impl Serialize,
+    request_id: Option<&HeaderValue>,
+) -> S3Result<S3Response<(StatusCode, Body)>> {
+    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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
+        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<WorkloadAdmissionView>,
+    pub runtime_status: ClusterRuntimeStatusView,
+    pub actionable_pressure: bool,
+}
+
+impl From<ClusterReadOnlySnapshot> 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<ClusterNodeMembershipView>,
+    pub drives: Vec<ClusterDriveMembershipView>,
+}
+
+impl From<ClusterMembershipSnapshot> 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<usize>,
+}
+
+impl From<ClusterNodeMembership> 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<ClusterDriveMembership> 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<ClusterPoolStateItemView>,
+}
+
+impl From<ClusterPoolStateSnapshot> 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<ClusterPoolState> 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<ClusterLocalNodeStorageView>,
+}
+
+impl From<ClusterLocalNodeStorageSnapshot> 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<usize>,
+    pub drive_count: usize,
+    pub path_drive_count: usize,
+    pub url_drive_count: usize,
+}
+
+impl From<ClusterLocalNodeStorage> 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<ClusterPeerHealthItemView>,
+}
+
+impl From<ClusterPeerHealthSnapshot> 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<ClusterPeerHealth> 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<usize>,
+    pub queued: Option<usize>,
+    pub limit: Option<usize>,
+    pub reason: Option<String>,
+}
+
+#[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<ClusterRuntimeStatusSnapshot> 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<WorkloadAdmissionView> {
+    snapshot.entries().iter().cloned().map(WorkloadAdmissionView::from).collect()
+}
+
+impl From<WorkloadAdmissionSnapshot> 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::<Vec<_>>()
+                .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::<Vec<_>>();
+    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<const N: usize>(
+    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::<Vec<_>>();
+    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::<Vec<_>>();
+    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()
+    }
+}

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<AdminOperation>) -> std::io::Re
 pub(crate) struct ExtensionCatalogResponse {
     pub extensions: Vec<ExtensionSchema>,
     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<ExtensionCatalogResponse,
     extensions.push(target_marketplace_extension_schema(&example.manifest));
     extensions.sort_by(|a, b| a.extension_id.cmp(&b.extension_id));
     let runtime_capabilities = system::build_runtime_capabilities_response().await?;
+    let cluster_snapshot = cluster_snapshot::build_cluster_snapshot_discovery_response().await;
 
     Ok(ExtensionCatalogResponse {
         extensions,
         runtime_capabilities: build_extension_runtime_capabilities_response(&runtime_capabilities),
+        cluster_snapshot,
         external_plugin_flow: TargetPluginExternalFlowGate::default().status(),
     })
 }
@@ -450,6 +453,8 @@ mod tests {
             Some(&runtime_capabilities.summary.userspace_profiling)
         );
         assert!(validate_ops_profiler_contract(&response.runtime_capabilities.ops_profiler.contract).is_ok());
+
+        assert_eq!(response.cluster_snapshot.path, format!("{}{}", ADMIN_PREFIX, "/v4/cluster/snapshot"));
     }
 
     #[test]

rustfs/src/admin/handlers/mod.rs

@@ -16,6 +16,7 @@ pub mod account_info;
 pub mod audit;
 mod audit_runtime_config;
 pub mod bucket_meta;
+pub mod cluster_snapshot;
 pub mod config_admin;
 pub mod event;
 pub mod extensions;
@@ -73,6 +74,7 @@ mod tests {
         let _set_config_handler = config_admin::SetConfigHandler {};
         let _list_audit_targets = audit::ListAuditTargets {};
         let _get_module_switches = module_switch::GetModuleSwitchesHandler {};
+        let _get_cluster_snapshot = cluster_snapshot::GetClusterSnapshotHandler {};
         let _get_extension_catalog = extensions::GetExtensionCatalogHandler {};
         let _list_extension_instances = extensions::ListExtensionInstancesHandler {};
         let _get_plugin_catalog = plugins_catalog::GetPluginCatalogHandler {};

rustfs/src/admin/handlers/module_switch.rs

@@ -16,6 +16,7 @@ use crate::admin::{
     auth::validate_admin_request,
     router::{AdminOperation, Operation, S3Router},
 };
+use crate::app::admin_usecase::DefaultAdminUsecase;
 use crate::auth::{check_key_valid, get_session_token};
 use crate::server::{
     ADMIN_PREFIX, ModuleSwitchSnapshot, ModuleSwitchSource, PersistedModuleSwitches, RemoteAddr, current_module_switch_snapshot,
@@ -62,10 +63,22 @@ struct ModuleSwitchesResponse {
     persisted_audit_enabled: bool,
     notify_source: ModuleSwitchSource,
     audit_source: ModuleSwitchSource,
+    admin_discovery: ModuleSwitchDiscovery,
+}
+
+#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
+struct ModuleSwitchDiscovery {
+    #[serde(rename = "runtimeCapabilities")]
+    runtime_capabilities: &'static str,
+    #[serde(rename = "clusterSnapshot")]
+    cluster_snapshot: &'static str,
+    #[serde(rename = "extensionsCatalog")]
+    extensions_catalog: &'static str,
 }
 
 impl From<ModuleSwitchSnapshot> 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<ModuleSwitchSnapshot> 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)

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()

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<usize> {
     (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");
+    }
 }

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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
@@ -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<CapabilityStatus>,
     pub observability: rustfs_storage_api::ObservabilitySnapshot,
     pub workload_admission: WorkloadAdmissionRegistrySnapshot,
     pub topology: Option<TopologySnapshot>,
@@ -265,9 +307,11 @@ pub struct RuntimeCapabilitiesHandler {}
 
 pub(crate) async fn build_runtime_capabilities_response()
 -> Result<RuntimeCapabilitiesResponse, rustfs_storage_api::CapabilitySnapshotError> {
+    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

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<AdminOperation>, 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!(

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<AdminOperation>) -> 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)?;

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<PluginCatalogEntry>,
+    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"
+                }
             })
         );
     }

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",

rustfs/src/admin/route_registration_test.rs

@@ -225,6 +225,7 @@ fn expected_admin_route_matrix() -> Vec<RouteMatrixEntry> {
         ),
         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"));

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<T> = Result<T, ApiError>;
+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<ClusterReadOnlySnapshot> {
+        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);

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<ReadinessDegradedReason>,
+}
+
+#[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<ClusterReadOnlySnapshot> {
+    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::<Vec<_>>();
+
+        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(),
+        }])
+    }
+}

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;

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);

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
 }