From 6b05fb6b427810d1b0f95aa90b907136e7eb60d7 Mon Sep 17 00:00:00 2001 From: cxymds Date: Tue, 8 Sep 2026 07:21:19 +0800 Subject: [PATCH] feat(observability): expose pool write-block diagnostics (#7417) * feat(observability): expose pool write-block diagnostics * fix(error): merge equivalent api message branches Combine the MaxVersionsExceeded and internal IO message branches so Clippy no longer flags identical if blocks while preserving the existing response messages. Co-Authored-By: heihutu Co-Authored-By: zhi22915 * fix(heal): cleanup consumed MRF replay journals Do not retain Accepted or Merged replay intents as startup anchors after they have been handed to the heal manager. Only refused or still-pending replay records keep the journal on disk until a successor snapshot can persist them. This keeps successor snapshots limited to the pending queue, which lets successful replay remove both authoritative and legacy journal paths and restores the crash-boundary tests around successor flush. Co-Authored-By: heihutu Co-Authored-By: zhi22915 (cherry picked from commit d5b8f49c9d63afafa8e86a1ac085f9af96ec6406) --------- Signed-off-by: houseme Co-authored-by: houseme Co-authored-by: zhi22915 Co-authored-by: Zhengchao An --- .../src/admin_timeout_regression_test.rs | 12 +- crates/e2e_test/src/common.rs | 9 + .../src/distributed/observability_test.rs | 189 +++++++++++++++++- crates/ecstore/src/core/pools.rs | 49 ++++- crates/ecstore/src/store/mod.rs | 8 + docs/architecture/readiness-matrix.md | 6 +- docs/operations/pool-metadata-recovery.md | 2 +- .../s3-write-failure-diagnostics.md | 15 +- docs/testing/distributed-e2e.md | 2 +- rustfs/src/admin/handlers/cluster_snapshot.rs | 83 +++++++- rustfs/src/app/object/put.rs | 3 + rustfs/src/cluster_snapshot.rs | 17 +- rustfs/src/error.rs | 43 +++- rustfs/src/server/readiness.rs | 97 +++++++-- rustfs/src/shared_types.rs | 2 + rustfs/src/storage/storage_api.rs | 9 +- rustfs/src/storage_api.rs | 3 +- 17 files changed, 500 insertions(+), 49 deletions(-) diff --git a/crates/e2e_test/src/admin_timeout_regression_test.rs b/crates/e2e_test/src/admin_timeout_regression_test.rs index 94a91f03e..8337504c2 100644 --- a/crates/e2e_test/src/admin_timeout_regression_test.rs +++ b/crates/e2e_test/src/admin_timeout_regression_test.rs @@ -12,7 +12,7 @@ // See the License for the specific language governing permissions and // limitations under the License. -use crate::common::{RustFSTestClusterEnvironment, init_logging, local_http_client}; +use crate::common::{RustFSTestClusterEnvironment, init_logging, local_http_client, signal_process}; use aws_sdk_s3::primitives::ByteStream; use http::header::HOST; use reqwest::StatusCode; @@ -22,7 +22,6 @@ use rustfs_signer::sign_v4; use s3s::Body; use serde::Deserialize; use std::error::Error; -use std::process::Command; use tokio::time::{Duration, sleep, timeout}; use uuid::Uuid; @@ -82,15 +81,6 @@ async fn parse_json_response( Ok(serde_json::from_slice(&body)?) } -fn signal_process(pid: u32, signal: &str) -> Result<(), Box> { - let output = Command::new("kill").arg(format!("-{signal}")).arg(pid.to_string()).output()?; - if output.status.success() { - return Ok(()); - } - - Err(format!("kill -{signal} {pid} failed: {}", String::from_utf8_lossy(&output.stderr)).into()) -} - fn offline_server_count(info: &InfoMessage) -> usize { info.servers .as_ref() diff --git a/crates/e2e_test/src/common.rs b/crates/e2e_test/src/common.rs index 554f55239..ed4ba96b6 100644 --- a/crates/e2e_test/src/common.rs +++ b/crates/e2e_test/src/common.rs @@ -264,6 +264,15 @@ pub fn local_http_client() -> HttpClient { .expect("failed to build local reqwest client") } +pub(crate) fn signal_process(pid: u32, signal: &str) -> Result<(), Box> { + let output = Command::new("kill").arg(format!("-{signal}")).arg(pid.to_string()).output()?; + if output.status.success() { + return Ok(()); + } + + Err(format!("kill -{signal} {pid} failed: {}", String::from_utf8_lossy(&output.stderr)).into()) +} + pub(crate) async fn signed_s3_request( method: http::Method, url: &str, diff --git a/crates/e2e_test/src/distributed/observability_test.rs b/crates/e2e_test/src/distributed/observability_test.rs index bb722b92e..81811a2c5 100644 --- a/crates/e2e_test/src/distributed/observability_test.rs +++ b/crates/e2e_test/src/distributed/observability_test.rs @@ -12,8 +12,10 @@ // See the License for the specific language governing permissions and // limitations under the License. -use super::harness::{DistCluster, DistLayout, TestResult, cluster_admin_ok, unique_bucket, wait_for_ready}; -use crate::common::{admin_request, init_logging, local_http_client}; +use super::harness::{ + DistCluster, DistLayout, TestResult, assert_object_bytes, cluster_admin_ok, unique_bucket, wait_for_ready, wait_until, +}; +use crate::common::{admin_request, init_logging, local_http_client, signal_process, signed_request}; use aws_sdk_s3::operation::RequestId; use aws_sdk_s3::primitives::ByteStream; use bytes::Bytes; @@ -24,7 +26,7 @@ use hyper::service::service_fn; use hyper::{Request, Response}; use hyper_util::rt::TokioIo; use local_ip_address::local_ip; -use rustfs_madmin::metrics::RealtimeMetrics; +use rustfs_madmin::metrics::{HttpMetrics, RealtimeMetrics}; use rustfs_utils::egress::ENV_OUTBOUND_ALLOW_ORIGINS; use serde_json::Value; use std::convert::Infallible; @@ -126,6 +128,7 @@ async fn four_node_health_inventory_metrics_and_audit_delivery_are_consistent() let (audit_endpoint, mut audit_entries, collector) = spawn_audit_collector().await?; let audit_origin = reqwest::Url::parse(&audit_endpoint)?.origin().ascii_serialization(); let audit_env = [ + ("RUST_LOG", "warn"), ("RUSTFS_AUDIT_ENABLE", "true"), ("RUSTFS_AUDIT_WEBHOOK_ENABLE_DISTRIBUTED", "on"), ("RUSTFS_AUDIT_WEBHOOK_ENDPOINT_DISTRIBUTED", audit_endpoint.as_str()), @@ -231,6 +234,186 @@ async fn four_node_health_inventory_metrics_and_audit_delivery_are_consistent() "audit entry leaked the root secret key" ); + let result = verify_write_observations_during_peer_failure(&dist, &bucket).await; collector.abort(); + result +} + +async fn node_admin_body(dist: &DistCluster, node: usize, path: &str) -> TestResult { + let (status, body) = timeout( + Duration::from_secs(30), + admin_request( + &dist.cluster.nodes[node].url, + Method::GET, + path, + None, + &dist.cluster.access_key, + &dist.cluster.secret_key, + ), + ) + .await??; + assert!(status.is_success(), "node {node} admin request {path}: {status} {body}"); + Ok(body) +} + +async fn http_put_counts(dist: &DistCluster, node: usize) -> TestResult<[u64; 2]> { + let body = node_admin_body(dist, node, "/rustfs/admin/v3/metrics?types=512&by-host=true&n=1").await?; + let sample: RealtimeMetrics = serde_json::from_str(body.lines().next().ok_or("empty HTTP metrics stream")?)?; + assert!(sample.errors.is_empty(), "HTTP metrics returned errors: {:?}", sample.errors); + let http = sample.aggregated.http.ok_or("HTTP metrics missing at WARN log level")?; + let count = |http: &HttpMetrics, outcome: &str| { + http.requests + .iter() + .filter(|row| row.method == "PUT" && row.outcome == outcome) + .map(|row| row.total) + .sum::() + }; + assert_eq!(sample.by_host.len(), 1, "HTTP admin metrics must remain node-local"); + let host = sample.by_host.values().next().expect("one reporting host"); + let host = host.http.as_ref().ok_or("by-host HTTP metrics missing")?; + let totals = [count(&http, "2xx"), count(&http, "5xx")]; + assert_eq!(totals, [count(host, "2xx"), count(host, "5xx")]); + Ok(totals) +} + +async fn observed_put(dist: &DistCluster, node: usize, bucket: &str, key: &str) -> TestResult { + // One signed HTTP attempt: SDK retries must not change the expected denominator. + timeout(Duration::from_secs(90), async { + let response = signed_request( + Method::PUT, + &format!("{}/{bucket}/{key}", dist.cluster.nodes[node].url), + &dist.cluster.access_key, + &dist.cluster.secret_key, + Some(b"write-observation".to_vec()), + Some("application/octet-stream"), + ) + .await?; + assert!(response.headers().contains_key("x-amz-request-id"), "PUT omitted correlation ID"); + let status = response.status(); + let body = response.text().await?; + assert!(!body.contains(&dist.cluster.secret_key), "PUT response leaked credentials"); + Ok::<_, Box>(status) + }) + .await? +} + +struct SuspendedPeer<'a> { + // Borrowing the owned child keeps its PID from being reaped/reused before cleanup. + child: &'a std::process::Child, + suspended: bool, +} + +impl<'a> SuspendedPeer<'a> { + fn suspend(dist: &'a DistCluster, node: usize) -> TestResult { + let child = dist.cluster.nodes[node].process.as_ref().ok_or("peer process missing")?; + signal_process(child.id(), "STOP")?; + Ok(Self { child, suspended: true }) + } + + fn resume(&mut self) -> TestResult { + signal_process(self.child.id(), "CONT")?; + self.suspended = false; + Ok(()) + } +} + +impl Drop for SuspendedPeer<'_> { + fn drop(&mut self) { + if self.suspended { + let _ = signal_process(self.child.id(), "CONT"); + } + } +} + +async fn verify_write_observations_during_peer_failure(dist: &DistCluster, bucket: &str) -> TestResult { + let mut baseline = Vec::new(); + for node in 0..dist.cluster.nodes.len() { + let before = http_put_counts(dist, node).await?; + assert!( + observed_put(dist, node, bucket, &format!("healthy-{node}")) + .await? + .is_success() + ); + let after = http_put_counts(dist, node).await?; + assert_eq!(after, [before[0] + 1, before[1]], "node {node} lost its successful PUT denominator"); + baseline.push(after); + } + + // Refresh provenance immediately before the first failed probe, within the cache age budget. + node_admin_body(dist, 0, "/rustfs/admin/v3/storageinfo").await?; + let mut suspended = [SuspendedPeer::suspend(dist, 2)?, SuspendedPeer::suspend(dist, 3)?]; + let storage: Value = serde_json::from_str(&node_admin_body(dist, 0, "/rustfs/admin/v3/storageinfo").await?)?; + let observations = storage["info"]["observations"] + .as_array() + .ok_or("storageinfo omitted observations")?; + let disks = storage["info"]["disks"] + .as_array() + .ok_or("storageinfo omitted disks during peer failure")?; + assert_eq!(disks.len(), 16, "failed peers must not vanish from inventory"); + for node in [2, 3] { + let endpoint = &dist.cluster.nodes[node].address; + let observation = observations + .iter() + .find(|item| item["endpoint"].as_str().is_some_and(|value| value.contains(endpoint))) + .ok_or_else(|| format!("missing failed peer observation {endpoint}: {storage}"))?; + assert_eq!(observation["status"], "failed", "suspension did not affect peer RPC: {observation}"); + assert_eq!(observation["cached"], true, "first failure must identify the warm cache: {observation}"); + assert!(observation["last_success_unix_millis"].as_u64().is_some()); + assert!(observation["snapshot_age_seconds"].as_u64().is_some_and(|age| age < 60)); + let peer_disks: Vec<_> = disks + .iter() + .filter(|disk| disk["endpoint"].as_str().is_some_and(|value| value.contains(endpoint))) + .collect(); + assert_eq!(peer_disks.len(), 4, "failed peer lost its four drive identities: {storage}"); + for disk in peer_disks { + assert_eq!(disk["state"], "unknown"); + assert_eq!(disk["runtimeState"], "unknown"); + } + } + + let snapshot: Value = serde_json::from_str(&node_admin_body(dist, 0, "/rustfs/admin/v4/cluster/snapshot").await?)?; + let metadata = &snapshot["snapshot"]["pool_meta_write_gate"]; + assert_eq!( + metadata["state"], "writable", + "peer probe failure must not invent a metadata latch: {snapshot}" + ); + assert!(metadata.get("sinceUnixSecs").is_none()); + + for attempt in 0..2 { + let status = observed_put(dist, 0, bucket, &format!("unavailable-{attempt}")).await?; + assert!(status.is_server_error(), "sub-quorum write unexpectedly returned {status}"); + } + assert_eq!(http_put_counts(dist, 0).await?, [baseline[0][0], baseline[0][1] + 2]); + assert_eq!( + http_put_counts(dist, 1).await?, + baseline[1], + "internal RPCs must not count as external PUTs" + ); + + for peer in &mut suspended { + peer.resume()?; + } + wait_until( + Duration::from_secs(90), + || async { + let storage: Value = serde_json::from_str(&node_admin_body(dist, 0, "/rustfs/admin/v3/storageinfo").await?)?; + let observations = storage["info"]["observations"] + .as_array() + .ok_or("recovery omitted observations")?; + Ok(observations.len() == 4 + && observations + .iter() + .all(|item| item["status"] == "succeeded" && item["cached"] == false)) + }, + "peer probes recover to fresh successful observations", + ) + .await?; + wait_for_ready(&dist.cluster).await?; + assert!(observed_put(dist, 0, bucket, "recovered").await?.is_success()); + assert_eq!(http_put_counts(dist, 0).await?, [baseline[0][0] + 1, baseline[0][1] + 2]); + for node in 0..dist.cluster.nodes.len() { + assert_object_bytes(&dist.client(node)?, bucket, "healthy-0", b"write-observation").await?; + assert_object_bytes(&dist.client(node)?, bucket, "recovered", b"write-observation").await?; + } Ok(()) } diff --git a/crates/ecstore/src/core/pools.rs b/crates/ecstore/src/core/pools.rs index c041e5b41..9fc378822 100644 --- a/crates/ecstore/src/core/pools.rs +++ b/crates/ecstore/src/core/pools.rs @@ -10682,18 +10682,30 @@ impl ECStore { else { return PoolMetaWriteGateStatus { writes_ready: false, + check_timed_out: true, ..PoolMetaWriteGateStatus::default() }; }; let transaction_aborted = write_state.aborted_transaction.load(Ordering::SeqCst); + let writes_ready = !write_state.write_blocked && !transaction_aborted; + let failure = if writes_ready { + None + } else { + write_state.ensure_write_safe("pool metadata snapshot").err() + }; + let context = failure.as_ref().and_then(Error::pool_metadata_failure); PoolMetaWriteGateStatus { - writes_ready: !write_state.write_blocked && !transaction_aborted, + writes_ready, + check_timed_out: false, 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, + reason: context.map(|context| context.kind.as_str()), + phase: context.map(|context| context.phase), + since_unix_secs: context.map(|context| context.since.unix_timestamp()), } } @@ -19002,6 +19014,12 @@ mod tests { drop(outstanding); assert!(store.recover_pool_meta_transaction().await.unwrap()); assert!(store.pool_meta_writes_ready().await); + let recovered = store.pool_meta_write_gate_status().await; + assert!(recovered.writes_ready); + assert!(!recovered.check_timed_out); + assert!(!recovered.write_blocked); + assert!(!recovered.transaction_aborted); + assert_eq!((recovered.reason, recovered.phase, recovered.since_unix_secs), (None, None, None)); assert_eq!(store.pool_meta.read().await.pools[0].last_update, requested.pools[0].last_update); assert!(!store.recover_pool_meta_transaction().await.unwrap()); store.pool_meta_save_gate.lock().await.block_writes_after_fence_loss(); @@ -19018,17 +19036,39 @@ mod tests { .await .unwrap(); assert!(matches!(result, Err(Error::Timeout))); + let snapshot = tokio::time::timeout(std::time::Duration::from_secs(1), store.pool_meta_write_gate_status()) + .await + .unwrap(); + assert!(!snapshot.writes_ready); + assert!(snapshot.check_timed_out); + assert!(!snapshot.write_blocked); + assert!(!snapshot.transaction_aborted); + assert_eq!((snapshot.reason, snapshot.phase, snapshot.since_unix_secs), (None, None, None)); } #[tokio::test] #[serial_test::serial] async fn pool_meta_cancelled_recovery_and_new_integrity_block_never_clear_original_gate() { let (_dirs, store, _peer) = crate::services::rebalance::test_two_pool_stores(None).await; + let since = OffsetDateTime::from_unix_timestamp(1_700_000_000).unwrap(); { let state = store.pool_meta_save_gate.lock().await; let mut arm = state.arm_transaction(); arm.phase = Some("publication"); + drop(arm); + // An old fixed timestamp detects polling-time resets without sleeping. + state.transaction_failure.lock().unwrap().as_mut().unwrap().since = since; + *state.block_started_at.lock().unwrap() = Some(since); } + let original = store.pool_meta_write_gate_status().await; + assert!(!original.writes_ready); + assert!(!original.check_timed_out); + assert!(!original.write_blocked); + assert!(original.transaction_aborted); + assert_eq!(original.reason, Some("transaction_unknown")); + assert_eq!(original.phase, Some("publication")); + assert_eq!(original.since_unix_secs, Some(since.unix_timestamp())); + assert_eq!(store.pool_meta_write_gate_status().await, original); let publication_guard = store.pool_meta.write().await; let recovery = tokio::spawn({ let store = store.clone(); @@ -19048,6 +19088,7 @@ mod tests { assert!(recovery.await.unwrap_err().is_cancelled()); drop(publication_guard); assert!(!store.pool_meta_writes_ready().await); + assert_eq!(store.pool_meta_write_gate_status().await, original); let start_guard = store.start_gate.lock().await; let recovery = tokio::spawn({ let store = store.clone(); @@ -19066,6 +19107,12 @@ mod tests { .kind, crate::error::PoolMetadataFailure::FenceLost ); + let fenced = store.pool_meta_write_gate_status().await; + assert!(!fenced.writes_ready); + assert!(fenced.write_blocked); + assert_eq!(fenced.reason, Some("fence_lost")); + assert_eq!(fenced.phase, Some("format_heal")); + assert_eq!(fenced.since_unix_secs, original.since_unix_secs); } #[test] diff --git a/crates/ecstore/src/store/mod.rs b/crates/ecstore/src/store/mod.rs index 78b7508c5..7207e1c8b 100644 --- a/crates/ecstore/src/store/mod.rs +++ b/crates/ecstore/src/store/mod.rs @@ -530,24 +530,32 @@ impl Default for ScannerDataMovementPauseStatus { #[derive(Clone, Debug, PartialEq, Eq, serde::Serialize)] pub struct PoolMetaWriteGateStatus { pub writes_ready: bool, + pub check_timed_out: 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, + pub reason: Option<&'static str>, + pub phase: Option<&'static str>, + pub since_unix_secs: Option, } impl Default for PoolMetaWriteGateStatus { fn default() -> Self { Self { writes_ready: true, + check_timed_out: false, write_blocked: false, transaction_aborted: false, pool_meta_absent: false, identity_initialized: None, identity_needs_repair: false, cluster_epoch: None, + reason: None, + phase: None, + since_unix_secs: None, } } } diff --git a/docs/architecture/readiness-matrix.md b/docs/architecture/readiness-matrix.md index 45e396101..75b180bb6 100644 --- a/docs/architecture/readiness-matrix.md +++ b/docs/architecture/readiness-matrix.md @@ -41,8 +41,12 @@ an unknown or unsupported peer-health snapshot degrades readiness with lock quorum, or peer health. - Node readiness reports local dependency readiness. - A blocked pool metadata writer degrades node and cluster-write readiness with - `pool_metadata_blocked`. Metadata save-gate inspection is bounded to 100 ms; + `pool_meta_write_blocked`. Metadata save-gate inspection is bounded to 100 ms; contention reports `pool_metadata_check_timeout` without installing a block. +- The authenticated cluster snapshot extends its existing node-local metadata + gate inspection with safe reason, failure phase, and original block time. It + distinguishes timeout from a block and changes no admission or recovery + decision. Runtime readiness and gate status are separate bounded observations. - Cluster write readiness requires write quorum and the runtime dependency readiness used by `FullReady`. - Cluster read readiness may use the read-quorum path and cluster-health diff --git a/docs/operations/pool-metadata-recovery.md b/docs/operations/pool-metadata-recovery.md index 0cb2513ff..d4281530a 100644 --- a/docs/operations/pool-metadata-recovery.md +++ b/docs/operations/pool-metadata-recovery.md @@ -49,7 +49,7 @@ An unreadable replica, lost fence, or conditional-write conflict leaves the orig - The first block emits `decommission_state` with `state=pool_metadata_blocked`, `reason`, `phase`, and `blocked_since`. A change in recovery failure classification emits `state=pool_metadata_recovery_pending`; successful recovery emits `state=pool_metadata_recovered` with the original timestamp. - `rustfs_pool_metadata_blocks_total{reason}` and `rustfs_pool_metadata_recoveries_total` count block and recovery transitions. The original cause and phase remain attached to local typed errors; storage/RPC error numbers and on-disk formats are unchanged. -- Node and cluster-write readiness include `pool_metadata_blocked`. Waiting for the metadata save mutex is bounded to 100 ms and reports `pool_metadata_check_timeout`, not a persistent block. Cluster probes retain their existing cache and overall timeout behavior. Liveness and cluster-read quorum checks are unchanged. +- Node and cluster-write readiness include `pool_meta_write_blocked`. Waiting for the metadata save mutex is bounded to 100 ms and reports `pool_metadata_check_timeout`, not a persistent block. Cluster probes retain their existing cache and overall timeout behavior. Liveness and cluster-read quorum checks are unchanged. The authenticated node-local status and safe error fields are described in [S3 write failure diagnostics](s3-write-failure-diagnostics.md). If a block persists, inspect the first block and subsequent recovery phase, restore disk/peer readability, and verify every metadata and identity copy before restarting. Do not delete metadata to make readiness green. diff --git a/docs/operations/s3-write-failure-diagnostics.md b/docs/operations/s3-write-failure-diagnostics.md index d186dac70..16d56e788 100644 --- a/docs/operations/s3-write-failure-diagnostics.md +++ b/docs/operations/s3-write-failure-diagnostics.md @@ -45,10 +45,23 @@ The probe round timeout is configured independently; see [Admin peer probe timeo ## Correlate bounded diagnostics -Normal operation does not require success logs at WARN. Request counters remain available with WARN logging, while existing runtime readiness diagnostics distinguish `pool_meta_write_blocked` from insufficient storage quorum. Do not clear a metadata write fence merely to make readiness green. +Normal operation does not require success logs at WARN. Request counters remain available with WARN logging, while runtime readiness diagnostics distinguish `pool_meta_write_blocked`, `pool_metadata_check_timeout`, and insufficient storage quorum. Do not clear a metadata write fence merely to make readiness green. + +Query the authenticated `/rustfs/admin/v4/cluster/snapshot` endpoint on the affected node. `snapshot.pool_meta_write_gate` describes that node's metadata writer, not a fleet-wide aggregate. Its existing booleans are preserved; `state` and optional block details extend the same bounded, read-only gate inspection. This starts no recovery, disk reads, or additional RPCs. Runtime readiness and the gate are inspected separately, so a whole cluster snapshot is not atomic across sections. + +| `state` | Meaning | +| --- | --- | +| `writable` | No metadata write block was observed. Other dependencies may still make the node unready. | +| `blocked` | A metadata write block was observed; `reason`, `phase`, and `sinceUnixSecs` describe its typed failure context. | +| `check_timeout` | The existing 100 ms inspection budget expired. Readiness remains false, but a write block is not asserted. | +| `unavailable` | The object store or a recognized metadata observation was unavailable; no block details are invented. | + +For `blocked`, `reason` reuses the metadata failure classification, `phase` identifies the operation stage that caused the block (not live recovery-worker progress), and `sinceUnixSecs` is the original block time in Unix seconds. Polling does not reset that time. Recovery replaces the observation with `writable` and removes the previous block details. An unavailable store reports `writesReady=false` with `state="unavailable"`. Older responses may omit `state` and block details; missing fields alone do not prove a healthy writer. Operation text, raw replica errors, disk paths, and credentials are excluded. See [Pool metadata recovery](pool-metadata-recovery.md) for recovery and escalation boundaries. PUT storage failures retain their typed source chain internally and emit bounded S3/storage error codes, I/O kinds, and RPC status codes alongside the existing request ID, bucket, and key. Raw nested error strings and RPC metadata are not logged by this diagnostic. A repeated PUT diagnostic is limited to one event per five seconds; HTTP server-error logs are limited per status code over the same interval for accounted S3 traffic. `suppressed_errors` reports suppressed events at the next emitted event; use the HTTP counter, not log-line counts, to measure failures. HTTP server-error URI diagnostics omit query strings, including presigned credentials. +Typed pool metadata failures additionally carry `pool_metadata_reason`, `pool_metadata_phase`, and `pool_metadata_since_unix_secs` in the PUT diagnostic. These describe the request's failure context: `read_unavailable` before write dispatch is retryable and does not by itself imply a latched write block. Use the current admin snapshot to distinguish that case from a persistent block. Public S3 errors remain sanitized `503 ServiceUnavailable` responses. + Storage inventory emits a WARN event on the first failed probe and an INFO event on recovery, using `event="storage_info_probe"`. A recovery event confirms the RPC succeeded, not that every reported disk is healthy. Bucket metadata load/retry errors include the bucket and a bounded error code, so one failing bucket can be identified without dumping its metadata. No new environment variable, admin authorization action, or recovery command is required. diff --git a/docs/testing/distributed-e2e.md b/docs/testing/distributed-e2e.md index 5d997ddbf..bde2ee748 100644 --- a/docs/testing/distributed-e2e.md +++ b/docs/testing/distributed-e2e.md @@ -29,7 +29,7 @@ The expansion fixture is an all-current-binary fleet, so it initializes pool met - Object Lock COMPLIANCE, GOVERNANCE and bypass, legal hold, bucket default retention, and non-lock bucket rejection - Versioning, exact historical reads, delete-marker removal, and suspended null-version overwrite semantics - Bucket replication between two 4-node clusters, including metadata/tags and target-outage retry; hard quota admission and absence of rejected keys -- Ready/live probes on every node, exact 4-server/16-disk inventory, realtime metrics on every node, and correlated audit-webhook delivery +- Ready/live probes on every node, exact 4-server/16-disk inventory, realtime metrics on every node, and correlated audit-webhook delivery. The observability case runs at WARN and suspends two node processes to model nonresponsive peers: cached drives become unknown immediately, exact per-node HTTP PUT deltas expose sub-quorum failures, the local metadata snapshot does not invent a write latch, and resumed peers allow new writes and byte-identical reads from every node. This models stalled processes, not a physical network partition. - Pool expand, decommission, rebalance, checksum integrity, versioned and multipart data, and S3 during active movement - Bidirectional site-replication convergence plus enabled/synchronized peer state on both sites - A 24-worker mixed PUT/HEAD/GET/COPY/DELETE workload; concurrent PUT during active decommission diff --git a/rustfs/src/admin/handlers/cluster_snapshot.rs b/rustfs/src/admin/handlers/cluster_snapshot.rs index 18cd26168..4d4ec0dcf 100644 --- a/rustfs/src/admin/handlers/cluster_snapshot.rs +++ b/rustfs/src/admin/handlers/cluster_snapshot.rs @@ -193,6 +193,7 @@ impl ClusterSnapshotView { #[derive(Debug, Clone, PartialEq, Eq, Serialize)] #[serde(rename_all = "camelCase")] pub(crate) struct ClusterPoolMetaWriteGateView { + pub state: &'static str, pub writes_ready: bool, pub write_blocked: bool, pub transaction_aborted: bool, @@ -200,11 +201,27 @@ pub(crate) struct ClusterPoolMetaWriteGateView { pub identity_initialized: Option, pub identity_needs_repair: bool, pub cluster_epoch: Option, + #[serde(skip_serializing_if = "Option::is_none")] + pub reason: Option<&'static str>, + #[serde(skip_serializing_if = "Option::is_none")] + pub phase: Option<&'static str>, + #[serde(skip_serializing_if = "Option::is_none")] + pub since_unix_secs: Option, } impl From for ClusterPoolMetaWriteGateView { fn from(snapshot: ClusterPoolMetaWriteGateSnapshot) -> Self { + let state = if snapshot.check_timed_out { + "check_timeout" + } else if snapshot.writes_ready { + "writable" + } else if snapshot.write_blocked || snapshot.transaction_aborted { + "blocked" + } else { + "unavailable" + }; Self { + state, writes_ready: snapshot.writes_ready, write_blocked: snapshot.write_blocked, transaction_aborted: snapshot.transaction_aborted, @@ -212,6 +229,9 @@ impl From for ClusterPoolMetaWriteGateView { identity_initialized: snapshot.identity_initialized, identity_needs_repair: snapshot.identity_needs_repair, cluster_epoch: snapshot.cluster_epoch, + reason: snapshot.reason, + phase: snapshot.phase, + since_unix_secs: snapshot.since_unix_secs, } } } @@ -686,6 +706,7 @@ fn summarize_storage_readiness(snapshot: &ClusterReadOnlySnapshot) -> Capability .filter_map(|reason| match reason { ReadinessDegradedReason::StorageQuorumUnavailable | ReadinessDegradedReason::PoolMetaWriteBlocked + | ReadinessDegradedReason::PoolMetadataCheckTimeout | ReadinessDegradedReason::StorageAndIamUnavailable | ReadinessDegradedReason::StorageAndLockUnavailable | ReadinessDegradedReason::StorageIamAndLockUnavailable => Some(reason.as_str()), @@ -997,7 +1018,9 @@ fn summarize_named_capability_statuses( #[cfg(test)] mod tests { - use super::{ClusterMembershipView, ClusterSnapshotResponse, ClusterSnapshotSummary, ClusterSnapshotView}; + use super::{ + ClusterMembershipView, ClusterPoolMetaWriteGateView, 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::{ @@ -1064,6 +1087,64 @@ mod tests { assert_eq!(value, serde_json::json!({ "snapshot": null })); } + #[test] + fn pool_meta_write_gate_details_are_additive_and_clear_when_not_blocked() { + #[derive(serde::Deserialize)] + #[serde(rename_all = "camelCase")] + struct LegacyGate { + writes_ready: bool, + write_blocked: bool, + transaction_aborted: bool, + } + + let blocked = ClusterPoolMetaWriteGateSnapshot { + writes_ready: false, + transaction_aborted: true, + reason: Some("transaction_unknown"), + phase: Some("publication"), + since_unix_secs: Some(1_700_000_000), + ..Default::default() + }; + let value = serde_json::to_value(ClusterPoolMetaWriteGateView::from(blocked)).unwrap(); + assert_eq!(value["state"], "blocked"); + assert_eq!(value["reason"], "transaction_unknown"); + assert_eq!(value["phase"], "publication"); + assert_eq!(value["sinceUnixSecs"], 1_700_000_000); + assert!(value.get("since_unix_secs").is_none()); + assert!(value.get("operation").is_none()); + assert!(value.get("source").is_none()); + let legacy: LegacyGate = serde_json::from_value(value).unwrap(); + assert!(!legacy.writes_ready); + assert!(!legacy.write_blocked); + assert!(legacy.transaction_aborted); + + for (snapshot, state) in [ + (ClusterPoolMetaWriteGateSnapshot::default(), "writable"), + ( + ClusterPoolMetaWriteGateSnapshot { + writes_ready: false, + check_timed_out: true, + ..Default::default() + }, + "check_timeout", + ), + ( + ClusterPoolMetaWriteGateSnapshot { + writes_ready: false, + ..Default::default() + }, + "unavailable", + ), + ] { + let value = serde_json::to_value(ClusterPoolMetaWriteGateView::from(snapshot)).unwrap(); + assert_eq!(value["state"], state); + assert_eq!(value["writesReady"], state == "writable"); + for field in ["reason", "phase", "sinceUnixSecs"] { + assert!(value.get(field).is_none(), "{state} must not retain {field}: {value}"); + } + } + } + #[tokio::test] async fn cluster_snapshot_discovery_reports_path_without_snapshot() { let response = super::build_cluster_snapshot_discovery_response().await; diff --git a/rustfs/src/app/object/put.rs b/rustfs/src/app/object/put.rs index 3299a66ed..f7672369a 100644 --- a/rustfs/src/app/object/put.rs +++ b/rustfs/src/app/object/put.rs @@ -1995,6 +1995,9 @@ impl DefaultObjectUsecase { storage_error_code = ?diagnostic.storage_code, io_error_kind = ?diagnostic.io_kind, rpc_error_code = ?diagnostic.rpc_code, + pool_metadata_reason = diagnostic.pool_metadata.map(|context| context.reason), + pool_metadata_phase = diagnostic.pool_metadata.map(|context| context.phase), + pool_metadata_since_unix_secs = diagnostic.pool_metadata.map(|context| context.since_unix_secs), source_chain_truncated = diagnostic.truncated, suppressed_errors, "PutObject store write returned" diff --git a/rustfs/src/cluster_snapshot.rs b/rustfs/src/cluster_snapshot.rs index a1433fdec..9d38302a8 100644 --- a/rustfs/src/cluster_snapshot.rs +++ b/rustfs/src/cluster_snapshot.rs @@ -108,24 +108,32 @@ pub struct ClusterListingDiagnosticsSnapshot { #[derive(Debug, Clone, PartialEq, Eq)] pub struct ClusterPoolMetaWriteGateSnapshot { pub writes_ready: bool, + pub check_timed_out: 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, + pub reason: Option<&'static str>, + pub phase: Option<&'static str>, + pub since_unix_secs: Option, } impl Default for ClusterPoolMetaWriteGateSnapshot { fn default() -> Self { Self { writes_ready: true, + check_timed_out: false, write_blocked: false, transaction_aborted: false, pool_meta_absent: false, identity_initialized: None, identity_needs_repair: false, cluster_epoch: None, + reason: None, + phase: None, + since_unix_secs: None, } } } @@ -222,15 +230,22 @@ async fn current_pool_meta_write_gate_snapshot() -> ClusterPoolMetaWriteGateSnap let status = store.pool_meta_write_gate_status().await; ClusterPoolMetaWriteGateSnapshot { writes_ready: status.writes_ready, + check_timed_out: status.check_timed_out, 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, + reason: status.reason, + phase: status.phase, + since_unix_secs: status.since_unix_secs, } } - None => ClusterPoolMetaWriteGateSnapshot::default(), + None => ClusterPoolMetaWriteGateSnapshot { + writes_ready: false, + ..Default::default() + }, } } diff --git a/rustfs/src/error.rs b/rustfs/src/error.rs index be2a86187..50d59a74f 100644 --- a/rustfs/src/error.rs +++ b/rustfs/src/error.rs @@ -13,7 +13,7 @@ // limitations under the License. use crate::storage_api::error::contract::{StorageErrorCode, range::HTTPRangeError}; -use crate::storage_api::error::{QuotaError, StorageError}; +use crate::storage_api::error::{PoolMetadataError, QuotaError, StorageError}; use http::StatusCode; use rustfs_kms::KmsUnavailableError; use s3s::{S3Error, S3ErrorCode}; @@ -89,9 +89,28 @@ pub(crate) struct ApiErrorDiagnostic { pub storage_code: Option, pub io_kind: Option, pub rpc_code: Option, + pub pool_metadata: Option, pub truncated: bool, } +/// Safe projection of local metadata failure context, without operation text or sources. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +pub(crate) struct PoolMetadataDiagnostic { + pub reason: &'static str, + pub phase: &'static str, + pub since_unix_secs: i64, +} + +impl From<&PoolMetadataError> for PoolMetadataDiagnostic { + fn from(context: &PoolMetadataError) -> Self { + Self { + reason: context.kind.as_str(), + phase: context.phase, + since_unix_secs: context.since.unix_timestamp(), + } + } +} + impl ApiError { pub(crate) fn diagnostic(&self) -> ApiErrorDiagnostic { let mut diagnostic = ApiErrorDiagnostic::default(); @@ -106,6 +125,11 @@ impl ApiError { if let Some(status) = error.downcast_ref::() { diagnostic.rpc_code = Some(status.code()); } + if diagnostic.pool_metadata.is_none() + && let Some(context) = error.downcast_ref::() + { + diagnostic.pool_metadata = Some(context.into()); + } current = if let Some(io) = error.downcast_ref::() { diagnostic.io_kind = Some(io.kind()); // io::Error::source can skip the wrapped error itself. @@ -954,6 +978,7 @@ mod tests { #[test] fn pool_metadata_failures_map_to_503_and_preserve_typed_private_context() { use crate::storage_api::error::{PoolMetadataError, PoolMetadataFailure}; + let since = time::OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("fixed failure time"); for kind in [ PoolMetadataFailure::ReadUnavailable, PoolMetadataFailure::RecoveryRequired, @@ -962,9 +987,9 @@ mod tests { ] { let error = StorageError::other(PoolMetadataError { kind, - operation: "pool metadata test".to_owned(), + operation: "private operation path".to_owned(), phase: "prepare_cas", - since: time::OffsetDateTime::now_utc(), + since, source: Some(std::sync::Arc::new(StorageError::other("private disk failure"))), }); let error = StorageError::Io(std::io::Error::new(std::io::ErrorKind::TimedOut, error)); @@ -977,6 +1002,18 @@ mod tests { assert!(!api.message.contains("private")); let source = api.source.as_ref().unwrap().downcast_ref::().unwrap(); assert_eq!(source.pool_metadata_failure().unwrap().kind, kind); + let diagnostic = api.diagnostic(); + assert_eq!( + diagnostic.pool_metadata, + Some(PoolMetadataDiagnostic { + reason: kind.as_str(), + phase: "prepare_cas", + since_unix_secs: since.unix_timestamp(), + }) + ); + assert!(!diagnostic.truncated); + assert!(!format!("{diagnostic:?}").contains("private")); + assert_eq!(api.diagnostic(), diagnostic, "inspection must preserve the original failure time"); } } diff --git a/rustfs/src/server/readiness.rs b/rustfs/src/server/readiness.rs index d87cd1f9d..e7122ba9c 100644 --- a/rustfs/src/server/readiness.rs +++ b/rustfs/src/server/readiness.rs @@ -16,6 +16,7 @@ use crate::server::runtime_sources; use crate::server::{ServiceState, ServiceStateManager}; use crate::server::{has_path_prefix, is_table_catalog_path}; use crate::storage_api::cluster::control_plane::ClusterControlPlane; +use crate::storage_api::error::StorageError; use crate::storage_api::server::readiness::contract::admin::StorageAdminApi; use crate::storage_api::server::readiness::{Endpoint, EndpointServerPools, is_dist_erasure}; #[cfg(test)] @@ -262,7 +263,29 @@ struct StorageReadinessCacheEntry { #[derive(Debug, Clone, Copy, Default, PartialEq, Eq)] struct StorageWriteReadinessStatus { ready: bool, - pool_meta_write_blocked: bool, + pool_metadata_reason: Option, +} + +fn pool_metadata_write_readiness(result: Result<(), StorageError>) -> StorageWriteReadinessStatus { + match result { + Ok(()) => StorageWriteReadinessStatus { + ready: true, + pool_metadata_reason: None, + }, + Err(error) => { + let pool_metadata_reason = if error.pool_metadata_failure().is_some() { + Some(ReadinessDegradedReason::PoolMetaWriteBlocked) + } else if matches!(error, StorageError::Timeout) { + Some(ReadinessDegradedReason::PoolMetadataCheckTimeout) + } else { + None + }; + StorageWriteReadinessStatus { + ready: false, + pool_metadata_reason, + } + } + } } #[derive(Debug, Clone, Copy)] @@ -677,14 +700,12 @@ fn degraded_reasons(readiness: DependencyReadiness) -> Vec, ) -> Vec { let mut reasons = degraded_reasons(readiness); - if pool_meta_write_blocked { + if let Some(reason) = pool_metadata_reason { reasons.retain(|reason| *reason != ReadinessDegradedReason::StorageQuorumUnavailable); - if !reasons.contains(&ReadinessDegradedReason::PoolMetaWriteBlocked) { - reasons.insert(0, ReadinessDegradedReason::PoolMetaWriteBlocked); - } + reasons.insert(0, reason); } reasons } @@ -717,7 +738,7 @@ fn dependency_readiness_report_from_write_status( storage: StorageWriteReadinessStatus, ) -> DependencyReadinessReport { DependencyReadinessReport { - degraded_reasons: degraded_reasons_with_pool_meta_status(readiness, storage.pool_meta_write_blocked), + degraded_reasons: degraded_reasons_with_pool_meta_status(readiness, storage.pool_metadata_reason), readiness, } } @@ -846,11 +867,7 @@ async fn collect_lock_quorum_status() -> LockQuorumStatus { async fn node_pool_meta_write_readiness() -> StorageWriteReadinessStatus { if let Some(store) = runtime_sources::current_object_store_handle() { - let ready = store.pool_meta_writes_ready().await; - return StorageWriteReadinessStatus { - ready, - pool_meta_write_blocked: !ready, - }; + return pool_metadata_write_readiness(store.pool_meta_write_status().await); } StorageWriteReadinessStatus::default() @@ -858,16 +875,14 @@ async fn node_pool_meta_write_readiness() -> StorageWriteReadinessStatus { async fn collect_storage_write_readiness_uncached() -> StorageWriteReadinessStatus { if let Some(store) = runtime_sources::current_object_store_handle() { - if !store.pool_meta_writes_ready().await { - return StorageWriteReadinessStatus { - ready: false, - pool_meta_write_blocked: true, - }; + let metadata = pool_metadata_write_readiness(store.pool_meta_write_status().await); + if !metadata.ready { + return metadata; } let storage_info = StorageAdminApi::storage_info(store.as_ref()).await; StorageWriteReadinessStatus { ready: storage_ready_from_runtime_state(&storage_info), - pool_meta_write_blocked: false, + pool_metadata_reason: None, } } else { StorageWriteReadinessStatus::default() @@ -1900,6 +1915,48 @@ mod tests { ); } + fn blocked_pool_metadata_status() -> StorageWriteReadinessStatus { + use crate::storage_api::error::{PoolMetadataError, PoolMetadataFailure}; + pool_metadata_write_readiness(Err(StorageError::other(PoolMetadataError { + kind: PoolMetadataFailure::TransactionUnknown, + operation: "private operation".to_owned(), + phase: "prepare_cas", + since: time::OffsetDateTime::from_unix_timestamp(1_700_000_000).expect("fixed block time"), + source: Some(Arc::new(StorageError::other("private replica failure"))), + }))) + } + + #[test] + fn pool_metadata_observation_distinguishes_block_timeout_and_unavailable() { + let blocked = blocked_pool_metadata_status(); + assert!(!blocked.ready); + assert_eq!(blocked.pool_metadata_reason, Some(ReadinessDegradedReason::PoolMetaWriteBlocked)); + assert!(!format!("{blocked:?}").contains("private")); + let writable = pool_metadata_write_readiness(Ok(())); + assert!(writable.ready); + assert_eq!(writable.pool_metadata_reason, None); + let timed_out = pool_metadata_write_readiness(Err(StorageError::Timeout)); + assert!(!timed_out.ready); + assert_eq!(timed_out.pool_metadata_reason, Some(ReadinessDegradedReason::PoolMetadataCheckTimeout)); + let unavailable = pool_metadata_write_readiness(Err(StorageError::other("pool metadata writes remain blocked"))); + assert!(!unavailable.ready); + assert_eq!( + unavailable.pool_metadata_reason, None, + "error text alone must not be interpreted as a typed write block" + ); + + let readiness = DependencyReadiness { + storage_ready: false, + iam_ready: true, + lock_quorum_ready: true, + peer_health_ready: true, + }; + let report = dependency_readiness_report_from_write_status(readiness, timed_out); + assert!(!report.readiness.storage_ready, "inspection timeout must remain fail-closed"); + assert_eq!(report.degraded_reasons, vec![ReadinessDegradedReason::PoolMetadataCheckTimeout]); + assert_eq!(report.degraded_reasons[0].as_str(), "pool_metadata_check_timeout"); + } + #[test] fn degraded_reasons_report_pool_meta_write_blocked() { let readiness = DependencyReadiness { @@ -1910,7 +1967,7 @@ mod tests { }; assert_eq!( - degraded_reasons_with_pool_meta_status(readiness, true), + degraded_reasons_with_pool_meta_status(readiness, blocked_pool_metadata_status().pool_metadata_reason), vec![ReadinessDegradedReason::PoolMetaWriteBlocked] ); } @@ -1925,7 +1982,7 @@ mod tests { }; assert_eq!( - degraded_reasons_with_pool_meta_status(readiness, true), + degraded_reasons_with_pool_meta_status(readiness, blocked_pool_metadata_status().pool_metadata_reason), vec![ ReadinessDegradedReason::PoolMetaWriteBlocked, ReadinessDegradedReason::StorageAndLockUnavailable, diff --git a/rustfs/src/shared_types.rs b/rustfs/src/shared_types.rs index ebbafc834..7406a0978 100644 --- a/rustfs/src/shared_types.rs +++ b/rustfs/src/shared_types.rs @@ -44,6 +44,7 @@ pub enum ReadinessDegradedReason { ObjectReadStalled, ObjectWriteStalled, PoolMetaWriteBlocked, + PoolMetadataCheckTimeout, ClusterHealthTimeout, PeerHealthUnavailable, StartupFinalizationPending, @@ -63,6 +64,7 @@ impl ReadinessDegradedReason { ReadinessDegradedReason::ObjectReadStalled => "object_read_stalled", ReadinessDegradedReason::ObjectWriteStalled => "object_write_stalled", ReadinessDegradedReason::PoolMetaWriteBlocked => "pool_meta_write_blocked", + ReadinessDegradedReason::PoolMetadataCheckTimeout => "pool_metadata_check_timeout", ReadinessDegradedReason::ClusterHealthTimeout => "cluster_health_timeout", ReadinessDegradedReason::PeerHealthUnavailable => "peer_health_unavailable", ReadinessDegradedReason::StartupFinalizationPending => "startup_finalization_pending", diff --git a/rustfs/src/storage/storage_api.rs b/rustfs/src/storage/storage_api.rs index 67cf8938c..612168e2a 100644 --- a/rustfs/src/storage/storage_api.rs +++ b/rustfs/src/storage/storage_api.rs @@ -476,11 +476,12 @@ pub(crate) mod ecstore_disk { } pub(crate) mod ecstore_error { - pub(crate) use rustfs_ecstore::api::error::{ - Error, Result, StorageError, is_err_bucket_not_found, is_err_object_not_found, is_err_version_not_found, - }; #[cfg(test)] - pub(crate) use rustfs_ecstore::api::error::{PoolMetadataError, PoolMetadataFailure}; + pub(crate) use rustfs_ecstore::api::error::PoolMetadataFailure; + pub(crate) use rustfs_ecstore::api::error::{ + Error, PoolMetadataError, Result, StorageError, is_err_bucket_not_found, is_err_object_not_found, + is_err_version_not_found, + }; } pub(crate) mod ecstore_event { diff --git a/rustfs/src/storage_api.rs b/rustfs/src/storage_api.rs index dc284be5c..a98683e6a 100644 --- a/rustfs/src/storage_api.rs +++ b/rustfs/src/storage_api.rs @@ -83,8 +83,9 @@ pub(crate) mod error { } } + pub(crate) use crate::storage::storage_api::ecstore_error::PoolMetadataError; #[cfg(test)] - pub(crate) use crate::storage::storage_api::ecstore_error::{PoolMetadataError, PoolMetadataFailure}; + pub(crate) use crate::storage::storage_api::ecstore_error::PoolMetadataFailure; pub(crate) use crate::storage::storage_api::{QuotaError, StorageError}; }