From 62c2f81afdcffaa20bf7f370252a4f7e1b7d4e02 Mon Sep 17 00:00:00 2001 From: houseme Date: Tue, 21 Jul 2026 22:33:10 +0800 Subject: [PATCH] fix(admin): expose cluster diagnostic components (#5090) Add machine-readable storage, listing, usage, and workload component diagnostics to the admin cluster snapshot response while keeping the external S3 API untouched. Co-authored-by: heihutu --- rustfs/src/admin/handlers/cluster_snapshot.rs | 447 +++++++++++++++++- rustfs/src/cluster_snapshot.rs | 82 +++- 2 files changed, 518 insertions(+), 11 deletions(-) diff --git a/rustfs/src/admin/handlers/cluster_snapshot.rs b/rustfs/src/admin/handlers/cluster_snapshot.rs index 18d2431f4..db6cae50c 100644 --- a/rustfs/src/admin/handlers/cluster_snapshot.rs +++ b/rustfs/src/admin/handlers/cluster_snapshot.rs @@ -28,7 +28,7 @@ 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 crate::server::{ADMIN_PREFIX, ReadinessDegradedReason, RemoteAddr}; use http::{HeaderMap, HeaderValue, StatusCode}; use hyper::Method; use matchit::Params; @@ -63,6 +63,7 @@ pub(crate) struct ClusterSnapshotDiscoveryResponse { pub rpc_boundary: Option, pub workload_admission: Option, pub runtime: Option, + pub components: Option, } async fn authorize_cluster_snapshot_request(req: &S3Request) -> S3Result<()> { @@ -119,7 +120,8 @@ pub(crate) async fn build_cluster_snapshot_discovery_response() -> ClusterSnapsh match snapshot { Some(snapshot) => { - let summary = ClusterSnapshotSummary::from(&snapshot); + let components = ClusterComponentStatusView::from_snapshot(&snapshot); + let summary = ClusterSnapshotSummary::from_snapshot_and_components(&snapshot, &components); ClusterSnapshotDiscoveryResponse { path, summary: Some(summary.actionable_pressure.clone()), @@ -128,6 +130,7 @@ pub(crate) async fn build_cluster_snapshot_discovery_response() -> ClusterSnapsh rpc_boundary: Some(summary.rpc_boundary), workload_admission: Some(summary.workload_admission), runtime: Some(summary.runtime), + components: Some(components), } } None => ClusterSnapshotDiscoveryResponse { @@ -138,6 +141,7 @@ pub(crate) async fn build_cluster_snapshot_discovery_response() -> ClusterSnapsh rpc_boundary: None, workload_admission: None, runtime: None, + components: None, }, } } @@ -147,6 +151,7 @@ pub(crate) struct ClusterSnapshotView { pub summary: ClusterSnapshotSummary, pub runtime_capabilities_path: String, pub extensions_catalog_path: String, + pub components: ClusterComponentStatusView, pub topology: TopologySnapshot, pub membership: ClusterMembershipView, pub pool_state: ClusterPoolStateView, @@ -161,11 +166,14 @@ pub(crate) struct ClusterSnapshotView { impl From for ClusterSnapshotView { fn from(snapshot: ClusterReadOnlySnapshot) -> Self { + let components = ClusterComponentStatusView::from_snapshot(&snapshot); + let summary = ClusterSnapshotSummary::from_snapshot_and_components(&snapshot, &components); let actionable_pressure = cluster_has_actionable_pressure(&snapshot); Self { - summary: ClusterSnapshotSummary::from(&snapshot), + summary, runtime_capabilities_path: format!("{}{}", ADMIN_PREFIX, system::RUNTIME_CAPABILITIES_ROUTE_SUFFIX), extensions_catalog_path: format!("{}{}", ADMIN_PREFIX, "/v4/extensions/catalog"), + components, topology: snapshot.topology, membership: ClusterMembershipView::from(snapshot.membership), pool_state: ClusterPoolStateView::from(snapshot.pool_state), @@ -180,12 +188,93 @@ impl From for ClusterSnapshotView { } } +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct ClusterComponentStatusView { + pub storage: ClusterComponentStatus, + pub peer_health: ClusterComponentStatus, + pub listing: ClusterListingMetacacheStatus, + pub usage: ClusterUsageFreshnessStatus, + pub workload_admission: ClusterComponentStatus, +} + +impl ClusterComponentStatusView { + fn from_snapshot(snapshot: &ClusterReadOnlySnapshot) -> Self { + Self { + storage: component_status_with_condition( + "runtime_readiness", + summarize_storage_readiness(snapshot), + if snapshot.runtime_status.readiness.storage_ready { + "healthy" + } else { + "degraded" + }, + ), + peer_health: component_status("cluster_peer_health", summarize_peer_health(snapshot)), + listing: summarize_listing_metacache(snapshot), + usage: summarize_usage_freshness(snapshot), + workload_admission: component_status("workload_admission", summarize_workload_admission(snapshot)), + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct ClusterComponentStatus { + pub source: &'static str, + pub condition: &'static str, + pub status: CapabilityStatus, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct ClusterListingMetacacheStatus { + pub source: &'static str, + pub condition: &'static str, + pub status: CapabilityStatus, + pub internode_stall_timeouts_total: u64, + pub hint: &'static str, +} + +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +pub(crate) struct ClusterUsageFreshnessStatus { + pub source: &'static str, + pub condition: &'static str, + pub status: CapabilityStatus, + pub dirty_pending_buckets: u64, + pub last_dirty_mark_unix_secs: u64, + pub last_dirty_clear_unix_secs: u64, + pub last_cycle_dirty_buckets: u64, + pub last_cycle_cleared_dirty_buckets: u64, + pub last_usage_save_unix_secs: u64, + pub last_usage_save_result: String, + pub last_success_unix_secs: Option, + pub last_error: Option, +} + +fn component_status(source: &'static str, status: CapabilityStatus) -> ClusterComponentStatus { + let condition = condition_from_capability(&status); + component_status_with_condition(source, status, condition) +} + +fn component_status_with_condition( + source: &'static str, + status: CapabilityStatus, + condition: &'static str, +) -> ClusterComponentStatus { + ClusterComponentStatus { + source, + condition, + status, + } +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub(crate) struct ClusterSnapshotSummary { pub runtime: CapabilityStatus, pub topology: CapabilityStatus, pub membership: CapabilityStatus, + pub storage: CapabilityStatus, pub peer_health: CapabilityStatus, + pub listing: CapabilityStatus, + pub usage: CapabilityStatus, pub rpc_boundary: CapabilityStatus, pub observability: CapabilityStatus, pub workload_admission: CapabilityStatus, @@ -194,12 +283,17 @@ pub(crate) struct ClusterSnapshotSummary { impl From<&ClusterReadOnlySnapshot> for ClusterSnapshotSummary { fn from(snapshot: &ClusterReadOnlySnapshot) -> Self { + let components = ClusterComponentStatusView::from_snapshot(snapshot); + Self::from_snapshot_and_components(snapshot, &components) + } +} + +impl ClusterSnapshotSummary { + fn from_snapshot_and_components(snapshot: &ClusterReadOnlySnapshot, components: &ClusterComponentStatusView) -> Self { let topology = summarize_topology(snapshot); let membership = summarize_membership(snapshot); - let peer_health = summarize_peer_health(snapshot); let rpc_boundary = summarize_rpc_boundary(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 { @@ -210,10 +304,13 @@ impl From<&ClusterReadOnlySnapshot> for ClusterSnapshotSummary { runtime: summarize_runtime(snapshot), topology, membership, - peer_health, + storage: components.storage.status.clone(), + peer_health: components.peer_health.status.clone(), + listing: components.listing.status.clone(), + usage: components.usage.status.clone(), rpc_boundary, observability, - workload_admission, + workload_admission: components.workload_admission.status.clone(), actionable_pressure, } } @@ -511,6 +608,130 @@ fn runtime_readiness_state_label(state: ClusterRuntimeReadinessState) -> &'stati } } +fn condition_from_capability(status: &CapabilityStatus) -> &'static str { + match status.state { + CapabilityState::Supported => "healthy", + CapabilityState::Unsupported => "unsupported", + CapabilityState::Disabled => "not_reported", + CapabilityState::Unknown => "unknown", + } +} + +fn summarize_storage_readiness(snapshot: &ClusterReadOnlySnapshot) -> CapabilityStatus { + if snapshot.runtime_status.readiness.storage_ready { + return CapabilityStatus::supported().with_reason("storage readiness reports ready"); + } + + let storage_reasons = snapshot + .runtime_status + .degraded_reasons + .iter() + .filter_map(|reason| match reason { + ReadinessDegradedReason::StorageQuorumUnavailable + | ReadinessDegradedReason::StorageAndIamUnavailable + | ReadinessDegradedReason::StorageAndLockUnavailable + | ReadinessDegradedReason::StorageIamAndLockUnavailable => Some(reason.as_str()), + _ => None, + }) + .collect::>(); + if storage_reasons.is_empty() { + CapabilityStatus::unknown().with_reason("storage readiness is false without a storage-specific degraded reason") + } else { + CapabilityStatus::unknown().with_reason(format!("storage readiness degraded: {}", storage_reasons.join(", "))) + } +} + +fn summarize_listing_readiness(snapshot: &ClusterReadOnlySnapshot) -> CapabilityStatus { + match snapshot + .workload_admission + .entries() + .iter() + .find(|entry| entry.class == WorkloadClass::ForegroundRead) + { + Some(entry) if entry.state == AdmissionState::Open => { + CapabilityStatus::supported().with_reason("foreground read admission is open") + } + Some(entry) => { + let mut reason = format!("foreground read admission is {}", admission_state_label(entry.state)); + if let Some(detail) = entry.reason.as_deref() { + reason.push_str(": "); + reason.push_str(detail); + } + CapabilityStatus::unknown().with_reason(reason) + } + None => CapabilityStatus::unknown().with_reason("foreground read admission is not reported"), + } +} + +fn summarize_listing_metacache(snapshot: &ClusterReadOnlySnapshot) -> ClusterListingMetacacheStatus { + const HINT: &str = "bucket and prefix attribution is emitted through metacache quorum logs; inspect operation-labelled walk_dir metrics for internode traffic"; + + let admission = summarize_listing_readiness(snapshot); + let stalls = snapshot.listing_diagnostics.internode_stall_timeouts_total; + let (condition, status) = if admission.state == CapabilityState::Supported { + ("healthy", admission) + } else { + (condition_from_capability(&admission), admission) + }; + + ClusterListingMetacacheStatus { + source: "workload_admission+internode_metrics", + condition, + status, + internode_stall_timeouts_total: stalls, + hint: HINT, + } +} + +fn summarize_usage_freshness(snapshot: &ClusterReadOnlySnapshot) -> ClusterUsageFreshnessStatus { + let freshness = &snapshot.usage_freshness; + let (condition, status) = match freshness.last_usage_save_result.as_str() { + "success" if freshness.dirty_pending_buckets == 0 => ( + "healthy", + CapabilityStatus::supported().with_reason("usage cache was saved successfully and has no pending dirty buckets"), + ), + "success" | "" if freshness.dirty_pending_buckets > 0 => ( + "stale", + CapabilityStatus::unknown() + .with_reason(format!("usage cache has {} pending dirty buckets", freshness.dirty_pending_buckets)), + ), + "skipped_stale" => ( + "stale", + CapabilityStatus::unknown().with_reason("last usage cache save was skipped because scanner data was stale"), + ), + "failed" => ("degraded", CapabilityStatus::unknown().with_reason("last usage cache save failed")), + "encode_failed" => ( + "degraded", + CapabilityStatus::unknown().with_reason("last usage cache save failed during encoding"), + ), + _ => ( + "unknown", + CapabilityStatus::unknown().with_reason("no usage cache save result has been reported"), + ), + }; + let last_success_unix_secs = (freshness.last_usage_save_result == "success" && freshness.last_usage_save_unix_secs > 0) + .then_some(freshness.last_usage_save_unix_secs); + let last_error = match freshness.last_usage_save_result.as_str() { + "failed" | "skipped_stale" | "encode_failed" => Some(freshness.last_usage_save_result.clone()), + _ => None, + }; + + ClusterUsageFreshnessStatus { + source: "scanner_metrics", + condition, + status, + dirty_pending_buckets: freshness.dirty_pending_buckets, + last_dirty_mark_unix_secs: freshness.last_dirty_mark_unix_secs, + last_dirty_clear_unix_secs: freshness.last_dirty_clear_unix_secs, + last_cycle_dirty_buckets: freshness.last_cycle_dirty_buckets, + last_cycle_cleared_dirty_buckets: freshness.last_cycle_cleared_dirty_buckets, + last_usage_save_unix_secs: freshness.last_usage_save_unix_secs, + last_usage_save_result: freshness.last_usage_save_result.clone(), + last_success_unix_secs, + last_error, + } +} + fn summarize_runtime(snapshot: &ClusterReadOnlySnapshot) -> CapabilityStatus { match snapshot.runtime_status.state { ClusterRuntimeReadinessState::Ready => CapabilityStatus::supported().with_reason("runtime readiness reports ready"), @@ -678,7 +899,10 @@ mod tests { ClusterMembershipSnapshot, ClusterNodeMembership, ClusterPeerHealth, ClusterPeerHealthSnapshot, ClusterPoolState, ClusterPoolStateSnapshot, ClusterRpcBoundarySnapshot, ClusterRpcChannelSnapshot, ClusterRpcPlane, ClusterRpcTransport, }; - use crate::cluster_snapshot::{ClusterReadOnlySnapshot, ClusterRuntimeReadinessState, ClusterRuntimeStatusSnapshot}; + use crate::cluster_snapshot::{ + ClusterListingDiagnosticsSnapshot, ClusterReadOnlySnapshot, ClusterRuntimeReadinessState, ClusterRuntimeStatusSnapshot, + ClusterUsageFreshnessSnapshot, + }; use crate::server::{DependencyReadiness, ReadinessDegradedReason}; use rustfs_concurrency::{AdmissionState, WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshot, WorkloadClass}; @@ -716,6 +940,7 @@ mod tests { assert_eq!(response.rpc_boundary, None); assert_eq!(response.workload_admission, None); assert_eq!(response.runtime, None); + assert_eq!(response.components, None); } #[test] @@ -783,11 +1008,29 @@ mod tests { state: ClusterRuntimeReadinessState::Degraded, degraded_reasons: vec![ReadinessDegradedReason::StorageAndLockUnavailable], }, + usage_freshness: ClusterUsageFreshnessSnapshot { + dirty_pending_buckets: 3, + last_usage_save_unix_secs: 123, + last_usage_save_result: "skipped_stale".to_string(), + last_usage_save_result_code: 3, + ..Default::default() + }, + listing_diagnostics: ClusterListingDiagnosticsSnapshot { + internode_stall_timeouts_total: 2, + }, }; 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["components"]["storage"]["source"], "runtime_readiness"); + assert_eq!(value["components"]["storage"]["condition"], "degraded"); + assert_eq!(value["components"]["peer_health"]["source"], "cluster_peer_health"); + assert_eq!(value["components"]["listing"]["source"], "workload_admission+internode_metrics"); + assert_eq!(value["components"]["listing"]["condition"], "unknown"); + assert_eq!(value["components"]["listing"]["internode_stall_timeouts_total"], 2); + assert_eq!(value["components"]["usage"]["source"], "scanner_metrics"); + assert_eq!(value["components"]["usage"]["condition"], "stale"); 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"); @@ -825,17 +1068,203 @@ mod tests { state: ClusterRuntimeReadinessState::Ready, degraded_reasons: Vec::new(), }, + usage_freshness: ClusterUsageFreshnessSnapshot::default(), + listing_diagnostics: ClusterListingDiagnosticsSnapshot::default(), }; let summary = ClusterSnapshotSummary::from(&snapshot); assert_eq!(summary.runtime.state, CapabilityState::Supported); assert_eq!(summary.membership.state, CapabilityState::Unknown); + assert_eq!(summary.storage.state, CapabilityState::Supported); + assert_eq!(summary.storage.reason.as_deref(), Some("storage readiness reports ready")); assert_eq!(summary.peer_health.state, CapabilityState::Unknown); + assert_eq!(summary.listing.state, CapabilityState::Supported); + assert_eq!(summary.listing.reason.as_deref(), Some("foreground read admission is open")); + assert_eq!(summary.usage.state, CapabilityState::Unknown); + assert_eq!(summary.usage.reason.as_deref(), Some("no usage cache save result has been reported")); assert_eq!(summary.rpc_boundary.state, CapabilityState::Supported); assert_eq!(summary.workload_admission.state, CapabilityState::Supported); assert_eq!(summary.actionable_pressure.state, CapabilityState::Disabled); } + #[test] + fn cluster_snapshot_components_split_storage_peer_listing_and_usage_states() { + 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::new(), + }, + pool_state: ClusterPoolStateSnapshot::default(), + local_storage: ClusterLocalNodeStorageSnapshot::default(), + peer_health: ClusterPeerHealthSnapshot { + peers: vec![ClusterPeerHealth { + node_id: "node-a".to_string(), + is_local: true, + status: CapabilityStatus::disabled().with_reason("peer health not reported by endpoints"), + }], + }, + rpc_boundary: sample_rpc_boundary_snapshot(), + observability: ObservabilitySnapshot::default(), + workload_admission: WorkloadAdmissionRegistrySnapshot::new(vec![ + WorkloadAdmissionSnapshot::new(WorkloadClass::ForegroundRead, AdmissionState::Throttled) + .with_counts(Some(4), Some(8), Some(4)) + .with_reason("metacache listing backpressure"), + ]), + runtime_status: ClusterRuntimeStatusSnapshot { + readiness: DependencyReadiness { + storage_ready: false, + iam_ready: true, + lock_quorum_ready: true, + peer_health_ready: true, + }, + state: ClusterRuntimeReadinessState::Degraded, + degraded_reasons: vec![ReadinessDegradedReason::StorageQuorumUnavailable], + }, + usage_freshness: ClusterUsageFreshnessSnapshot { + dirty_pending_buckets: 3, + last_usage_save_unix_secs: 123, + last_usage_save_result: "skipped_stale".to_string(), + last_usage_save_result_code: 3, + ..Default::default() + }, + listing_diagnostics: ClusterListingDiagnosticsSnapshot { + internode_stall_timeouts_total: 0, + }, + }; + + let view = ClusterSnapshotView::from(snapshot); + + assert_eq!(view.components.storage.status.state, CapabilityState::Unknown); + assert_eq!( + view.components.storage.status.reason.as_deref(), + Some("storage readiness degraded: storage_quorum_unavailable") + ); + assert_eq!(view.components.storage.condition, "degraded"); + assert_eq!(view.components.peer_health.status.state, CapabilityState::Disabled); + assert_eq!(view.components.peer_health.condition, "not_reported"); + assert_eq!(view.components.listing.status.state, CapabilityState::Unknown); + assert_eq!( + view.components.listing.status.reason.as_deref(), + Some("foreground read admission is throttled: metacache listing backpressure") + ); + assert_eq!(view.components.listing.condition, "unknown"); + assert_eq!(view.components.usage.status.state, CapabilityState::Unknown); + assert_eq!(view.components.usage.condition, "stale"); + assert_eq!(view.components.usage.last_usage_save_unix_secs, 123); + assert_eq!(view.components.usage.last_usage_save_result, "skipped_stale"); + } + + #[test] + fn cluster_snapshot_listing_component_keeps_historical_stalls_as_evidence() { + let snapshot = ClusterReadOnlySnapshot { + topology: TopologySnapshot::default(), + membership: ClusterMembershipSnapshot::default(), + pool_state: ClusterPoolStateSnapshot::default(), + local_storage: ClusterLocalNodeStorageSnapshot::default(), + peer_health: ClusterPeerHealthSnapshot::default(), + rpc_boundary: ClusterRpcBoundarySnapshot::default(), + observability: ObservabilitySnapshot::default(), + workload_admission: WorkloadAdmissionRegistrySnapshot::new(vec![WorkloadAdmissionSnapshot::new( + WorkloadClass::ForegroundRead, + AdmissionState::Open, + )]), + runtime_status: ClusterRuntimeStatusSnapshot { + readiness: DependencyReadiness::default(), + state: ClusterRuntimeReadinessState::Unknown, + degraded_reasons: Vec::new(), + }, + usage_freshness: ClusterUsageFreshnessSnapshot::default(), + listing_diagnostics: ClusterListingDiagnosticsSnapshot { + internode_stall_timeouts_total: 2, + }, + }; + + let component = super::summarize_listing_metacache(&snapshot); + + assert_eq!(component.status.state, CapabilityState::Supported); + assert_eq!(component.status.reason.as_deref(), Some("foreground read admission is open")); + assert_eq!(component.condition, "healthy"); + assert_eq!(component.internode_stall_timeouts_total, 2); + } + + #[test] + fn cluster_component_condition_preserves_capability_states() { + assert_eq!(super::condition_from_capability(&CapabilityStatus::supported()), "healthy"); + assert_eq!(super::condition_from_capability(&CapabilityStatus::unsupported()), "unsupported"); + assert_eq!(super::condition_from_capability(&CapabilityStatus::disabled()), "not_reported"); + assert_eq!(super::condition_from_capability(&CapabilityStatus::unknown()), "unknown"); + } + + #[test] + fn cluster_snapshot_usage_component_reports_clean_success() { + let snapshot = ClusterReadOnlySnapshot { + topology: TopologySnapshot::default(), + membership: ClusterMembershipSnapshot::default(), + pool_state: ClusterPoolStateSnapshot::default(), + local_storage: ClusterLocalNodeStorageSnapshot::default(), + peer_health: ClusterPeerHealthSnapshot::default(), + rpc_boundary: ClusterRpcBoundarySnapshot::default(), + observability: ObservabilitySnapshot::default(), + workload_admission: WorkloadAdmissionRegistrySnapshot::default(), + runtime_status: ClusterRuntimeStatusSnapshot { + readiness: DependencyReadiness::default(), + state: ClusterRuntimeReadinessState::Unknown, + degraded_reasons: Vec::new(), + }, + usage_freshness: ClusterUsageFreshnessSnapshot { + dirty_pending_buckets: 0, + last_usage_save_unix_secs: 456, + last_usage_save_result: "success".to_string(), + last_usage_save_result_code: 1, + ..Default::default() + }, + listing_diagnostics: ClusterListingDiagnosticsSnapshot::default(), + }; + + let component = super::summarize_usage_freshness(&snapshot); + + assert_eq!(component.status.state, CapabilityState::Supported); + assert_eq!(component.condition, "healthy"); + assert_eq!(component.last_usage_save_unix_secs, 456); + assert_eq!(component.last_usage_save_result, "success"); + } + + #[test] + fn cluster_snapshot_usage_component_reports_dirty_without_save_result_as_stale() { + let snapshot = ClusterReadOnlySnapshot { + topology: TopologySnapshot::default(), + membership: ClusterMembershipSnapshot::default(), + pool_state: ClusterPoolStateSnapshot::default(), + local_storage: ClusterLocalNodeStorageSnapshot::default(), + peer_health: ClusterPeerHealthSnapshot::default(), + rpc_boundary: ClusterRpcBoundarySnapshot::default(), + observability: ObservabilitySnapshot::default(), + workload_admission: WorkloadAdmissionRegistrySnapshot::default(), + runtime_status: ClusterRuntimeStatusSnapshot { + readiness: DependencyReadiness::default(), + state: ClusterRuntimeReadinessState::Unknown, + degraded_reasons: Vec::new(), + }, + usage_freshness: ClusterUsageFreshnessSnapshot { + dirty_pending_buckets: 2, + ..Default::default() + }, + listing_diagnostics: ClusterListingDiagnosticsSnapshot::default(), + }; + + let component = super::summarize_usage_freshness(&snapshot); + + assert_eq!(component.status.state, CapabilityState::Unknown); + assert_eq!(component.condition, "stale"); + assert_eq!(component.status.reason.as_deref(), Some("usage cache has 2 pending dirty buckets")); + } + #[test] fn cluster_snapshot_summary_treats_not_reported_peer_health_as_disabled() { let snapshot = ClusterReadOnlySnapshot { @@ -874,6 +1303,8 @@ mod tests { state: ClusterRuntimeReadinessState::Ready, degraded_reasons: Vec::new(), }, + usage_freshness: ClusterUsageFreshnessSnapshot::default(), + listing_diagnostics: ClusterListingDiagnosticsSnapshot::default(), }; let summary = ClusterSnapshotSummary::from(&snapshot); diff --git a/rustfs/src/cluster_snapshot.rs b/rustfs/src/cluster_snapshot.rs index 718060b09..ec2994526 100644 --- a/rustfs/src/cluster_snapshot.rs +++ b/rustfs/src/cluster_snapshot.rs @@ -24,7 +24,9 @@ use crate::storage_api::cluster::control_plane::{ ClusterPeerHealthSnapshot, ClusterPoolStateSnapshot, ClusterRpcBoundarySnapshot, }; use crate::workload_admission::workload_admission_registry_snapshot; +use rustfs_common::metrics::{ScannerMetricsReport, global_metrics}; use rustfs_concurrency::{AdmissionState, WorkloadAdmissionRegistrySnapshot}; +use rustfs_io_metrics::internode_metrics::{InternodeMetricsSnapshot, global_internode_metrics}; #[derive(Debug, Clone, PartialEq, Eq)] pub struct ClusterReadOnlySnapshot { @@ -37,6 +39,8 @@ pub struct ClusterReadOnlySnapshot { pub observability: ObservabilitySnapshot, pub workload_admission: WorkloadAdmissionRegistrySnapshot, pub runtime_status: ClusterRuntimeStatusSnapshot, + pub usage_freshness: ClusterUsageFreshnessSnapshot, + pub listing_diagnostics: ClusterListingDiagnosticsSnapshot, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -53,6 +57,46 @@ pub enum ClusterRuntimeReadinessState { Unknown, } +#[derive(Debug, Clone, Default, PartialEq, Eq)] +pub struct ClusterUsageFreshnessSnapshot { + pub dirty_pending_buckets: u64, + pub last_dirty_mark_unix_secs: u64, + pub last_dirty_clear_unix_secs: u64, + pub last_cycle_dirty_buckets: u64, + pub last_cycle_cleared_dirty_buckets: u64, + pub last_usage_save_unix_secs: u64, + pub last_usage_save_result: String, + pub last_usage_save_result_code: u64, +} + +impl From<&ScannerMetricsReport> for ClusterUsageFreshnessSnapshot { + fn from(report: &ScannerMetricsReport) -> Self { + Self { + dirty_pending_buckets: report.usage_freshness.dirty_pending_buckets, + last_dirty_mark_unix_secs: report.usage_freshness.last_dirty_mark_unix_secs, + last_dirty_clear_unix_secs: report.usage_freshness.last_dirty_clear_unix_secs, + last_cycle_dirty_buckets: report.usage_freshness.last_cycle_dirty_buckets, + last_cycle_cleared_dirty_buckets: report.usage_freshness.last_cycle_cleared_dirty_buckets, + last_usage_save_unix_secs: report.usage_freshness.last_usage_save_unix_secs, + last_usage_save_result: report.usage_freshness.last_usage_save_result.clone(), + last_usage_save_result_code: report.usage_freshness.last_usage_save_result_code, + } + } +} + +#[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] +pub struct ClusterListingDiagnosticsSnapshot { + pub internode_stall_timeouts_total: u64, +} + +impl From for ClusterListingDiagnosticsSnapshot { + fn from(snapshot: InternodeMetricsSnapshot) -> Self { + Self { + internode_stall_timeouts_total: snapshot.operation_stall_timeouts_total, + } + } +} + impl ClusterRuntimeReadinessState { fn from_report(report: &DependencyReadinessReport) -> Self { if report.readiness.storage_ready @@ -103,12 +147,30 @@ pub fn cluster_read_only_snapshot_from_control_plane( observability: runtime_observability_snapshot(), workload_admission: workload_admission_registry_snapshot(), runtime_status, + usage_freshness: ClusterUsageFreshnessSnapshot::default(), + listing_diagnostics: ClusterListingDiagnosticsSnapshot::default(), } } 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)) + let mut snapshot = cluster_read_only_snapshot_from_endpoint_pools(endpoint_pools, runtime_status); + snapshot.usage_freshness = current_usage_freshness_snapshot().await; + snapshot.listing_diagnostics = current_listing_diagnostics_snapshot(); + Some(snapshot) +} + +async fn current_usage_freshness_snapshot() -> ClusterUsageFreshnessSnapshot { + let report = match crate::runtime_sources::current_scanner_metrics_report().await { + Some(report) => report, + None => global_metrics().report().await, + }; + ClusterUsageFreshnessSnapshot::from(&report) +} + +fn current_listing_diagnostics_snapshot() -> ClusterListingDiagnosticsSnapshot { + let metrics = crate::runtime_sources::current_internode_metrics().unwrap_or_else(|| global_internode_metrics().clone()); + ClusterListingDiagnosticsSnapshot::from(metrics.snapshot()) } pub fn cluster_has_actionable_pressure(snapshot: &ClusterReadOnlySnapshot) -> bool { @@ -181,8 +243,20 @@ mod tests { snapshot.runtime_status.degraded_reasons, vec![ReadinessDegradedReason::StorageAndLockUnavailable] ); - // Local node (peers[0]) is supported — no peer health probing needed. - assert_eq!(snapshot.peer_health.peers[0].status.state, CapabilityState::Supported); + assert!( + snapshot + .peer_health + .peers + .iter() + .any(|peer| peer.is_local && peer.status.state == CapabilityState::Supported) + ); + assert!( + snapshot + .peer_health + .peers + .iter() + .any(|peer| !peer.is_local && peer.status.state == CapabilityState::Disabled) + ); if cfg!(target_os = "linux") { assert_eq!(snapshot.observability.platform.numa.state, CapabilityState::Unknown); } else { @@ -221,6 +295,8 @@ mod tests { state: ClusterRuntimeReadinessState::Ready, degraded_reasons: Vec::new(), }, + usage_freshness: ClusterUsageFreshnessSnapshot::default(), + listing_diagnostics: ClusterListingDiagnosticsSnapshot::default(), }; assert!(!cluster_has_actionable_pressure(&no_pressure));