From 7ca69eb39cff7c0a20d4c4127eaa91ea20ab6b08 Mon Sep 17 00:00:00 2001 From: cxymds Date: Mon, 10 Aug 2026 20:32:01 +0800 Subject: [PATCH] fix: correct SNSD cluster diagnostics (#5930) --- crates/scanner/src/lib.rs | 39 ++++++++++ crates/scanner/src/scanner.rs | 4 +- rustfs/src/admin/handlers/cluster_snapshot.rs | 71 +++++++++++++++---- rustfs/src/workload_admission.rs | 50 ++++++++----- 4 files changed, 134 insertions(+), 30 deletions(-) diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index 02758948d..6b6deb29d 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -82,6 +82,7 @@ pub use storage_api::ScannerReplicationConfig as ReplicationConfig; pub use storage_api::scan::SCANNER_ACTIVITY_PROTOCOL_VERSION; static SCANNER_ACTIVE_WORK_UNITS: AtomicU64 = AtomicU64::new(0); +static SCANNER_RUNTIME_INSTANCES: AtomicU64 = AtomicU64::new(0); static SCANNER_FOREGROUND_READ_ACTIVITY: AtomicU64 = AtomicU64::new(0); static SCANNER_FOREGROUND_STREAM_READS: AtomicU64 = AtomicU64::new(0); @@ -89,6 +90,10 @@ pub fn current_scanner_activity() -> u64 { SCANNER_ACTIVE_WORK_UNITS.load(Ordering::Relaxed) } +pub fn scanner_runtime_initialized() -> bool { + SCANNER_RUNTIME_INSTANCES.load(Ordering::Relaxed) > 0 +} + pub fn set_foreground_read_activity(active: usize) { let active = u64::try_from(active).unwrap_or(u64::MAX); SCANNER_FOREGROUND_READ_ACTIVITY.store(active, Ordering::Relaxed); @@ -138,6 +143,26 @@ impl ScannerActivityGuard { } } +pub(crate) struct ScannerRuntimeGuard; + +impl ScannerRuntimeGuard { + pub(crate) fn new() -> Self { + SCANNER_RUNTIME_INSTANCES.fetch_add(1, Ordering::Relaxed); + Self + } +} + +impl Drop for ScannerRuntimeGuard { + fn drop(&mut self) { + let _ = SCANNER_RUNTIME_INSTANCES.fetch_update(Ordering::Relaxed, Ordering::Relaxed, |current| current.checked_sub(1)); + } +} + +#[cfg(test)] +fn reset_scanner_runtime_instances_for_test() { + SCANNER_RUNTIME_INSTANCES.store(0, Ordering::Relaxed); +} + impl Drop for ScannerActivityGuard { fn drop(&mut self) { let _ = SCANNER_ACTIVE_WORK_UNITS @@ -559,4 +584,18 @@ mod tests { set_foreground_read_activity(0); assert_eq!(current_foreground_read_activity(), 1); } + + #[test] + #[serial] + fn scanner_runtime_guard_tracks_runtime_lifetime() { + reset_scanner_runtime_instances_for_test(); + assert!(!scanner_runtime_initialized()); + + { + let _guard = ScannerRuntimeGuard::new(); + assert!(scanner_runtime_initialized()); + } + + assert!(!scanner_runtime_initialized()); + } } diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 162f2ff32..59cd21e7e 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -34,7 +34,7 @@ use crate::scanner_io::{ scanner_dirty_usage_state, scanner_maintenance_changed, scanner_maintenance_generation, }; use crate::sleeper::{SCANNER_SLEEPER, set_scanner_default_speed}; -use crate::{DataUsageInfo, ScannerActivityGuard, ScannerError}; +use crate::{DataUsageInfo, ScannerActivityGuard, ScannerError, ScannerRuntimeGuard}; use crate::{ScannerConfigObjectDelete, ScannerObjectIO, ScannerObjectOptions}; use bytes::Bytes; use chrono::{DateTime, Utc}; @@ -1312,7 +1312,9 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc) { let replication_active = startup_features.replication; let ctx_clone = ctx; let storeapi_clone = storeapi; + let runtime_guard = ScannerRuntimeGuard::new(); tokio::spawn(async move { + let _runtime_guard = runtime_guard; let (usage_cache_is_cold, has_buckets) = initial_scanner_startup_usage_state(&storeapi_clone).await; let sleep_time = initial_scanner_delay_for_startup( scanner_start_delay().map(|duration| duration.as_secs()), diff --git a/rustfs/src/admin/handlers/cluster_snapshot.rs b/rustfs/src/admin/handlers/cluster_snapshot.rs index db6cae50c..76ad4da62 100644 --- a/rustfs/src/admin/handlers/cluster_snapshot.rs +++ b/rustfs/src/admin/handlers/cluster_snapshot.rs @@ -105,10 +105,9 @@ pub struct GetClusterSnapshotHandler {} impl Operation for GetClusterSnapshotHandler { async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { authorize_cluster_snapshot_request(&req).await?; - let snapshot = default_admin_usecase() - .execute_collect_cluster_read_only_snapshot() - .await - .map(ClusterSnapshotView::from); + let snapshot = default_admin_usecase().execute_collect_cluster_read_only_snapshot().await; + let server_info_endpoint = crate::runtime_sources::current_local_node_name().await; + let snapshot = snapshot.map(|snapshot| ClusterSnapshotView::from_snapshot(snapshot, server_info_endpoint)); build_json_response(StatusCode::OK, &ClusterSnapshotResponse { snapshot }, req.headers.get("x-request-id")) } } @@ -166,6 +165,12 @@ pub(crate) struct ClusterSnapshotView { impl From for ClusterSnapshotView { fn from(snapshot: ClusterReadOnlySnapshot) -> Self { + Self::from_snapshot(snapshot, None) + } +} + +impl ClusterSnapshotView { + fn from_snapshot(snapshot: ClusterReadOnlySnapshot, server_info_endpoint: Option) -> Self { let components = ClusterComponentStatusView::from_snapshot(&snapshot); let summary = ClusterSnapshotSummary::from_snapshot_and_components(&snapshot, &components); let actionable_pressure = cluster_has_actionable_pressure(&snapshot); @@ -175,7 +180,7 @@ impl From for ClusterSnapshotView { extensions_catalog_path: format!("{}{}", ADMIN_PREFIX, "/v4/extensions/catalog"), components, topology: snapshot.topology, - membership: ClusterMembershipView::from(snapshot.membership), + membership: ClusterMembershipView::from_snapshot(snapshot.membership, server_info_endpoint), pool_state: ClusterPoolStateView::from(snapshot.pool_state), local_storage: ClusterLocalStorageView::from(snapshot.local_storage), peer_health: ClusterPeerHealthView::from(snapshot.peer_health), @@ -324,8 +329,25 @@ pub(crate) struct ClusterMembershipView { impl From for ClusterMembershipView { fn from(snapshot: ClusterMembershipSnapshot) -> Self { + Self::from_snapshot(snapshot, None) + } +} + +impl ClusterMembershipView { + fn from_snapshot(snapshot: ClusterMembershipSnapshot, server_info_endpoint: Option) -> Self { Self { - nodes: snapshot.nodes.into_iter().map(ClusterNodeMembershipView::from).collect(), + nodes: snapshot + .nodes + .into_iter() + .map(|node| { + let endpoint = if node.is_local && node.node_id == "local" { + server_info_endpoint.clone() + } else { + None + }; + ClusterNodeMembershipView::from_node(node, endpoint) + }) + .collect(), drives: snapshot.drives.into_iter().map(ClusterDriveMembershipView::from).collect(), } } @@ -335,15 +357,18 @@ impl From for ClusterMembershipView { pub(crate) struct ClusterNodeMembershipView { pub node_id: String, pub grid_host: String, + #[serde(skip_serializing_if = "Option::is_none")] + pub server_info_endpoint: Option, pub is_local: bool, pub pools: Vec, } -impl From for ClusterNodeMembershipView { - fn from(node: ClusterNodeMembership) -> Self { +impl ClusterNodeMembershipView { + fn from_node(node: ClusterNodeMembership, server_info_endpoint: Option) -> Self { Self { node_id: node.node_id, grid_host: node.grid_host, + server_info_endpoint, is_local: node.is_local, pools: node.pools, } @@ -891,7 +916,7 @@ fn summarize_named_capability_statuses( #[cfg(test)] mod tests { - use super::{ClusterSnapshotResponse, ClusterSnapshotSummary, ClusterSnapshotView}; + use super::{ClusterMembershipView, ClusterSnapshotResponse, ClusterSnapshotSummary, ClusterSnapshotView}; use crate::admin::storage_api::cluster::CapabilityState; use crate::admin::storage_api::cluster::{CapabilityStatus, ObservabilitySnapshot, TopologySnapshot}; use crate::admin::storage_api::cluster::{ @@ -917,6 +942,11 @@ mod tests { handler_block.contains("authorize_cluster_snapshot_request(&req).await?;"), "cluster snapshot handler should require admin authorization" ); + assert!( + handler_block.contains("current_local_node_name().await") + && handler_block.contains("ClusterSnapshotView::from_snapshot"), + "cluster snapshot handler should attach the v3 server-info identity" + ); assert!( auth_block.contains("AdminAction::ServerInfoAdminAction"), "cluster snapshot should require server info admin permission" @@ -949,8 +979,8 @@ mod tests { topology: TopologySnapshot::default(), membership: ClusterMembershipSnapshot { nodes: vec![ClusterNodeMembership { - node_id: "node-a".to_string(), - grid_host: "node-a:9000".to_string(), + node_id: "local".to_string(), + grid_host: String::new(), is_local: true, pools: vec![0], }], @@ -1020,7 +1050,8 @@ mod tests { }, }; - let value = serde_json::to_value(ClusterSnapshotView::from(snapshot)).expect("serialize view"); + let value = serde_json::to_value(ClusterSnapshotView::from_snapshot(snapshot, Some(":::9000".to_string()))) + .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"); @@ -1031,6 +1062,7 @@ mod tests { 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"]["nodes"][0]["server_info_endpoint"], ":::9000"); 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"); @@ -1042,6 +1074,21 @@ mod tests { assert_eq!(value["summary"]["rpc_boundary"]["state"], "supported"); assert_eq!(value["runtime_status"]["degraded_reasons"][0], "storage_and_lock_unavailable"); assert_eq!(value["actionable_pressure"], true); + + let remote_membership = ClusterMembershipView::from_snapshot( + ClusterMembershipSnapshot { + nodes: vec![ClusterNodeMembership { + node_id: "node-b:9000".to_string(), + grid_host: "http://node-b:9000".to_string(), + is_local: false, + pools: vec![0], + }], + drives: Vec::new(), + }, + Some(":::9000".to_string()), + ); + let remote_value = serde_json::to_value(remote_membership).expect("serialize remote membership"); + assert!(remote_value["nodes"][0].get("server_info_endpoint").is_none()); } #[test] diff --git a/rustfs/src/workload_admission.rs b/rustfs/src/workload_admission.rs index 32e13d10e..e7626eefa 100644 --- a/rustfs/src/workload_admission.rs +++ b/rustfs/src/workload_admission.rs @@ -28,7 +28,7 @@ const REPLICATION_RUNTIME_NOT_INITIALIZED: &str = "replication runtime not initi const REPLICATION_QUEUE_BACKLOG_PRESENT: &str = "replication queue has pending work"; const REPLICATION_QUEUE_STATS_UNAVAILABLE: &str = "replication queue stats unavailable"; const SCANNER_ADMISSION_SATURATED: &str = "scanner active work reached configured set-scan limit"; -const SCANNER_ACTIVITY_IDLE_OR_NOT_INITIALIZED: &str = "scanner activity idle or not initialized"; +const SCANNER_RUNTIME_NOT_INITIALIZED: &str = "scanner runtime not initialized"; const STORAGE_CONCURRENCY_PROVIDER_MISSING_FOREGROUND_READ: &str = "storage concurrency provider did not expose foreground read admission"; const STORAGE_CONCURRENCY_PROVIDER_MISSING_FOREGROUND_WRITE: &str = @@ -107,19 +107,24 @@ fn metadata_workload_admission_snapshot_from_initialized(runtime_initialized: bo pub fn scanner_workload_admission_snapshot() -> WorkloadAdmissionSnapshot { let runtime_config = rustfs_scanner::scanner_runtime_config_status(); scanner_workload_admission_snapshot_from_activity( + rustfs_scanner::scanner_runtime_initialized(), rustfs_scanner::current_scanner_activity(), runtime_config.max_concurrent_set_scans.value, ) } -fn scanner_workload_admission_snapshot_from_activity(active: u64, limit: usize) -> WorkloadAdmissionSnapshot { +fn scanner_workload_admission_snapshot_from_activity( + runtime_initialized: bool, + active: u64, + limit: usize, +) -> WorkloadAdmissionSnapshot { let effective_limit = if limit == 0 { None } else { Some(limit) }; - let state = if effective_limit.is_some_and(|limit| usize::try_from(active).ok().is_some_and(|active| active >= limit)) { - AdmissionState::Saturated - } else if active > 0 { - AdmissionState::Open - } else { + let state = if !runtime_initialized { AdmissionState::Unknown + } else if effective_limit.is_some_and(|limit| usize::try_from(active).ok().is_some_and(|active| active >= limit)) { + AdmissionState::Saturated + } else { + AdmissionState::Open }; let snapshot = WorkloadAdmissionSnapshot::new(WorkloadClass::Scanner, state).with_counts( @@ -130,7 +135,7 @@ fn scanner_workload_admission_snapshot_from_activity(active: u64, limit: usize) match state { AdmissionState::Saturated => snapshot.with_reason(SCANNER_ADMISSION_SATURATED), - AdmissionState::Unknown => snapshot.with_reason(SCANNER_ACTIVITY_IDLE_OR_NOT_INITIALIZED), + AdmissionState::Unknown => snapshot.with_reason(SCANNER_RUNTIME_NOT_INITIALIZED), _ => snapshot, } } @@ -277,7 +282,7 @@ mod tests { #[test] fn scanner_snapshot_reports_active_work_units() { - let snapshot = scanner_workload_admission_snapshot_from_activity(5, 8); + let snapshot = scanner_workload_admission_snapshot_from_activity(true, 5, 8); assert_eq!(snapshot.class, WorkloadClass::Scanner); assert_eq!(snapshot.state, AdmissionState::Open); @@ -288,29 +293,40 @@ mod tests { } #[test] - fn scanner_snapshot_is_unknown_when_idle_or_uninitialized() { - let snapshot = scanner_workload_admission_snapshot_from_activity(0, 8); + fn scanner_snapshot_is_open_when_initialized_and_idle() { + let snapshot = scanner_workload_admission_snapshot_from_activity(true, 0, 8); + + assert_eq!(snapshot.class, WorkloadClass::Scanner); + assert_eq!(snapshot.state, AdmissionState::Open); + assert_eq!(snapshot.active, Some(0)); + assert_eq!(snapshot.limit, Some(8)); + assert_eq!(snapshot.reason, None); + } + + #[test] + fn scanner_snapshot_is_unknown_before_runtime_initialization() { + let snapshot = scanner_workload_admission_snapshot_from_activity(false, 0, 8); assert_eq!(snapshot.class, WorkloadClass::Scanner); assert_eq!(snapshot.state, AdmissionState::Unknown); assert_eq!(snapshot.active, Some(0)); assert_eq!(snapshot.limit, Some(8)); - assert_eq!(snapshot.reason.as_deref(), Some(SCANNER_ACTIVITY_IDLE_OR_NOT_INITIALIZED)); + assert_eq!(snapshot.reason.as_deref(), Some(SCANNER_RUNTIME_NOT_INITIALIZED)); } #[test] fn scanner_snapshot_treats_zero_set_scan_limit_as_topology_derived() { - let snapshot = scanner_workload_admission_snapshot_from_activity(0, 0); + let snapshot = scanner_workload_admission_snapshot_from_activity(true, 0, 0); assert_eq!(snapshot.class, WorkloadClass::Scanner); - assert_eq!(snapshot.state, AdmissionState::Unknown); + assert_eq!(snapshot.state, AdmissionState::Open); assert_eq!(snapshot.limit, None); - assert_eq!(snapshot.reason.as_deref(), Some(SCANNER_ACTIVITY_IDLE_OR_NOT_INITIALIZED)); + assert_eq!(snapshot.reason, None); } #[test] fn scanner_snapshot_with_topology_derived_limit_reports_active_work_open() { - let snapshot = scanner_workload_admission_snapshot_from_activity(4, 0); + let snapshot = scanner_workload_admission_snapshot_from_activity(true, 4, 0); assert_eq!(snapshot.class, WorkloadClass::Scanner); assert_eq!(snapshot.state, AdmissionState::Open); @@ -321,7 +337,7 @@ mod tests { #[test] fn scanner_snapshot_reports_saturation_when_active_work_reaches_limit() { - let snapshot = scanner_workload_admission_snapshot_from_activity(4, 4); + let snapshot = scanner_workload_admission_snapshot_from_activity(true, 4, 4); assert_eq!(snapshot.class, WorkloadClass::Scanner); assert_eq!(snapshot.state, AdmissionState::Saturated);