gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-09-06 12:09:12 +00:00
Code Issues Packages Projects Releases Wiki Activity

test(startup): observe pool repair classification and replicas

Browse Source
This commit is contained in:
overtrue
2026-09-06 16:38:09 +08:00
parent 6233466c5d
commit b4e0838b01
2 changed files with 112 additions and 6 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/ecstore/src/core/pools.rs
+93 -2
View File
@@ -5108,7 +5108,49 @@ async fn read_pool_meta_replicas<S>(pools: Vec<Arc<S>>, no_lock: bool) -> Vec<Po
where
S: EcstoreObjectIO,
{
join_all(pools.into_iter().map(|pool| read_pool_meta_replica(pool, no_lock))).await
let reads = join_all(pools.into_iter().map(|pool| read_pool_meta_replica(pool, no_lock))).await;
#[cfg(feature = "e2e-test-hooks")]
if STARTUP_CAS_OBSERVATION.try_with(|_| ()).is_ok() {
let batch = uuid::Uuid::new_v4();
for (pool, read) in reads.iter().enumerate() {
let mut observation = serde_json::json!({
"kind": "replica-read", "object": POOL_META_NAME, "batch": batch, "pool": pool,
"cas": match &read.cas {
PoolMetaCasToken::Missing => "missing",
PoolMetaCasToken::Existing(_) => "existing",
PoolMetaCasToken::Unsafe => "unsafe",
},
"etag": match &read.cas { PoolMetaCasToken::Existing(etag) => Some(etag), _ => None },
});
match &read.replica {
PoolMetaReplica::Valid {
raw,
canonical,
meta,
revision,
committed,
..
} => {
observation["state"] = serde_json::json!("valid");
observation["committed"] = serde_json::json!(committed);
observation["version"] = serde_json::json!(revision.version);
observation["cluster_id"] = serde_json::json!(revision.cluster_id);
observation["epoch"] = serde_json::json!(revision.epoch);
observation["generation"] = serde_json::json!(revision.generation);
observation["transaction_id"] = serde_json::json!(revision.transaction_id);
observation["pool_count"] = serde_json::json!(meta.pools.len());
observation["payload_sha256"] = serde_json::json!(rustfs_utils::crypto::hex(Sha256::digest(canonical)));
observation["raw_sha256"] = serde_json::json!(rustfs_utils::crypto::hex(Sha256::digest(raw)));
}
PoolMetaReplica::Missing => observation["state"] = serde_json::json!("missing"),
PoolMetaReplica::Corrupt(_) => observation["state"] = serde_json::json!("corrupt"),
PoolMetaReplica::Incompatible(_) => observation["state"] = serde_json::json!("incompatible"),
PoolMetaReplica::Unreadable(_) => observation["state"] = serde_json::json!("unreadable"),
}
startup_cas_test_observe(observation);
}
}
reads
}
fn select_pool_meta_replicas_observing<R>(write_state: &mut PoolMetaWriteState, replicas: Vec<R>) -> Result<PoolMetaSelection>
@@ -5480,9 +5522,44 @@ fn pool_meta_cas_preconditions(token: &PoolMetaCasToken, object: &str) -> Result
}
}
#[cfg(feature = "e2e-test-hooks")]
struct StartupCasObservation {
attempt: uuid::Uuid,
phase: &'static str,
pools: Vec<usize>,
}
#[cfg(feature = "e2e-test-hooks")]
tokio::task_local! {
static STARTUP_CAS_OBSERVATION: StartupCasObservation;
}
// This scope follows only the directly polled startup future. Spawned work
// does not inherit it; receiver evidence retains its existing RPC tuple.
#[cfg(feature = "e2e-test-hooks")]
pub(crate) async fn startup_cas_test_scope<S, F: std::future::Future>(
attempt: uuid::Uuid,
phase: &'static str,
pools: &[Arc<S>],
future: F,
) -> F::Output {
STARTUP_CAS_OBSERVATION
.scope(
StartupCasObservation {
attempt,
phase,
// These identities are never dereferenced or logged. The
// caller and operation keep the same pool Arcs alive.
pools: pools.iter().map(|pool| Arc::as_ptr(pool) as usize).collect(),
},
future,
)
.await
}
// Direct JSON diagnostics are independent of the startup tracing subscriber.
#[cfg(feature = "e2e-test-hooks")]
fn startup_cas_test_observe(mut observation: serde_json::Value) {
pub(crate) fn startup_cas_test_observe(mut observation: serde_json::Value) {
let Some(nonce) = std::env::var("RUSTFS_E2E_STARTUP_CAS_NONCE")
.ok()
.and_then(|value| uuid::Uuid::parse_str(&value).ok())
@@ -5491,6 +5568,10 @@ fn startup_cas_test_observe(mut observation: serde_json::Value) {
};
observation["nonce"] = serde_json::json!(nonce);
observation["pid"] = serde_json::json!(std::process::id());
let _ = STARTUP_CAS_OBSERVATION.try_with(|scope| {
observation["attempt"] = serde_json::json!(scope.attempt);
observation["startup_phase"] = serde_json::json!(scope.phase);
});
let line = format!("RUSTFS_E2E_STARTUP_CAS {observation}\n");
let _ = std::io::Write::write_all(&mut std::io::stderr().lock(), line.as_bytes());
}
@@ -5519,6 +5600,9 @@ where
let observation = std::env::var_os("RUSTFS_E2E_STARTUP_CAS_NONCE").map(|_| {
serde_json::json!({
"kind": "cas", "object": object, "phase": phase,
"pool": STARTUP_CAS_OBSERVATION.try_with(|scope| {
scope.pools.iter().position(|identity| *identity == Arc::as_ptr(&pool) as usize)
}).ok().flatten(),
"payload_sha256": rustfs_utils::crypto::hex(Sha256::digest(&data)),
"if_match": opts.http_preconditions.as_ref().and_then(|p| p.if_match.as_deref()),
"if_none_match": opts.http_preconditions.as_ref().and_then(|p| p.if_none_match.as_deref()),
@@ -5538,6 +5622,13 @@ where
if let Some(mut observation) = observation {
observation["ok"] = serde_json::json!(result.is_ok());
observation["etag"] = serde_json::json!(result.as_ref().ok().and_then(|info| info.etag.as_deref()));
observation["mod_time"] = serde_json::json!(
result
.as_ref()
.ok()
.and_then(|info| info.mod_time)
.map(|time| time.unix_timestamp_nanos().to_string())
);
observation["error"] = serde_json::json!(result.as_ref().err().map(ToString::to_string));
startup_cas_test_observe(observation);
}
crates/ecstore/src/store/init.rs
+19 -4
View File
@@ -630,14 +630,27 @@ impl ECStore {
.pools
.first()
.is_some_and(|pool| pool_first_endpoint_is_local(&pool.endpoints));
#[cfg(feature = "e2e-test-hooks")]
let startup_attempt = uuid::Uuid::new_v4();
let (meta, pool_meta_replica_state) = {
let mut write_state = self.pool_meta_save_gate.lock().await;
establish_pool_meta_bootstrap_identity_if_proven(self.pools.clone(), &mut write_state, should_persist_pool_meta)
.await
.map_err(|err| Error::other(format!("store init failed during establish_pool_meta_bootstrap_identity: {err}")))?;
load_pool_meta_for_startup(self.pools.clone(), &mut write_state).await?
let load = load_pool_meta_for_startup(self.pools.clone(), &mut write_state);
#[cfg(feature = "e2e-test-hooks")]
let load = crate::core::pools::startup_cas_test_scope(startup_attempt, "load", &self.pools, load);
load.await?
};
let update = meta.validate(self.pools.clone())?;
#[cfg(feature = "e2e-test-hooks")]
crate::core::pools::startup_cas_test_observe(serde_json::json!({
"kind": "startup-classifier", "attempt": startup_attempt,
"elected_writer": should_persist_pool_meta,
"needs_repair": pool_meta_replica_state.needs_repair,
"repair_write_safe": pool_meta_replica_state.repair_write_safe,
"topology_update": update,
}));
let endpoints = runtime_sources::endpoint_pools_or_default();
let mut installed_pool_meta = if update {
@@ -649,15 +662,17 @@ impl ECStore {
// distributed startup can race on the same lock and replay the prior init bug.
{
let mut write_state = self.pool_meta_save_gate.lock().await;
installed_pool_meta = persist_pool_meta_for_startup_if_safe(
let persist = persist_pool_meta_for_startup_if_safe(
&installed_pool_meta,
self.pools.clone(),
pool_meta_replica_state,
&mut write_state,
update,
should_persist_pool_meta,
)
.await?;
);
#[cfg(feature = "e2e-test-hooks")]
let persist = crate::core::pools::startup_cas_test_scope(startup_attempt, "persist", &self.pools, persist);
installed_pool_meta = persist.await?;
}
{