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

heal: verify admin recreate pool metadata (#7474)

Browse Source 
Co-authored-by: zhi22915 <qiuzgang@gmail.com>
This commit is contained in:
houseme
2026-09-08 16:57:13 +08:00
committed by GitHub
parent 0b05b6c6ff
commit cc5487e7de
6 changed files with 252 additions and 18 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/heal/src/heal/erasure_healer.rs
+81 -10
View File
@@ -113,6 +113,7 @@ pub struct ErasureSetHealer {
heal_opts: HealOpts,
source: HealRequestSource,
target_endpoints: Arc<[String]>,
pool_metadata_target_endpoints: Arc<[String]>,
replacement_task_id: Option<String>,
replacement_target_identities: Option<Arc<[ReplacementTargetIdentity]>>,
mainline_pacer: Option<Arc<super::pacing::MainlinePacer>>,
@@ -362,6 +363,7 @@ impl ErasureSetHealer {
heal_opts,
source,
target_endpoints: Vec::new().into(),
pool_metadata_target_endpoints: Vec::new().into(),
replacement_task_id: None,
replacement_target_identities: None,
mainline_pacer: None,
@@ -385,6 +387,13 @@ impl ErasureSetHealer {
self
}
pub(crate) fn with_pool_metadata_targets(mut self, mut target_endpoints: Vec<String>) -> Self {
target_endpoints.sort_unstable();
target_endpoints.dedup();
self.pool_metadata_target_endpoints = target_endpoints.into();
self
}
pub(crate) fn with_replacement_identity_fence(
mut self,
replacement_target_identities: Option<Vec<ReplacementTargetIdentity>>,
@@ -948,10 +957,16 @@ impl ErasureSetHealer {
resume_manager: &ResumeManager,
checkpoint_manager: &CheckpointManager,
) -> Result<()> {
if self.replacement_task_id.is_none() {
let target_endpoints = if self.pool_metadata_target_endpoints.is_empty() {
self.target_endpoints.as_ref()
} else {
self.pool_metadata_target_endpoints.as_ref()
};
let target_scoped_recreate = !self.heal_opts.dry_run && self.heal_opts.recreate && !target_endpoints.is_empty();
if self.replacement_task_id.is_none() && !target_scoped_recreate {
return Ok(());
}
if self.target_endpoints.is_empty() {
if target_endpoints.is_empty() {
return Err(Error::TaskExecutionFailed {
message: "Replacement pool metadata heal requires target endpoints".to_string(),
});
@@ -978,17 +993,11 @@ impl ErasureSetHealer {
.heal_object(RUSTFS_META_BUCKET, POOL_META_NAME, None, &self.heal_opts)
.await
{
Ok((result, None)) if target_outcomes_complete(&result, &self.target_endpoints) => {
Ok((result, None)) if target_outcomes_complete(&result, target_endpoints) => {
let object_size = result_object_size_u64(&result);
match self
.storage
.replacement_targets_have_version(
RUSTFS_META_BUCKET,
POOL_META_NAME,
None,
&self.heal_opts,
&self.target_endpoints,
)
.replacement_targets_have_version(RUSTFS_META_BUCKET, POOL_META_NAME, None, &self.heal_opts, target_endpoints)
.await
{
Ok(true) => (object_size, Ok(())),
@@ -2762,6 +2771,68 @@ mod resume_loop_tests {
assert_eq!(env.storage.calls(), vec![(POOL_META_NAME.to_string(), None)]);
}
#[tokio::test]
async fn admin_recreate_target_heals_pool_metadata_before_completion() {
let env = make_env_with_targets(vec!["replacement-a".to_string()]).await;
let healer = ErasureSetHealer::new(
env.storage.clone(),
Arc::new(RwLock::new(HealProgress::new())),
CancellationToken::new(),
env.healer.disk.clone(),
HealOpts {
recreate: true,
pool: Some(0),
set: Some(0),
..Default::default()
},
HealRequestSource::Admin,
)
.with_pool_metadata_targets(vec!["replacement-a".to_string()]);
env.storage
.set_result(POOL_META_NAME, None, replacement_target_ok_result("replacement-a", POOL_META_NAME));
healer
.execute_heal_with_resume(&["b".to_string()], "pool_0_set_0", &env.resume, &env.checkpoint)
.await
.expect("admin recreate should heal and verify pool metadata");
assert!(env.resume.get_state().await.completed);
assert_eq!(env.storage.calls(), vec![(POOL_META_NAME.to_string(), None)]);
}
#[tokio::test]
async fn admin_recreate_pool_metadata_readback_failure_keeps_resume_state() {
let env = make_env_with_targets(vec!["replacement-a".to_string()]).await;
let healer = ErasureSetHealer::new(
env.storage.clone(),
Arc::new(RwLock::new(HealProgress::new())),
CancellationToken::new(),
env.healer.disk.clone(),
HealOpts {
recreate: true,
pool: Some(0),
set: Some(0),
..Default::default()
},
HealRequestSource::Admin,
)
.with_pool_metadata_targets(vec!["replacement-a".to_string()]);
env.storage
.set_result(POOL_META_NAME, None, replacement_target_ok_result("replacement-a", POOL_META_NAME));
env.storage.set_replacement_commit_evidence(POOL_META_NAME, None, false);
let error = healer
.execute_heal_with_resume(&["b".to_string()], "pool_0_set_0", &env.resume, &env.checkpoint)
.await
.expect_err("unconfirmed admin recreate pool metadata must keep the set incomplete");
assert!(error.to_string().contains("Erasure set heal incomplete"));
let state = env.resume.get_state().await;
assert!(!state.completed);
assert_eq!(state.retry_count, 1);
assert_eq!(env.storage.calls(), vec![(POOL_META_NAME.to_string(), None)]);
}
#[tokio::test]
async fn retry_exhaustion_keeps_resume_artifacts_for_recovery() {
let env = make_env().await;
crates/heal/src/heal/task.rs
+1 -1
View File
@@ -45,7 +45,7 @@ use tokio::sync::RwLock;
use tracing::{debug, error, info, warn};
use uuid::Uuid;
use super::{BUCKET_META_PREFIX, DATA_USAGE_CACHE_NAME, RUSTFS_META_BUCKET};
use super::{BUCKET_META_PREFIX, DATA_USAGE_CACHE_NAME, POOL_META_NAME, RUSTFS_META_BUCKET};
#[cfg(test)]
pub(crate) struct OutcomeFinishTestHook {
crates/heal/src/heal/task/heal_bucket.rs
+79
View File
@@ -340,9 +340,88 @@ impl HealTask {
return Err(self.record_batch_failure(failure).await);
}
if self.options.recreate_missing && !self.options.dry_run {
self.heal_cluster_pool_metadata().await?;
}
Ok(())
}
async fn heal_cluster_pool_metadata(&self) -> Result<()> {
let heal_opts = HealOpts {
recursive: false,
dry_run: self.options.dry_run,
remove: false,
recreate: self.options.recreate_missing,
scan_mode: self.options.scan_mode,
update_parity: self.options.update_parity,
no_lock: self.options.no_lock,
read_repair: false,
pool: self.options.pool_index,
set: self.options.set_index,
};
let heal_result = self
.await_with_control(self.storage.heal_object(RUSTFS_META_BUCKET, POOL_META_NAME, None, &heal_opts))
.await;
match heal_result {
Ok((result, None)) => {
debug!(
target: "rustfs::heal::task",
event = EVENT_HEAL_BUCKET_RESULT,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
bucket = RUSTFS_META_BUCKET,
object = POOL_META_NAME,
drives_healed = result.drives_healed(),
drives_total = result.drives_reported(),
result = "pool_metadata_ok",
"Heal cluster pool metadata repaired"
);
self.record_result_item(result).await;
Ok(())
}
Ok((result, Some(err))) => {
self.record_result_item(result).await;
warn!(
target: "rustfs::heal::task",
event = EVENT_HEAL_BUCKET_RESULT,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
bucket = RUSTFS_META_BUCKET,
object = POOL_META_NAME,
result = "pool_metadata_failed",
error = %err,
"Heal cluster pool metadata failed"
);
Err(Error::TaskExecutionFailed {
message: format!("Failed to heal cluster pool metadata: {err}"),
})
}
Err(Error::TaskCancelled) => Err(Error::TaskCancelled),
Err(Error::TaskTimeout) => Err(Error::TaskTimeout),
Err(err) => {
warn!(
target: "rustfs::heal::task",
event = EVENT_HEAL_BUCKET_RESULT,
component = LOG_COMPONENT_HEAL,
subsystem = LOG_SUBSYSTEM_TASK,
task_id = %self.id,
bucket = RUSTFS_META_BUCKET,
object = POOL_META_NAME,
result = "pool_metadata_failed",
error = %err,
"Heal cluster pool metadata failed"
);
Err(Error::TaskExecutionFailed {
message: format!("Failed to heal cluster pool metadata: {err}"),
})
}
}
}
pub(super) async fn heal_prefix(&self, bucket: &str, prefix: &str) -> Result<()> {
debug!(
target: "rustfs::heal::task",
crates/heal/src/heal/task/heal_erasure_set.rs
+5
View File
@@ -451,6 +451,11 @@ impl HealTask {
self.source,
)
.with_replacement_targets(replacement_targets, is_auto_replacement.then(|| self.id.clone()))
.with_pool_metadata_targets(if self.options.recreate_missing && !self.options.dry_run {
self.heal_endpoints.clone()
} else {
Vec::new()
})
.with_replacement_identity_fence(replacement_target_identities.clone())
.with_mainline_pacer(self.mainline_pacer.clone());
crates/heal/src/heal/task/heal_object.rs
-5
View File
@@ -162,11 +162,6 @@ impl HealTask {
pool: self.options.pool_index,
set: self.options.set_index,
};
let expected_bucket_incarnation_id = self.storage.bucket_incarnation_id(bucket).await?;
let mut expected_identity =
self.outcome_identity(bucket, object, version_id, self.options.pool_index, self.options.set_index);
expected_identity.bucket_incarnation_id = expected_bucket_incarnation_id;
let mut expected_identity =
self.outcome_identity(bucket, object, version_id, self.options.pool_index, self.options.set_index);
expected_identity.bucket_incarnation_id = self.outcome_bucket_incarnation_id(bucket, self.options.dry_run).await?;
crates/heal/src/heal/task/tests.rs
+86 -2
View File
@@ -66,7 +66,7 @@ mod canonical_outcome {
assert_eq!(task.get_progress().await.objects_scanned, 2);
assert_eq!(
storage.heal_object_calls.lock().expect("object calls").as_slice(),
["object-a", "object-b"]
["object-a", "object-b", POOL_META_NAME]
);
assert_eq!(
storage.listing_tokens.lock().expect("listing tokens").as_slice(),
@@ -2582,11 +2582,95 @@ async fn test_cluster_heal_visits_bucket_objects() {
assert_eq!(
storage.healed_objects.lock().unwrap().as_slice(),
["object-a".to_string(), "object-b".to_string()]
["object-a".to_string(), "object-b".to_string(), POOL_META_NAME.to_string()]
);
assert!(matches!(task.get_status().await, HealTaskStatus::Completed));
}
#[tokio::test]
async fn cluster_recreate_heals_pool_metadata_after_user_buckets() {
let storage = Arc::new(MockStorage::default());
let request = HealRequest::new(
HealType::Cluster,
HealOptions {
recursive: true,
recreate_missing: true,
timeout: None,
..Default::default()
},
HealPriority::Normal,
);
let task = HealTask::from_request(request, storage.clone());
task.execute()
.await
.expect("cluster recreate heal should include pool metadata");
assert_eq!(
storage.heal_object_calls.lock().expect("object calls").as_slice(),
["object-a".to_string(), "object-b".to_string(), POOL_META_NAME.to_string()]
);
let opts = storage.object_heal_opts.lock().expect("object opts");
assert!(opts.last().expect("pool metadata opts").recreate);
}
#[tokio::test]
async fn cluster_recreate_fails_when_pool_metadata_heal_fails() {
let storage = Arc::new(MockStorage::default());
storage.heal_object_outcomes.lock().expect("object outcomes").insert(
POOL_META_NAME.to_string(),
VecDeque::from([MockHealObjectOutcome::ErrOther("pool metadata missing")]),
);
let request = HealRequest::new(
HealType::Cluster,
HealOptions {
recursive: true,
recreate_missing: true,
timeout: None,
..Default::default()
},
HealPriority::Normal,
);
let task = HealTask::from_request(request, storage.clone());
let err = task
.execute()
.await
.expect_err("cluster recreate heal must not hide pool metadata failure");
assert!(matches!(err, Error::TaskExecutionFailed { .. }));
assert_eq!(
storage.heal_object_calls.lock().expect("object calls").as_slice(),
["object-a".to_string(), "object-b".to_string(), POOL_META_NAME.to_string()]
);
}
#[tokio::test]
async fn cluster_dry_run_does_not_heal_pool_metadata() {
let storage = Arc::new(MockStorage::default());
let request = HealRequest::new(
HealType::Cluster,
HealOptions {
recursive: true,
dry_run: true,
recreate_missing: true,
timeout: None,
..Default::default()
},
HealPriority::Normal,
);
let task = HealTask::from_request(request, storage.clone());
task.execute()
.await
.expect("dry-run cluster heal should preserve existing coverage");
assert_eq!(
storage.heal_object_calls.lock().expect("object calls").as_slice(),
["object-a".to_string(), "object-b".to_string()]
);
}
#[tokio::test]
async fn object_heal_skips_dangling_delete_grace_without_failing_task() {
let storage = Arc::new(MockStorage {