diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index 14a054990..c041e5b41 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -65,6 +65,7 @@ use crate::storage_api_contracts::{ namespace::NamespaceLocking as _, object::{EcstoreObjectIO, HTTPPreconditions, ObjectIO as _, ObjectOperations as _}, }; +use crate::store::PoolMetaWriteGateStatus; use crate::{core::sets::Sets, store::ECStore}; use byteorder::{ByteOrder, LittleEndian, WriteBytesExt}; use futures::{ @@ -10675,6 +10676,27 @@ impl ECStore { self.pool_meta_write_status().await.is_ok() } + /// A health probe must not wait behind a metadata transaction's disk I/O. + pub async fn pool_meta_write_gate_status(&self) -> PoolMetaWriteGateStatus { + let Ok(write_state) = tokio::time::timeout(std::time::Duration::from_millis(100), self.pool_meta_save_gate.lock()).await + else { + return PoolMetaWriteGateStatus { + writes_ready: false, + ..PoolMetaWriteGateStatus::default() + }; + }; + let transaction_aborted = write_state.aborted_transaction.load(Ordering::SeqCst); + PoolMetaWriteGateStatus { + writes_ready: !write_state.write_blocked && !transaction_aborted, + write_blocked: write_state.write_blocked, + transaction_aborted, + pool_meta_absent: write_state.pool_meta_absent, + identity_initialized: write_state.identity_initialized, + identity_needs_repair: write_state.identity_needs_repair, + cluster_epoch: write_state.cluster_epoch, + } + } + /// A health probe must not wait behind a metadata transaction's disk I/O. pub async fn pool_meta_write_status(&self) -> Result<()> { let write_state = tokio::time::timeout(std::time::Duration::from_millis(100), self.pool_meta_save_gate.lock()) @@ -17701,6 +17723,10 @@ mod tests { .expect("create single-pool bucket before blocking pool metadata writes"); let incarnation = store.bucket_incarnation_id(&bucket).await.expect("load bucket incarnation"); store.pool_meta_save_gate.lock().await.block_writes_after_fence_loss(); + let write_gate = store.pool_meta_write_gate_status().await; + assert!(!write_gate.writes_ready); + assert!(write_gate.write_blocked); + assert!(!write_gate.transaction_aborted); let object = "ordinary-put.bin"; let mut put_data = crate::object_api::PutObjReader::from_vec(b"ordinary single-pool body".to_vec()); diff --git a/crates/ecstore/src/lib.rs b/crates/ecstore/src/lib.rs index bd2f1fa58..c2076e3fd 100644 --- a/crates/ecstore/src/lib.rs +++ b/crates/ecstore/src/lib.rs @@ -55,6 +55,8 @@ mod set_disk; mod storage_api_contracts; mod store; +pub use store::PoolMetaWriteGateStatus; + // pub mod checksum; mod event; diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index b54a69438..78b7508c5 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/crates/ecstore/src/store/mod.rs @@ -527,6 +527,31 @@ impl Default for ScannerDataMovementPauseStatus { } } +#[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)] +pub struct PoolMetaWriteGateStatus { + pub writes_ready: bool, + pub write_blocked: bool, + pub transaction_aborted: bool, + pub pool_meta_absent: bool, + pub identity_initialized: Option, + pub identity_needs_repair: bool, + pub cluster_epoch: Option, +} + +impl Default for PoolMetaWriteGateStatus { + fn default() -> Self { + Self { + writes_ready: true, + write_blocked: false, + transaction_aborted: false, + pool_meta_absent: false, + identity_initialized: None, + identity_needs_repair: false, + cluster_epoch: None, + } + } +} + fn offset_unix_seconds(value: OffsetDateTime) -> u64 { u64::try_from(value.unix_timestamp()).unwrap_or(0) } diff --git a/rustfs/src/admin/handlers/cluster_snapshot.rs b/rustfs/src/admin/handlers/cluster_snapshot.rs index d5cf28a40..18cd26168 100644 --- a/rustfs/src/admin/handlers/cluster_snapshot.rs +++ b/rustfs/src/admin/handlers/cluster_snapshot.rs @@ -25,7 +25,8 @@ use crate::admin::{ system, }; use crate::cluster_snapshot::{ - ClusterReadOnlySnapshot, ClusterRuntimeReadinessState, ClusterRuntimeStatusSnapshot, cluster_has_actionable_pressure, + ClusterPoolMetaWriteGateSnapshot, ClusterReadOnlySnapshot, ClusterRuntimeReadinessState, ClusterRuntimeStatusSnapshot, + cluster_has_actionable_pressure, }; use crate::server::{ADMIN_PREFIX, ReadinessDegradedReason}; use http::{HeaderMap, HeaderValue, StatusCode}; @@ -154,6 +155,7 @@ pub(crate) struct ClusterSnapshotView { pub observability: ObservabilitySnapshot, pub workload_admission: Vec, pub runtime_status: ClusterRuntimeStatusView, + pub pool_meta_write_gate: ClusterPoolMetaWriteGateView, pub actionable_pressure: bool, } @@ -182,11 +184,38 @@ impl ClusterSnapshotView { observability: snapshot.observability, workload_admission: workload_admission_views(snapshot.workload_admission), runtime_status: ClusterRuntimeStatusView::from(snapshot.runtime_status), + pool_meta_write_gate: ClusterPoolMetaWriteGateView::from(snapshot.pool_meta_write_gate), actionable_pressure, } } } +#[derive(Debug, Clone, PartialEq, Eq, Serialize)] +#[serde(rename_all = "camelCase")] +pub(crate) struct ClusterPoolMetaWriteGateView { + pub writes_ready: bool, + pub write_blocked: bool, + pub transaction_aborted: bool, + pub pool_meta_absent: bool, + pub identity_initialized: Option, + pub identity_needs_repair: bool, + pub cluster_epoch: Option, +} + +impl From for ClusterPoolMetaWriteGateView { + fn from(snapshot: ClusterPoolMetaWriteGateSnapshot) -> Self { + Self { + writes_ready: snapshot.writes_ready, + write_blocked: snapshot.write_blocked, + transaction_aborted: snapshot.transaction_aborted, + pool_meta_absent: snapshot.pool_meta_absent, + identity_initialized: snapshot.identity_initialized, + identity_needs_repair: snapshot.identity_needs_repair, + cluster_epoch: snapshot.cluster_epoch, + } + } +} + #[derive(Debug, Clone, PartialEq, Eq, Serialize)] pub(crate) struct ClusterComponentStatusView { pub storage: ClusterComponentStatus, @@ -977,8 +1006,8 @@ mod tests { ClusterPoolStateSnapshot, ClusterRpcBoundarySnapshot, ClusterRpcChannelSnapshot, ClusterRpcPlane, ClusterRpcTransport, }; use crate::cluster_snapshot::{ - ClusterListingDiagnosticsSnapshot, ClusterReadOnlySnapshot, ClusterRuntimeReadinessState, ClusterRuntimeStatusSnapshot, - ClusterUsageFreshnessSnapshot, + ClusterListingDiagnosticsSnapshot, ClusterPoolMetaWriteGateSnapshot, ClusterReadOnlySnapshot, + ClusterRuntimeReadinessState, ClusterRuntimeStatusSnapshot, ClusterUsageFreshnessSnapshot, }; use crate::shared_types::{DependencyReadiness, ReadinessDegradedReason}; use rustfs_concurrency::{AdmissionState, WorkloadAdmissionRegistrySnapshot, WorkloadAdmissionSnapshot, WorkloadClass}; @@ -1124,6 +1153,15 @@ mod tests { listing_diagnostics: ClusterListingDiagnosticsSnapshot { internode_stall_timeouts_total: 2, }, + pool_meta_write_gate: ClusterPoolMetaWriteGateSnapshot { + writes_ready: false, + write_blocked: true, + transaction_aborted: true, + identity_initialized: Some(true), + identity_needs_repair: true, + cluster_epoch: Some(7), + ..Default::default() + }, }; let value = serde_json::to_value(ClusterSnapshotView::from_snapshot(snapshot, Some(":::9000".to_string()))) @@ -1136,6 +1174,13 @@ mod tests { 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["pool_meta_write_gate"]["writesReady"], false); + assert_eq!(value["pool_meta_write_gate"]["writeBlocked"], true); + assert_eq!(value["pool_meta_write_gate"]["transactionAborted"], true); + assert_eq!(value["pool_meta_write_gate"]["identityInitialized"], true); + assert_eq!(value["pool_meta_write_gate"]["identityNeedsRepair"], true); + assert_eq!(value["pool_meta_write_gate"]["clusterEpoch"], 7); + assert_eq!(value["actionable_pressure"], true); 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"); @@ -1193,6 +1238,7 @@ mod tests { }, usage_freshness: ClusterUsageFreshnessSnapshot::default(), listing_diagnostics: ClusterListingDiagnosticsSnapshot::default(), + pool_meta_write_gate: ClusterPoolMetaWriteGateSnapshot::default(), }; let summary = ClusterSnapshotSummary::from(&snapshot); @@ -1259,6 +1305,7 @@ mod tests { listing_diagnostics: ClusterListingDiagnosticsSnapshot { internode_stall_timeouts_total: 0, }, + pool_meta_write_gate: ClusterPoolMetaWriteGateSnapshot::default(), }; let view = ClusterSnapshotView::from(snapshot); @@ -1306,6 +1353,11 @@ mod tests { }, usage_freshness: ClusterUsageFreshnessSnapshot::default(), listing_diagnostics: ClusterListingDiagnosticsSnapshot::default(), + pool_meta_write_gate: ClusterPoolMetaWriteGateSnapshot { + writes_ready: false, + write_blocked: true, + ..Default::default() + }, }; let view = ClusterSnapshotView::from(snapshot); @@ -1341,6 +1393,7 @@ mod tests { listing_diagnostics: ClusterListingDiagnosticsSnapshot { internode_stall_timeouts_total: 2, }, + pool_meta_write_gate: ClusterPoolMetaWriteGateSnapshot::default(), }; let component = super::summarize_listing_metacache(&snapshot); @@ -1384,6 +1437,7 @@ mod tests { ..Default::default() }, listing_diagnostics: ClusterListingDiagnosticsSnapshot::default(), + pool_meta_write_gate: ClusterPoolMetaWriteGateSnapshot::default(), }; let component = super::summarize_usage_freshness(&snapshot); @@ -1421,6 +1475,7 @@ mod tests { ..Default::default() }, listing_diagnostics: ClusterListingDiagnosticsSnapshot::default(), + pool_meta_write_gate: ClusterPoolMetaWriteGateSnapshot::default(), }; let component = super::summarize_usage_freshness(&snapshot); @@ -1470,6 +1525,7 @@ mod tests { }, usage_freshness: ClusterUsageFreshnessSnapshot::default(), listing_diagnostics: ClusterListingDiagnosticsSnapshot::default(), + pool_meta_write_gate: ClusterPoolMetaWriteGateSnapshot::default(), }; let summary = ClusterSnapshotSummary::from(&snapshot); diff --git a/rustfs/src/cluster_snapshot.rs b/rustfs/src/cluster_snapshot.rs index 33abbcf54..a1433fdec 100644 --- a/rustfs/src/cluster_snapshot.rs +++ b/rustfs/src/cluster_snapshot.rs @@ -40,6 +40,7 @@ pub struct ClusterReadOnlySnapshot { pub runtime_status: ClusterRuntimeStatusSnapshot, pub usage_freshness: ClusterUsageFreshnessSnapshot, pub listing_diagnostics: ClusterListingDiagnosticsSnapshot, + pub pool_meta_write_gate: ClusterPoolMetaWriteGateSnapshot, } #[derive(Debug, Clone, PartialEq, Eq)] @@ -104,6 +105,31 @@ pub struct ClusterListingDiagnosticsSnapshot { pub internode_stall_timeouts_total: u64, } +#[derive(Debug, Clone, PartialEq, Eq)] +pub struct ClusterPoolMetaWriteGateSnapshot { + pub writes_ready: bool, + pub write_blocked: bool, + pub transaction_aborted: bool, + pub pool_meta_absent: bool, + pub identity_initialized: Option, + pub identity_needs_repair: bool, + pub cluster_epoch: Option, +} + +impl Default for ClusterPoolMetaWriteGateSnapshot { + fn default() -> Self { + Self { + writes_ready: true, + write_blocked: false, + transaction_aborted: false, + pool_meta_absent: false, + identity_initialized: None, + identity_needs_repair: false, + cluster_epoch: None, + } + } +} + impl From for ClusterListingDiagnosticsSnapshot { fn from(snapshot: InternodeMetricsSnapshot) -> Self { Self { @@ -164,6 +190,7 @@ pub fn cluster_read_only_snapshot_from_control_plane( runtime_status, usage_freshness: ClusterUsageFreshnessSnapshot::default(), listing_diagnostics: ClusterListingDiagnosticsSnapshot::default(), + pool_meta_write_gate: ClusterPoolMetaWriteGateSnapshot::default(), } } @@ -172,6 +199,7 @@ pub async fn collect_cluster_read_only_snapshot(endpoint_pools: &EndpointServerP 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(); + snapshot.pool_meta_write_gate = current_pool_meta_write_gate_snapshot().await; Some(snapshot) } @@ -188,8 +216,27 @@ fn current_listing_diagnostics_snapshot() -> ClusterListingDiagnosticsSnapshot { ClusterListingDiagnosticsSnapshot::from(metrics.snapshot()) } +async fn current_pool_meta_write_gate_snapshot() -> ClusterPoolMetaWriteGateSnapshot { + match crate::runtime_sources::current_object_store_handle() { + Some(store) => { + let status = store.pool_meta_write_gate_status().await; + ClusterPoolMetaWriteGateSnapshot { + writes_ready: status.writes_ready, + write_blocked: status.write_blocked, + transaction_aborted: status.transaction_aborted, + pool_meta_absent: status.pool_meta_absent, + identity_initialized: status.identity_initialized, + identity_needs_repair: status.identity_needs_repair, + cluster_epoch: status.cluster_epoch, + } + } + None => ClusterPoolMetaWriteGateSnapshot::default(), + } +} + pub fn cluster_has_actionable_pressure(snapshot: &ClusterReadOnlySnapshot) -> bool { snapshot.runtime_status.state == ClusterRuntimeReadinessState::Degraded + || !snapshot.pool_meta_write_gate.writes_ready || snapshot .workload_admission .entries() @@ -312,6 +359,7 @@ mod tests { }, usage_freshness: ClusterUsageFreshnessSnapshot::default(), listing_diagnostics: ClusterListingDiagnosticsSnapshot::default(), + pool_meta_write_gate: ClusterPoolMetaWriteGateSnapshot::default(), }; assert!(!cluster_has_actionable_pressure(&no_pressure)); @@ -335,9 +383,19 @@ mod tests { WorkloadClass::Repair, AdmissionState::Unknown, )]), - ..no_pressure + ..no_pressure.clone() }; assert!(cluster_has_actionable_pressure(&admission_pressure)); + + let pool_meta_pressure = ClusterReadOnlySnapshot { + pool_meta_write_gate: ClusterPoolMetaWriteGateSnapshot { + writes_ready: false, + write_blocked: true, + ..Default::default() + }, + ..no_pressure + }; + assert!(cluster_has_actionable_pressure(&pool_meta_pressure)); } #[test] diff --git a/scripts/error-other-format-baseline.txt b/scripts/error-other-format-baseline.txt index ddd906666..e1ab9154b 100644 --- a/scripts/error-other-format-baseline.txt +++ b/scripts/error-other-format-baseline.txt @@ -25,7 +25,7 @@ 10|crates/ecstore/src/cluster/rpc/remote_disk.rs 6|crates/ecstore/src/config/com.rs 14|crates/ecstore/src/config/storageclass.rs -181|crates/ecstore/src/core/pools.rs +180|crates/ecstore/src/core/pools.rs 7|crates/ecstore/src/data_movement/mod.rs 2|crates/ecstore/src/data_usage/local_snapshot.rs 12|crates/ecstore/src/data_usage/mod.rs