From 69b0c828c94c3a6bb75a16dc8817225837199b55 Mon Sep 17 00:00:00 2001 From: junxiang Mu <1948535941@qq.com> Date: Tue, 22 Jul 2025 11:08:01 +0800 Subject: [PATCH] fix: scanner add heal bucket Signed-off-by: junxiang Mu <1948535941@qq.com> --- crates/ahm/src/scanner/data_scanner.rs | 154 +++++++++++++++++++++++++ 1 file changed, 154 insertions(+) diff --git a/crates/ahm/src/scanner/data_scanner.rs b/crates/ahm/src/scanner/data_scanner.rs index 037879268..68f66d8a2 100644 --- a/crates/ahm/src/scanner/data_scanner.rs +++ b/crates/ahm/src/scanner/data_scanner.rs @@ -429,9 +429,16 @@ impl Scanner { info!("Scanning EC set {} in pool {}", set_index, pool_index); + // list all bucket for heal bucket + // Get online disks from this EC set let (disks, _) = set_disks.get_online_disks_with_healing(false).await; + // Check volume consistency across disks and heal missing buckets + if !disks.is_empty() { + self.check_and_heal_missing_volumes(&disks, set_index, pool_index).await?; + } + if disks.is_empty() { warn!("No online disks available for EC set {} in pool {}", set_index, pool_index); return Ok(Vec::new()); @@ -1196,6 +1203,91 @@ impl Scanner { Ok(()) } + /// Check volume consistency across disks and heal missing buckets + async fn check_and_heal_missing_volumes(&self, disks: &[DiskStore], set_index: usize, pool_index: usize) -> Result<()> { + info!("Checking volume consistency for EC set {} in pool {}", set_index, pool_index); + + // Step 1: Collect bucket lists from all online disks + let mut disk_bucket_lists = Vec::new(); + let mut all_buckets = std::collections::HashSet::new(); + + for (disk_idx, disk) in disks.iter().enumerate() { + match disk.list_volumes().await { + Ok(volumes) => { + let bucket_names: Vec = volumes.iter().map(|v| v.name.clone()).collect(); + for bucket in &bucket_names { + all_buckets.insert(bucket.clone()); + } + disk_bucket_lists.push((disk_idx, bucket_names)); + debug!("Disk {} has {} buckets", disk_idx, volumes.len()); + } + Err(e) => { + warn!("Failed to list volumes on disk {}: {}", disk_idx, e); + disk_bucket_lists.push((disk_idx, Vec::new())); + } + } + } + + // Step 2: Find missing buckets on each disk + let mut missing_buckets_count = 0; + for (disk_idx, disk_buckets) in &disk_bucket_lists { + let disk_bucket_set: std::collections::HashSet<_> = disk_buckets.iter().collect(); + let missing_buckets: Vec<_> = all_buckets + .iter() + .filter(|bucket| !disk_bucket_set.contains(bucket)) + .collect(); + + if !missing_buckets.is_empty() { + missing_buckets_count += missing_buckets.len(); + warn!("Disk {} is missing {} buckets: {:?}", disk_idx, missing_buckets.len(), missing_buckets); + + // Step 3: Submit heal tasks for missing buckets + let enable_healing = self.config.read().await.enable_healing; + if enable_healing { + if let Some(heal_manager) = &self.heal_manager { + for bucket in missing_buckets { + let req = crate::heal::HealRequest::bucket(bucket.clone()); + match heal_manager.submit_heal_request(req).await { + Ok(task_id) => { + info!( + "Submitted bucket heal task {} for missing bucket '{}' on disk {}", + task_id, bucket, disk_idx + ); + } + Err(e) => { + error!("Failed to submit bucket heal task for '{}' on disk {}: {}", bucket, disk_idx, e); + } + } + } + } else { + warn!("Healing is enabled but no heal manager available"); + } + } else { + info!("Healing is disabled, skipping bucket heal tasks"); + } + } + } + + if missing_buckets_count > 0 { + warn!( + "Found {} missing bucket instances across {} disks in EC set {} (pool {})", + missing_buckets_count, + disks.len(), + set_index, + pool_index + ); + } else { + info!( + "All buckets are consistent across {} disks in EC set {} (pool {})", + disks.len(), + set_index, + pool_index + ); + } + + Ok(()) + } + /// Clone scanner for background tasks fn clone_for_background(&self) -> Self { Self { @@ -1460,4 +1552,66 @@ mod tests { // clean up temp dir let _ = std::fs::remove_dir_all(std::path::Path::new(TEST_DIR_USAGE_STATS)); } + + #[tokio::test(flavor = "multi_thread")] + #[serial] + async fn test_volume_healing_functionality() { + const TEST_DIR_VOLUME_HEAL: &str = "/tmp/rustfs_ahm_test_volume_heal"; + let (disk_paths, ecstore) = prepare_test_env(Some(TEST_DIR_VOLUME_HEAL), Some(9003)).await; + + // Create test buckets + let bucket1 = "test-bucket-1"; + let bucket2 = "test-bucket-2"; + + ecstore.make_bucket(bucket1, &Default::default()).await.unwrap(); + ecstore.make_bucket(bucket2, &Default::default()).await.unwrap(); + + // Add some test objects + let mut pr1 = PutObjReader::from_vec(b"test data 1".to_vec()); + ecstore + .put_object(bucket1, "obj1", &mut pr1, &Default::default()) + .await + .unwrap(); + + let mut pr2 = PutObjReader::from_vec(b"test data 2".to_vec()); + ecstore + .put_object(bucket2, "obj2", &mut pr2, &Default::default()) + .await + .unwrap(); + + // Simulate missing bucket on one disk by removing bucket directory + let disk1_bucket1_path = disk_paths[0].join(bucket1); + if disk1_bucket1_path.exists() { + println!("Removing bucket directory to simulate missing volume: {disk1_bucket1_path:?}"); + match fs::remove_dir_all(&disk1_bucket1_path) { + Ok(_) => println!("Successfully removed bucket directory from disk 0"), + Err(e) => println!("Failed to remove bucket directory: {e}"), + } + } + + // Create scanner without heal manager for now (testing the detection logic) + let scanner = Scanner::new(None, None); + + // Enable healing in config + { + let mut config = scanner.config.write().await; + config.enable_healing = true; + } + + println!("=== Testing volume healing functionality ==="); + + // Run scan cycle which should detect missing volume + // The new check_and_heal_missing_volumes function should be called + let scan_result = scanner.scan_cycle().await; + assert!(scan_result.is_ok(), "Scan cycle should succeed"); + + // Get metrics to verify scan completed + let metrics = scanner.get_metrics().await; + assert!(metrics.total_cycles > 0, "Should have completed scan cycles"); + println!("Volume healing detection test completed successfully"); + println!("Scan metrics: {metrics:?}"); + + // Clean up + let _ = std::fs::remove_dir_all(std::path::Path::new(TEST_DIR_VOLUME_HEAL)); + } }