gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-11 15:46:53 +00:00
Code Issues Packages Projects Releases Wiki Activity

fix: correct SNSD cluster diagnostics (#5930)

Browse Source
This commit is contained in:
cxymds
2026-08-10 20:32:01 +08:00
committed by GitHub
parent 95627cb601
commit 7ca69eb39c
4 changed files with 134 additions and 30 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/scanner/src/lib.rs
+39
View File
@@ -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());
}
}
crates/scanner/src/scanner.rs
+3 -1
View File
@@ -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<ECStore>) {
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()),
rustfs/src/admin/handlers/cluster_snapshot.rs
+59 -12
View File
@@ -105,10 +105,9 @@ pub struct GetClusterSnapshotHandler {}
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 = 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<ClusterReadOnlySnapshot> for ClusterSnapshotView {
fn from(snapshot: ClusterReadOnlySnapshot) -> Self {
Self::from_snapshot(snapshot, None)
}
}
impl ClusterSnapshotView {
fn from_snapshot(snapshot: ClusterReadOnlySnapshot, server_info_endpoint: Option<String>) -> 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<ClusterReadOnlySnapshot> 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<ClusterMembershipSnapshot> for ClusterMembershipView {
fn from(snapshot: ClusterMembershipSnapshot) -> Self {
Self::from_snapshot(snapshot, None)
}
}
impl ClusterMembershipView {
fn from_snapshot(snapshot: ClusterMembershipSnapshot, server_info_endpoint: Option<String>) -> 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<ClusterMembershipSnapshot> 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<String>,
pub is_local: bool,
pub pools: Vec<usize>,
}
impl From<ClusterNodeMembership> for ClusterNodeMembershipView {
fn from(node: ClusterNodeMembership) -> Self {
impl ClusterNodeMembershipView {
fn from_node(node: ClusterNodeMembership, server_info_endpoint: Option<String>) -> 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<const N: usize>(
#[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]
rustfs/src/workload_admission.rs
+33 -17
View File
@@ -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);