// Copyright 2024 RustFS Team // // Licensed under the Apache License, Version 2.0 (the "License"); // you may not use this file except in compliance with the License. // You may obtain a copy of the License at // // http://www.apache.org/licenses/LICENSE-2.0 // // Unless required by applicable law or agreed to in writing, software // distributed under the License is distributed on an "AS IS" BASIS, // WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. // See the License for the specific language governing permissions and // limitations under the License. use std::{ collections::HashMap, sync::Arc, time::{Duration, SystemTime}, }; use ecstore::{ disk::{DiskAPI, DiskStore, WalkDirOptions}, set_disk::SetDisks, }; use rustfs_ecstore::{self as ecstore, data_usage::store_data_usage_in_backend, StorageAPI}; use rustfs_filemeta::MetacacheReader; use tokio::sync::{Mutex, RwLock}; use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, warn}; use super::metrics::{BucketMetrics, DiskMetrics, MetricsCollector, ScannerMetrics}; use crate::heal::HealManager; use crate::{ error::{Error, Result}, get_ahm_services_cancel_token, HealRequest, }; use rustfs_common::{ data_usage::DataUsageInfo, metrics::{globalMetrics, Metric, Metrics}, }; use rustfs_ecstore::disk::RUSTFS_META_BUCKET; /// Custom scan mode enum for AHM scanner #[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] pub enum ScanMode { /// Normal scan - basic object discovery and metadata collection #[default] Normal, /// Deep scan - includes EC verification and integrity checks Deep, } /// Scanner configuration #[derive(Debug, Clone)] pub struct ScannerConfig { /// Scan interval between cycles pub scan_interval: Duration, /// Deep scan interval (how often to perform deep scan) pub deep_scan_interval: Duration, /// Maximum concurrent scans pub max_concurrent_scans: usize, /// Whether to enable healing pub enable_healing: bool, /// Whether to enable metrics collection pub enable_metrics: bool, /// Current scan mode (normal, deep) pub scan_mode: ScanMode, /// Whether to enable data usage statistics collection pub enable_data_usage_stats: bool, } impl Default for ScannerConfig { fn default() -> Self { Self { scan_interval: Duration::from_secs(60), // 1 minute deep_scan_interval: Duration::from_secs(3600), // 1 hour max_concurrent_scans: 20, enable_healing: true, enable_metrics: true, scan_mode: ScanMode::Normal, enable_data_usage_stats: true, } } } /// Scanner state #[derive(Debug, Default)] pub struct ScannerState { /// Whether scanner is running pub is_running: bool, /// Current scan cycle pub current_cycle: u64, /// Last scan start time pub last_scan_start: Option, /// Last scan end time pub last_scan_end: Option, /// Current scan duration pub current_scan_duration: Option, /// Last deep scan time pub last_deep_scan_time: Option, /// Buckets being scanned pub scanning_buckets: Vec, /// Disks being scanned pub scanning_disks: Vec, } /// AHM Scanner - Automatic Health Management Scanner /// /// This scanner monitors the health of objects in the RustFS storage system. /// It integrates with ECStore to perform real data scanning across all EC sets /// and collects metrics. /// /// The scanner operates on the entire ECStore, scanning all EC (Erasure Coding) sets, /// where each set contains multiple disks that store the same objects with different shards. pub struct Scanner { /// Scanner configuration config: Arc>, /// Scanner state state: Arc>, /// Local metrics collector (for backward compatibility) metrics: Arc, /// Bucket metrics cache bucket_metrics: Arc>>, /// Disk metrics cache disk_metrics: Arc>>, /// Data usage statistics cache data_usage_stats: Arc>>, /// Last data usage statistics collection time last_data_usage_collection: Arc>>, /// Heal manager for auto-heal integration heal_manager: Option>, } impl Scanner { /// Create a new scanner pub fn new(config: Option, heal_manager: Option>) -> Self { let config = config.unwrap_or_default(); info!("Creating AHM scanner for all EC sets"); Self { config: Arc::new(RwLock::new(config)), state: Arc::new(RwLock::new(ScannerState::default())), metrics: Arc::new(MetricsCollector::new()), bucket_metrics: Arc::new(Mutex::new(HashMap::new())), disk_metrics: Arc::new(Mutex::new(HashMap::new())), data_usage_stats: Arc::new(Mutex::new(HashMap::new())), last_data_usage_collection: Arc::new(RwLock::new(None)), heal_manager, } } /// Set the heal manager after construction pub fn set_heal_manager(&mut self, heal_manager: Arc) { self.heal_manager = Some(heal_manager); } /// Start the scanner pub async fn start(&self) -> Result<()> { let mut state = self.state.write().await; if state.is_running { warn!("Scanner is already running"); return Ok(()); } state.is_running = true; state.last_scan_start = Some(SystemTime::now()); info!("Starting AHM scanner"); // Start background scan loop let scanner = self.clone_for_background(); tokio::spawn(async move { if let Err(e) = scanner.scan_loop().await { error!("Scanner loop failed: {}", e); } }); Ok(()) } /// Stop the scanner gracefully pub async fn stop(&self) -> Result<()> { let mut state = self.state.write().await; if !state.is_running { warn!("Scanner is not running"); return Ok(()); } info!("Stopping AHM scanner gracefully..."); // Trigger cancellation using global cancel token if let Some(cancel_token) = get_ahm_services_cancel_token() { cancel_token.cancel(); } state.is_running = false; state.last_scan_end = Some(SystemTime::now()); if let Some(start_time) = state.last_scan_start { state.current_scan_duration = Some(SystemTime::now().duration_since(start_time).unwrap_or(Duration::ZERO)); } info!("AHM scanner stopped"); Ok(()) } /// Get integrated data usage statistics for DataUsageInfoHandler pub async fn get_data_usage_info(&self) -> Result { let mut integrated_info = DataUsageInfo::new(); // Collect data from all buckets { let data_usage_guard = self.data_usage_stats.lock().await; for (bucket_name, bucket_data) in data_usage_guard.iter() { let _bucket_name = bucket_name; // Merge bucket data into integrated info integrated_info.merge(bucket_data); } } // Update capacity information from storage info if let Some(ecstore) = rustfs_ecstore::new_object_layer_fn() { let mut total_capacity = 0u64; let mut total_used_capacity = 0u64; let mut total_free_capacity = 0u64; // Collect capacity info from all SetDisks for pool in &ecstore.pools { for set_disks in &pool.disk_set { let (disks, _) = set_disks.get_online_disks_with_healing(false).await; for disk in disks { if let Ok(disk_info) = disk .disk_info(&ecstore::disk::DiskInfoOptions { disk_id: disk.path().to_string_lossy().to_string(), metrics: true, noop: false, }) .await { total_capacity += disk_info.total; total_used_capacity += disk_info.used; total_free_capacity += disk_info.free; } } } } if total_capacity > 0 { integrated_info.update_capacity(total_capacity, total_used_capacity, total_free_capacity); } } Ok(integrated_info) } /// Get current scanner metrics pub async fn get_metrics(&self) -> ScannerMetrics { let mut metrics = self.metrics.get_metrics(); // Add bucket metrics let bucket_metrics: HashMap = { let bucket_metrics_guard = self.bucket_metrics.lock().await; bucket_metrics_guard .iter() .map(|(key, value)| (key.clone(), value.clone())) .collect() }; metrics.bucket_metrics = bucket_metrics; // Add disk metrics let disk_metrics: HashMap = { let disk_metrics_guard = self.disk_metrics.lock().await; disk_metrics_guard .iter() .map(|(key, value)| (key.clone(), value.clone())) .collect() }; metrics.disk_metrics = disk_metrics; // Add current scan duration let state = self.state.read().await; metrics.current_scan_duration = state.current_scan_duration; metrics } /// Get global metrics from common crate pub async fn get_global_metrics(&self) -> rustfs_madmin::metrics::ScannerMetrics { globalMetrics.report().await } /// Perform a single scan cycle pub async fn scan_cycle(&self) -> Result<()> { let start_time = SystemTime::now(); // Start global metrics collection for this cycle let stop_fn = Metrics::time(Metric::ScanCycle); info!("Starting scan cycle {} for all EC sets", self.metrics.get_metrics().current_cycle + 1); // Update state { let mut state = self.state.write().await; state.current_cycle += 1; state.last_scan_start = Some(start_time); state.scanning_buckets.clear(); state.scanning_disks.clear(); } // Update global metrics cycle information let cycle_info = rustfs_common::metrics::CurrentCycle { current: self.state.read().await.current_cycle, cycle_completed: vec![chrono::Utc::now()], started: chrono::Utc::now(), }; globalMetrics.set_cycle(Some(cycle_info)).await; self.metrics.set_current_cycle(self.state.read().await.current_cycle); self.metrics.increment_total_cycles(); // Get ECStore and all SetDisks let Some(ecstore) = rustfs_ecstore::new_object_layer_fn() else { warn!("No ECStore available for scanning"); return Ok(()); }; // Get all SetDisks from all pools let mut all_set_disks = Vec::new(); for pool in &ecstore.pools { for set_disks in &pool.disk_set { all_set_disks.push(set_disks.clone()); } } if all_set_disks.is_empty() { warn!("No EC sets available for scanning"); return Ok(()); } info!("Scanning {} EC sets across {} pools", all_set_disks.len(), ecstore.pools.len()); // Phase 1: Scan all SetDisks concurrently let config = self.config.read().await; let semaphore = Arc::new(tokio::sync::Semaphore::new(config.max_concurrent_scans)); drop(config); let mut scan_futures = Vec::new(); for set_disks in all_set_disks { let semaphore = semaphore.clone(); let scanner = self.clone_for_background(); let future = async move { let _permit = semaphore.acquire().await.unwrap(); scanner.scan_set_disks(set_disks).await }; scan_futures.push(future); } // Wait for all scans to complete let mut results = Vec::new(); for future in scan_futures { results.push(future.await); } // Check results and collect object metadata let mut successful_scans = 0; let mut failed_scans = 0; let mut all_disk_objects = Vec::new(); for result in results { match result { Ok(disk_objects) => { successful_scans += 1; all_disk_objects.extend(disk_objects); } Err(e) => { failed_scans += 1; error!("SetDisks scan failed: {}", e); } } } // Phase 2: Analyze object distribution and perform EC verification if successful_scans > 0 { // Get all disks from all SetDisks for analysis let mut all_disks = Vec::new(); for pool in &ecstore.pools { for set_disks in &pool.disk_set { let (disks, _) = set_disks.get_online_disks_with_healing(false).await; all_disks.extend(disks); } } if let Err(e) = self.analyze_object_distribution(&all_disk_objects, &all_disks).await { error!("Object distribution analysis failed: {}", e); } } // Update scan duration let scan_duration = SystemTime::now().duration_since(start_time).unwrap_or(Duration::ZERO); { let mut state = self.state.write().await; state.last_scan_end = Some(SystemTime::now()); state.current_scan_duration = Some(scan_duration); } // Complete global metrics collection for this cycle stop_fn(); info!( "Completed scan cycle in {:?} ({} successful, {} failed)", scan_duration, successful_scans, failed_scans ); Ok(()) } /// Verify object integrity and trigger healing if necessary async fn verify_object_integrity(&self, bucket: &str, object: &str) -> Result<()> { debug!("Starting verify_object_integrity for {}/{}", bucket, object); let config = self.config.read().await; if !config.enable_healing || config.scan_mode != ScanMode::Deep { debug!("Healing disabled or not in deep scan mode, skipping verification"); return Ok(()); } if let Some(ecstore) = rustfs_ecstore::new_object_layer_fn() { // First try the standard integrity check let object_opts = ecstore::store_api::ObjectOptions::default(); let mut integrity_failed = false; debug!("Running standard object verification for {}/{}", bucket, object); match ecstore.verify_object_integrity(bucket, object, &object_opts).await { Ok(_) => { debug!("Standard verification passed for {}/{}", bucket, object); // Standard verification passed, now check for missing data parts match self.check_data_parts_integrity(bucket, object).await { Ok(_) => { // Object is completely healthy debug!("Data parts integrity check passed for {}/{}", bucket, object); self.metrics.increment_healthy_objects(); } Err(e) => { // Data parts are missing or corrupt debug!("Data parts integrity check failed for {}/{}: {}", bucket, object, e); warn!("Data parts integrity check failed for {}/{}: {}. Triggering heal.", bucket, object, e); integrity_failed = true; } } } Err(e) => { // Standard object verification failed debug!("Standard verification failed for {}/{}: {}", bucket, object, e); warn!("Object verification failed for {}/{}: {}. Triggering heal.", bucket, object, e); integrity_failed = true; } } debug!("integrity_failed = {} for {}/{}", integrity_failed, bucket, object); if integrity_failed { self.metrics.increment_corrupted_objects(); if let Some(heal_manager) = &self.heal_manager { debug!("Submitting heal request for {}/{}", bucket, object); let heal_request = HealRequest::object(bucket.to_string(), object.to_string(), None); if let Err(e) = heal_manager.submit_heal_request(heal_request).await { error!("Failed to submit heal task for {}/{}: {}", bucket, object, e); } else { debug!("Successfully submitted heal request for {}/{}", bucket, object); } } else { debug!("No heal manager available for {}/{}", bucket, object); } } } else { debug!("No ECStore available for {}/{}", bucket, object); } debug!("Completed verify_object_integrity for {}/{}", bucket, object); Ok(()) } /// Check data parts integrity by verifying all parts exist on disks async fn check_data_parts_integrity(&self, bucket: &str, object: &str) -> Result<()> { debug!("Checking data parts integrity for {}/{}", bucket, object); if let Some(ecstore) = rustfs_ecstore::new_object_layer_fn() { // Get object info let object_info = match ecstore.get_object_info(bucket, object, &Default::default()).await { Ok(info) => info, Err(e) => { return Err(Error::Other(format!("Failed to get object info: {}", e))); } }; debug!( "Object info for {}/{}: data_blocks={}, parity_blocks={}, parts={}", bucket, object, object_info.data_blocks, object_info.parity_blocks, object_info.parts.len() ); // Create FileInfo from ObjectInfo let file_info = rustfs_filemeta::FileInfo { volume: bucket.to_string(), name: object.to_string(), version_id: object_info.version_id, is_latest: object_info.is_latest, deleted: object_info.delete_marker, size: object_info.size, mod_time: object_info.mod_time, parts: object_info .parts .iter() .map(|p| rustfs_filemeta::ObjectPartInfo { etag: p.etag.clone(), number: 0, // Will be set by erasure info size: p.size, actual_size: p.actual_size, mod_time: p.mod_time, index: p.index.clone(), checksums: p.checksums.clone(), }) .collect(), erasure: rustfs_filemeta::ErasureInfo { algorithm: "ReedSolomon".to_string(), data_blocks: object_info.data_blocks, parity_blocks: object_info.parity_blocks, block_size: 0, // Default value index: 1, // Default index distribution: (1..=object_info.data_blocks + object_info.parity_blocks).collect(), checksums: vec![], }, ..Default::default() }; // Get all disks from ECStore's disk_map let mut has_missing_parts = false; let mut total_disks_checked = 0; let mut disks_with_errors = 0; debug!("Checking {} pools in disk_map", ecstore.disk_map.len()); for (pool_idx, pool_disks) in &ecstore.disk_map { debug!("Checking pool {}, {} disks", pool_idx, pool_disks.len()); for (disk_idx, disk_option) in pool_disks.iter().enumerate() { if let Some(disk) = disk_option { total_disks_checked += 1; debug!("Checking disk {} in pool {}: {}", disk_idx, pool_idx, disk.path().display()); match disk.check_parts(bucket, object, &file_info).await { Ok(check_result) => { debug!( "check_parts returned {} results for disk {}", check_result.results.len(), disk.path().display() ); // Check if any parts are missing or corrupt for (part_idx, &result) in check_result.results.iter().enumerate() { debug!("Part {} result: {} on disk {}", part_idx, result, disk.path().display()); if result == 4 || result == 5 { // CHECK_PART_FILE_NOT_FOUND or CHECK_PART_FILE_CORRUPT has_missing_parts = true; disks_with_errors += 1; warn!( "Found missing or corrupt part {} for object {}/{} on disk {} (pool {}): result={}", part_idx, bucket, object, disk.path().display(), pool_idx, result ); break; } } } Err(e) => { disks_with_errors += 1; warn!("Failed to check parts on disk {}: {}", disk.path().display(), e); // Continue checking other disks } } if has_missing_parts { break; // No need to check other disks if we found missing parts } } else { debug!("Disk {} in pool {} is None", disk_idx, pool_idx); } } if has_missing_parts { break; // No need to check other pools if we found missing parts } } debug!( "Data parts check completed for {}/{}: total_disks={}, disks_with_errors={}, has_missing_parts={}", bucket, object, total_disks_checked, disks_with_errors, has_missing_parts ); if has_missing_parts { return Err(Error::Other(format!("Object has missing or corrupt data parts: {}/{}", bucket, object))); } } debug!("Data parts integrity verified for {}/{}", bucket, object); Ok(()) } /// Scan a single SetDisks (EC set) async fn scan_set_disks( &self, set_disks: Arc, ) -> Result>>> { let set_index = set_disks.set_index; let pool_index = set_disks.pool_index; 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()); } info!("Scanning {} online disks in EC set {} (pool {})", disks.len(), set_index, pool_index); // Scan all disks in this SetDisks concurrently let config = self.config.read().await; let semaphore = Arc::new(tokio::sync::Semaphore::new(config.max_concurrent_scans)); drop(config); let mut scan_futures = Vec::new(); for disk in disks { let semaphore = semaphore.clone(); let scanner = self.clone_for_background(); let future = async move { let _permit = semaphore.acquire().await.unwrap(); scanner.scan_disk(&disk).await }; scan_futures.push(future); } // Wait for all scans to complete let mut results = Vec::new(); for future in scan_futures { results.push(future.await); } // Check results and collect object metadata let mut successful_scans = 0; let mut failed_scans = 0; let mut all_disk_objects = Vec::new(); for result in results { match result { Ok(disk_objects) => { successful_scans += 1; all_disk_objects.push(disk_objects); } Err(e) => { failed_scans += 1; error!("Disk scan failed in EC set {} (pool {}): {}", set_index, pool_index, e); // Add empty map for failed disk all_disk_objects.push(HashMap::new()); } } } info!( "Completed scanning EC set {} (pool {}): {} successful, {} failed", set_index, pool_index, successful_scans, failed_scans ); Ok(all_disk_objects) } /// Scan a single disk async fn scan_disk(&self, disk: &DiskStore) -> Result>> { let disk_path = disk.path().to_string_lossy().to_string(); // Start global metrics collection for disk scan let stop_fn = Metrics::time(Metric::ScanBucketDrive); info!("Scanning disk: {}", disk_path); // Update disk metrics { let mut disk_metrics_guard = self.disk_metrics.lock().await; let metrics = disk_metrics_guard.entry(disk_path.clone()).or_insert_with(|| DiskMetrics { disk_path: disk_path.clone(), ..Default::default() }); metrics.is_scanning = true; metrics.last_scan_time = Some(SystemTime::now()); // Get disk info using DiskStore's disk_info interface if let Ok(disk_info) = disk .disk_info(&ecstore::disk::DiskInfoOptions { disk_id: disk_path.clone(), metrics: true, noop: false, }) .await { metrics.total_space = disk_info.total; metrics.used_space = disk_info.used; metrics.free_space = disk_info.free; metrics.is_online = disk.is_online().await; // check disk status, if offline, submit erasure set heal task if !metrics.is_online { let enable_healing = self.config.read().await.enable_healing; if enable_healing { if let Some(heal_manager) = &self.heal_manager { // Get bucket list for erasure set healing let buckets = match rustfs_ecstore::new_object_layer_fn() { Some(ecstore) => match ecstore.list_bucket(&ecstore::store_api::BucketOptions::default()).await { Ok(buckets) => buckets.iter().map(|b| b.name.clone()).collect::>(), Err(e) => { error!("Failed to get bucket list for disk healing: {}", e); return Err(Error::Storage(e)); } }, None => { error!("No ECStore available for getting bucket list"); return Err(Error::Storage(ecstore::error::StorageError::other("No ECStore available"))); } }; let set_disk_id = format!("pool_{}_set_{}", disk.endpoint().pool_idx, disk.endpoint().set_idx); let req = HealRequest::new( crate::heal::task::HealType::ErasureSet { buckets, set_disk_id }, crate::heal::task::HealOptions::default(), crate::heal::task::HealPriority::High, ); match heal_manager.submit_heal_request(req).await { Ok(task_id) => { warn!("disk offline, submit erasure set heal task: {} {}", task_id, disk_path); } Err(e) => { error!("disk offline, submit erasure set heal task failed: {} {}", disk_path, e); } } } } } // Additional disk info for debugging debug!( "Disk {}: total={}, used={}, free={}, online={}", disk_path, disk_info.total, disk_info.used, disk_info.free, metrics.is_online ); } } // Update state { let mut state = self.state.write().await; state.scanning_disks.push(disk_path.clone()); } // List volumes (buckets) on this disk let volumes = match disk.list_volumes().await { Ok(volumes) => volumes, Err(e) => { error!("Failed to list volumes on disk {}: {}", disk_path, e); // disk access failed, submit erasure set heal task let enable_healing = self.config.read().await.enable_healing; if enable_healing { if let Some(heal_manager) = &self.heal_manager { // Get bucket list for erasure set healing let buckets = match rustfs_ecstore::new_object_layer_fn() { Some(ecstore) => match ecstore.list_bucket(&ecstore::store_api::BucketOptions::default()).await { Ok(buckets) => buckets.iter().map(|b| b.name.clone()).collect::>(), Err(e) => { error!("Failed to get bucket list for disk healing: {}", e); return Err(Error::Storage(e)); } }, None => { error!("No ECStore available for getting bucket list"); return Err(Error::Storage(ecstore::error::StorageError::other("No ECStore available"))); } }; let set_disk_id = format!("pool_{}_set_{}", disk.endpoint().pool_idx, disk.endpoint().set_idx); let req = HealRequest::new( crate::heal::task::HealType::ErasureSet { buckets, set_disk_id }, crate::heal::task::HealOptions::default(), crate::heal::task::HealPriority::Urgent, ); match heal_manager.submit_heal_request(req).await { Ok(task_id) => { warn!("disk access failed, submit erasure set heal task: {} {}", task_id, disk_path); } Err(heal_err) => { error!("disk access failed, submit erasure set heal task failed: {} {}", disk_path, heal_err); } } } } return Err(Error::Storage(e.into())); } }; // Scan each volume and collect object metadata let mut disk_objects = HashMap::new(); for volume in volumes { // check cancel token if let Some(cancel_token) = get_ahm_services_cancel_token() { if cancel_token.is_cancelled() { info!("Cancellation requested, stopping disk scan"); break; } } match self.scan_volume(disk, &volume.name).await { Ok(object_metadata) => { disk_objects.insert(volume.name, object_metadata); } Err(e) => { error!("Failed to scan volume {} on disk {}: {}", volume.name, disk_path, e); continue; } } } // Update disk metrics after scan { let mut disk_metrics_guard = self.disk_metrics.lock().await; if let Some(existing_metrics) = disk_metrics_guard.get(&disk_path) { let mut updated_metrics = existing_metrics.clone(); updated_metrics.is_scanning = false; disk_metrics_guard.insert(disk_path.clone(), updated_metrics); } } // Update state { let mut state = self.state.write().await; state.scanning_disks.retain(|d| d != &disk_path); } // Complete global metrics collection for disk scan stop_fn(); Ok(disk_objects) } /// Scan a single volume (bucket) and collect object information /// /// This method collects all objects from a disk for a specific bucket. /// It returns a map of object names to their metadata for later analysis. async fn scan_volume(&self, disk: &DiskStore, bucket: &str) -> Result> { // Start global metrics collection for volume scan let stop_fn = Metrics::time(Metric::ScanObject); info!("Scanning bucket: {} on disk: {}", bucket, disk.to_string()); // Initialize bucket metrics if not exists { let mut bucket_metrics_guard = self.bucket_metrics.lock().await; bucket_metrics_guard .entry(bucket.to_string()) .or_insert_with(|| BucketMetrics { bucket: bucket.to_string(), ..Default::default() }); } // Update state { let mut state = self.state.write().await; state.scanning_buckets.push(bucket.to_string()); } self.metrics.increment_bucket_scans_started(1); let scan_start = SystemTime::now(); // Walk through all objects in the bucket let walk_opts = WalkDirOptions { bucket: bucket.to_string(), base_dir: String::new(), recursive: true, report_notfound: false, filter_prefix: None, forward_to: None, limit: 0, disk_id: String::new(), }; // Use a buffer to collect scan results for processing let mut scan_buffer = Vec::new(); if let Err(e) = disk.walk_dir(walk_opts, &mut scan_buffer).await { error!("Failed to walk directory for bucket {}: {}", bucket, e); return Err(Error::Storage(e.into())); } // Process the scan results using MetacacheReader let mut reader = MetacacheReader::new(std::io::Cursor::new(scan_buffer)); let mut objects_scanned = 0u64; let mut objects_with_issues = 0u64; let mut object_metadata = HashMap::new(); // Process each object entry while let Ok(Some(mut entry)) = reader.peek().await { objects_scanned += 1; // Check if this is an actual object (not just a directory) if entry.is_object() { debug!("Scanned object: {}", entry.name); // Parse object metadata if let Ok(file_meta) = entry.xl_meta() { if file_meta.versions.is_empty() { objects_with_issues += 1; warn!("Object {} has no versions", entry.name); // 对象元数据损坏，提交元数据heal任务 let enable_healing = self.config.read().await.enable_healing; if enable_healing { if let Some(heal_manager) = &self.heal_manager { let req = HealRequest::metadata(bucket.to_string(), entry.name.clone()); match heal_manager.submit_heal_request(req).await { Ok(task_id) => { warn!( "object metadata damaged, submit heal task: {} {} / {}", task_id, bucket, entry.name ); } Err(e) => { error!( "object metadata damaged, submit heal task failed: {} / {} {}", bucket, entry.name, e ); } } } } } else { // Store object metadata for later analysis object_metadata.insert(entry.name.clone(), file_meta.clone()); } } else { objects_with_issues += 1; warn!("Failed to parse metadata for object {}", entry.name); // 对象元数据解析失败，提交元数据heal任务 let enable_healing = self.config.read().await.enable_healing; if enable_healing { if let Some(heal_manager) = &self.heal_manager { let req = HealRequest::metadata(bucket.to_string(), entry.name.clone()); match heal_manager.submit_heal_request(req).await { Ok(task_id) => { warn!( "object metadata parse failed, submit heal task: {} {} / {}", task_id, bucket, entry.name ); } Err(e) => { error!( "object metadata parse failed, submit heal task failed: {} / {} {}", bucket, entry.name, e ); } } } } } } } // Update metrics self.metrics.increment_objects_scanned(objects_scanned); self.metrics.increment_objects_with_issues(objects_with_issues); self.metrics.increment_bucket_scans_finished(1); // Update bucket metrics { let mut bucket_metrics_guard = self.bucket_metrics.lock().await; if let Some(existing_metrics) = bucket_metrics_guard.get(bucket) { let mut updated_metrics = existing_metrics.clone(); updated_metrics.total_objects = objects_scanned; updated_metrics.objects_with_issues = objects_with_issues; updated_metrics.scan_duration = Some(SystemTime::now().duration_since(scan_start).unwrap_or(Duration::ZERO)); bucket_metrics_guard.insert(bucket.to_string(), updated_metrics); } } // Update state { let mut state = self.state.write().await; state.scanning_buckets.retain(|b| b != bucket); } // Complete global metrics collection for volume scan stop_fn(); debug!( "Completed scanning bucket: {} on disk {} ({} objects, {} issues)", bucket, disk.to_string(), objects_scanned, objects_with_issues ); Ok(object_metadata) } /// Analyze object distribution across all disks and perform EC verification /// /// This method takes the collected object metadata from all disks and: /// 1. Creates a union of all objects across all disks /// 2. Identifies missing objects on each disk (for healing) /// 3. Performs EC decode verification for deep scan mode async fn analyze_object_distribution( &self, all_disk_objects: &[HashMap>], disks: &[DiskStore], ) -> Result<()> { info!("Analyzing object distribution across {} disks", disks.len()); // Step 1: Create union of all objects across all disks let mut all_objects = HashMap::new(); // bucket -> Set let mut object_locations = HashMap::new(); // (bucket, object) -> Vec for (disk_idx, disk_objects) in all_disk_objects.iter().enumerate() { for (bucket, objects) in disk_objects { if bucket == RUSTFS_META_BUCKET { // Skip internal system bucket during analysis to speed up tests continue; } // Add bucket to all_objects let bucket_objects = all_objects .entry(bucket.clone()) .or_insert_with(std::collections::HashSet::new); for (object_name, _file_meta) in objects.iter() { bucket_objects.insert(object_name.clone()); // Record which disk has this object let key = (bucket.clone(), object_name.clone()); let locations = object_locations.entry(key).or_insert_with(Vec::new); locations.push(disk_idx); } } } info!( "Found {} buckets with {} total objects", all_objects.len(), all_objects.values().map(|s| s.len()).sum::() ); // Step 2: Identify missing objects and perform EC verification let mut objects_needing_heal = 0u64; let mut objects_with_ec_issues = 0u64; for (bucket, objects) in &all_objects { // Skip internal RustFS system bucket to avoid lengthy checks on temporary/trash objects if bucket == RUSTFS_META_BUCKET { continue; } for object_name in objects { let key = (bucket.clone(), object_name.clone()); let empty_vec = Vec::new(); let locations = object_locations.get(&key).unwrap_or(&empty_vec); // Check if object is missing from some disks if locations.len() < disks.len() { objects_needing_heal += 1; let missing_disks: Vec = (0..disks.len()).filter(|&i| !locations.contains(&i)).collect(); warn!("Object {}/{} missing from disks: {:?}", bucket, object_name, missing_disks); // submit heal task let enable_healing = self.config.read().await.enable_healing; if enable_healing { if let Some(heal_manager) = &self.heal_manager { use crate::heal::{HealPriority, HealRequest}; let req = HealRequest::new( crate::heal::HealType::Object { bucket: bucket.clone(), object: object_name.clone(), version_id: None, }, crate::heal::HealOptions::default(), HealPriority::High, ); match heal_manager.submit_heal_request(req).await { Ok(task_id) => { warn!( "object missing, submit heal task: {} {} / {} (missing disks: {:?})", task_id, bucket, object_name, missing_disks ); } Err(e) => { error!("object missing, submit heal task failed: {} / {} {}", bucket, object_name, e); } } } } } // Step 3: Deep scan EC verification let config = self.config.read().await; if config.scan_mode == ScanMode::Deep { if let Err(e) = self.verify_object_integrity(bucket, object_name).await { objects_with_ec_issues += 1; warn!("Object integrity verification failed for object {}/{}: {}", bucket, object_name, e); } } } } info!( "Analysis complete: {} objects need healing, {} objects have EC issues", objects_needing_heal, objects_with_ec_issues ); // Step 4: Collect data usage statistics if enabled let config = self.config.read().await; if config.enable_data_usage_stats { if let Err(e) = self.collect_data_usage_statistics(all_disk_objects).await { error!("Failed to collect data usage statistics: {}", e); } } drop(config); Ok(()) } /// Background scan loop with graceful shutdown async fn scan_loop(self) -> Result<()> { let config = self.config.read().await; let mut interval = tokio::time::interval(config.scan_interval); let deep_scan_interval = config.deep_scan_interval; drop(config); // Get global cancel token let cancel_token = if let Some(global_token) = get_ahm_services_cancel_token() { global_token.clone() } else { CancellationToken::new() }; loop { tokio::select! { _ = interval.tick() => { // Check if scanner should still be running if !self.state.read().await.is_running { break; } // 检查取消信号 if cancel_token.is_cancelled() { info!("Cancellation requested, exiting scanner loop"); break; } // Determine if it's time for a deep scan let current_time = SystemTime::now(); let last_deep_scan_time = self.state.read().await.last_deep_scan_time.unwrap_or(SystemTime::UNIX_EPOCH); if current_time.duration_since(last_deep_scan_time).unwrap_or(Duration::ZERO) >= deep_scan_interval { info!("Deep scan interval reached, switching to deep scan mode"); self.config.write().await.scan_mode = ScanMode::Deep; self.state.write().await.last_deep_scan_time = Some(current_time); } // Perform scan cycle if let Err(e) = self.scan_cycle().await { error!("Scan cycle failed: {}", e); } } _ = cancel_token.cancelled() => { info!("Received cancellation, stopping scanner loop"); break; } } } info!("Scanner loop stopped"); Ok(()) } /// Collect data usage statistics from scanned objects async fn collect_data_usage_statistics( &self, all_disk_objects: &[HashMap>], ) -> Result<()> { info!("Collecting data usage statistics from {} disk scans", all_disk_objects.len()); let mut data_usage = DataUsageInfo::default(); // Collect objects from all disks (avoid duplicates by using first occurrence) let mut processed_objects = std::collections::HashSet::new(); for disk_objects in all_disk_objects { for (bucket_name, objects) in disk_objects { if bucket_name == RUSTFS_META_BUCKET { continue; // skip internal bucket from data usage stats } for object_name in objects.keys() { let object_key = format!("{bucket_name}/{object_name}"); // Skip if already processed (avoid duplicates across disks) if !processed_objects.insert(object_key.clone()) { continue; } // Add object to data usage statistics (pass entire FileMeta for accurate version counting) data_usage.add_object_from_file_meta(&object_key, objects.get(object_name).unwrap()); } } } // Ensure buckets_count is correctly set data_usage.buckets_count = data_usage.buckets_usage.len() as u64; // Log statistics before storing info!( "Collected data usage statistics: {} buckets, {} total objects, {} total size", data_usage.buckets_count, data_usage.objects_total_count, data_usage.objects_total_size ); // Store in cache and update last collection time let current_time = SystemTime::now(); { let mut data_usage_guard = self.data_usage_stats.lock().await; data_usage_guard.insert("current".to_string(), data_usage.clone()); } { let mut last_collection = self.last_data_usage_collection.write().await; *last_collection = Some(current_time); } // Store to backend if configured (spawned to avoid blocking scan loop) let config = self.config.read().await; if config.enable_data_usage_stats { if let Some(store) = rustfs_ecstore::new_object_layer_fn() { // Offload persistence to background task let data_clone = data_usage.clone(); tokio::spawn(async move { if let Err(e) = store_data_usage_in_backend(data_clone, store).await { error!("Failed to store data usage statistics to backend: {}", e); } else { info!("Successfully stored data usage statistics to backend"); } }); } else { warn!("Storage not available, skipping backend persistence"); } } 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 { config: self.config.clone(), state: Arc::clone(&self.state), metrics: Arc::clone(&self.metrics), bucket_metrics: Arc::clone(&self.bucket_metrics), disk_metrics: Arc::clone(&self.disk_metrics), data_usage_stats: Arc::clone(&self.data_usage_stats), last_data_usage_collection: Arc::clone(&self.last_data_usage_collection), heal_manager: self.heal_manager.clone(), } } } #[cfg(test)] mod tests { use super::*; use crate::heal::manager::HealConfig; use rustfs_ecstore::data_usage::load_data_usage_from_backend; use rustfs_ecstore::disk::endpoint::Endpoint; use rustfs_ecstore::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints}; use rustfs_ecstore::store::ECStore; use rustfs_ecstore::{ StorageAPI, store_api::{MakeBucketOptions, ObjectIO, PutObjReader}, }; use serial_test::serial; use std::fs; use std::net::SocketAddr; use std::sync::OnceLock; // Global test environment cache to avoid repeated initialization static GLOBAL_TEST_ENV: OnceLock<(Vec, Arc)> = OnceLock::new(); async fn prepare_test_env(test_dir: Option<&str>, port: Option) -> (Vec, Arc) { // Check if global environment is already initialized if let Some((disk_paths, ecstore)) = GLOBAL_TEST_ENV.get() { return (disk_paths.clone(), ecstore.clone()); } // create temp dir as 4 disks let test_base_dir = test_dir.unwrap_or("/tmp/rustfs_ahm_test"); let temp_dir = std::path::PathBuf::from(test_base_dir); if temp_dir.exists() { fs::remove_dir_all(&temp_dir).unwrap(); } fs::create_dir_all(&temp_dir).unwrap(); // create 4 disk dirs let disk_paths = vec![ temp_dir.join("disk1"), temp_dir.join("disk2"), temp_dir.join("disk3"), temp_dir.join("disk4"), ]; for disk_path in &disk_paths { fs::create_dir_all(disk_path).unwrap(); } // create EndpointServerPools let mut endpoints = Vec::new(); for (i, disk_path) in disk_paths.iter().enumerate() { let mut endpoint = Endpoint::try_from(disk_path.to_str().unwrap()).unwrap(); // set correct index endpoint.set_pool_index(0); endpoint.set_set_index(0); endpoint.set_disk_index(i); endpoints.push(endpoint); } let pool_endpoints = PoolEndpoints { legacy: false, set_count: 1, drives_per_set: 4, endpoints: Endpoints::from(endpoints), cmd_line: "test".to_string(), platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH), }; let endpoint_pools = EndpointServerPools(vec![pool_endpoints]); // format disks rustfs_ecstore::store::init_local_disks(endpoint_pools.clone()).await.unwrap(); // create ECStore with dynamic port let port = port.unwrap_or(9000); let server_addr: SocketAddr = format!("127.0.0.1:{port}").parse().unwrap(); let ecstore = ECStore::new(server_addr, endpoint_pools).await.unwrap(); // init bucket metadata system let buckets_list = ecstore .list_bucket(&rustfs_ecstore::store_api::BucketOptions { no_metadata: true, ..Default::default() }) .await .unwrap(); let buckets = buckets_list.into_iter().map(|v| v.name).collect(); rustfs_ecstore::bucket::metadata_sys::init_bucket_metadata_sys(ecstore.clone(), buckets).await; // Store in global cache let _ = GLOBAL_TEST_ENV.set((disk_paths.clone(), ecstore.clone())); (disk_paths, ecstore) } #[tokio::test(flavor = "multi_thread")] #[serial] async fn test_scanner_basic_functionality() { const TEST_DIR_BASIC: &str = "/tmp/rustfs_ahm_test_basic"; let (disk_paths, ecstore) = prepare_test_env(Some(TEST_DIR_BASIC), Some(9001)).await; // create some test data let bucket_name = "test-bucket"; let object_name = "test-object"; let test_data = b"Hello, RustFS!"; // create bucket and verify let bucket_opts = MakeBucketOptions::default(); ecstore .make_bucket(bucket_name, &bucket_opts) .await .expect("make_bucket failed"); // check bucket really exists let buckets = ecstore .list_bucket(&rustfs_ecstore::store_api::BucketOptions::default()) .await .unwrap(); assert!(buckets.iter().any(|b| b.name == bucket_name), "bucket not found after creation"); // write object let mut put_reader = PutObjReader::from_vec(test_data.to_vec()); let object_opts = rustfs_ecstore::store_api::ObjectOptions::default(); ecstore .put_object(bucket_name, object_name, &mut put_reader, &object_opts) .await .expect("put_object failed"); // create Scanner and test basic functionality let scanner = Scanner::new(None, None); // Test 1: Normal scan - verify object is found println!("=== Test 1: Normal scan ==="); let scan_result = scanner.scan_cycle().await; assert!(scan_result.is_ok(), "Normal scan should succeed"); let metrics = scanner.get_metrics().await; assert!(metrics.objects_scanned > 0, "Objects scanned should be positive"); println!("Normal scan completed successfully"); // Test 2: Simulate disk corruption - delete object data from disk1 println!("=== Test 2: Simulate disk corruption ==="); let disk1_bucket_path = disk_paths[0].join(bucket_name); let disk1_object_path = disk1_bucket_path.join(object_name); // Try to delete the object file from disk1 (simulate corruption) // Note: This might fail if ECStore is actively using the file match fs::remove_dir_all(&disk1_object_path) { Ok(_) => { println!("Successfully deleted object from disk1: {disk1_object_path:?}"); // Verify deletion by checking if the directory still exists if disk1_object_path.exists() { println!("WARNING: Directory still exists after deletion: {disk1_object_path:?}"); } else { println!("Confirmed: Directory was successfully deleted"); } } Err(e) => { println!("Could not delete object from disk1 (file may be in use): {disk1_object_path:?} - {e}"); // This is expected behavior - ECStore might be holding file handles } } // Scan again - should still complete (even with missing data) let scan_result_after_corruption = scanner.scan_cycle().await; println!("Scan after corruption result: {scan_result_after_corruption:?}"); // Scanner should handle missing data gracefully assert!(scan_result_after_corruption.is_ok(), "Scanner should handle missing data gracefully"); // Test 3: Verify EC decode capability println!("=== Test 3: Verify EC decode ==="); // Note: EC decode verification is done internally during scan_cycle // We can verify that the scanner handles missing data gracefully println!("EC decode verification is handled internally during scan cycles"); // Test 4: Test metrics collection println!("=== Test 4: Metrics collection ==="); let final_metrics = scanner.get_metrics().await; println!("Final metrics: {final_metrics:?}"); // Verify metrics are reasonable assert!(final_metrics.total_cycles > 0, "Should have completed scan cycles"); assert!(final_metrics.last_activity.is_some(), "Should have scan activity"); // clean up temp dir let temp_dir = std::path::PathBuf::from(TEST_DIR_BASIC); if let Err(e) = fs::remove_dir_all(&temp_dir) { eprintln!("Warning: Failed to clean up temp directory {temp_dir:?}: {e}"); } } // test data usage statistics collection and validation #[tokio::test(flavor = "multi_thread")] #[serial] async fn test_scanner_usage_stats() { const TEST_DIR_USAGE_STATS: &str = "/tmp/rustfs_ahm_test_usage_stats"; let (_, ecstore) = prepare_test_env(Some(TEST_DIR_USAGE_STATS), Some(9002)).await; // prepare test bucket and object let bucket = "test-bucket"; ecstore.make_bucket(bucket, &Default::default()).await.unwrap(); let mut pr = PutObjReader::from_vec(b"hello".to_vec()); ecstore .put_object(bucket, "obj1", &mut pr, &Default::default()) .await .unwrap(); let scanner = Scanner::new(None, None); // enable statistics { let mut cfg = scanner.config.write().await; cfg.enable_data_usage_stats = true; } // first scan and get statistics scanner.scan_cycle().await.unwrap(); let du_initial = scanner.get_data_usage_info().await.unwrap(); assert!(du_initial.objects_total_count > 0); // write 3 more objects and get statistics again for size in [1024, 2048, 4096] { let name = format!("obj_{size}"); let mut pr = PutObjReader::from_vec(vec![b'x'; size]); ecstore.put_object(bucket, &name, &mut pr, &Default::default()).await.unwrap(); } scanner.scan_cycle().await.unwrap(); let du_after = scanner.get_data_usage_info().await.unwrap(); assert!(du_after.objects_total_count >= du_initial.objects_total_count + 3); // verify correctness of persisted data tokio::time::sleep(std::time::Duration::from_millis(100)).await; let persisted = load_data_usage_from_backend(ecstore.clone()) .await .expect("load persisted usage"); assert_eq!(persisted.objects_total_count, du_after.objects_total_count); assert_eq!(persisted.buckets_count, du_after.buckets_count); let p_bucket = persisted.buckets_usage.get(bucket).unwrap(); let m_bucket = du_after.buckets_usage.get(bucket).unwrap(); assert_eq!(p_bucket.objects_count, m_bucket.objects_count); assert_eq!(p_bucket.size, m_bucket.size); // consistency - again scan should not change count scanner.scan_cycle().await.unwrap(); let du_cons = scanner.get_data_usage_info().await.unwrap(); assert_eq!(du_after.objects_total_count, du_cons.objects_total_count); // 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)); } #[tokio::test(flavor = "multi_thread")] #[serial] async fn test_scanner_detect_missing_data_parts() { const TEST_DIR_MISSING_PARTS: &str = "/tmp/rustfs_ahm_test_missing_parts"; let (disk_paths, ecstore) = prepare_test_env(Some(TEST_DIR_MISSING_PARTS), Some(9004)).await; // Create test bucket let bucket_name = "test-bucket-parts"; let object_name = "large-object-20mb"; ecstore.make_bucket(bucket_name, &Default::default()).await.unwrap(); // Create a 20MB object to ensure it has multiple parts (MIN_PART_SIZE is 16MB) let large_data = vec![b'A'; 20 * 1024 * 1024]; // 20MB of 'A' characters let mut put_reader = PutObjReader::from_vec(large_data); let object_opts = rustfs_ecstore::store_api::ObjectOptions::default(); println!("=== Creating 20MB object ==="); ecstore .put_object(bucket_name, object_name, &mut put_reader, &object_opts) .await .expect("put_object failed for large object"); // Verify object was created and get its info let obj_info = ecstore .get_object_info(bucket_name, object_name, &object_opts) .await .expect("get_object_info failed"); println!( "Object info: size={}, parts={}, inlined={}", obj_info.size, obj_info.parts.len(), obj_info.inlined ); assert!(!obj_info.inlined, "20MB object should not be inlined"); // Note: Even 20MB might be stored as single part depending on configuration println!("Object has {} parts", obj_info.parts.len()); // Create HealManager and Scanner with shorter heal interval for testing let heal_storage = Arc::new(crate::heal::storage::ECStoreHealStorage::new(ecstore.clone())); let heal_config = HealConfig { enable_auto_heal: true, heal_interval: Duration::from_millis(100), // 100ms for faster testing max_concurrent_heals: 4, task_timeout: Duration::from_secs(300), queue_size: 1000, }; let heal_manager = Arc::new(crate::heal::HealManager::new(heal_storage, Some(heal_config))); heal_manager.start().await.unwrap(); let scanner = Scanner::new(None, Some(heal_manager.clone())); // Enable healing to detect missing parts { let mut config = scanner.config.write().await; config.enable_healing = true; config.scan_mode = ScanMode::Deep; } println!("=== Initial scan (all parts present) ==="); let initial_scan = scanner.scan_cycle().await; assert!(initial_scan.is_ok(), "Initial scan should succeed"); let initial_metrics = scanner.get_metrics().await; println!("Initial scan metrics: objects_scanned={}", initial_metrics.objects_scanned); // Simulate data part loss by deleting part files from some disks println!("=== Simulating data part loss ==="); let mut deleted_parts = 0; let mut deleted_part_paths = Vec::new(); // Track deleted file paths for later verification for (disk_idx, disk_path) in disk_paths.iter().enumerate() { if disk_idx > 0 { // Only delete from first two disks break; } let bucket_path = disk_path.join(bucket_name); let object_path = bucket_path.join(object_name); if !object_path.exists() { continue; } // Find the data directory (UUID) if let Ok(entries) = fs::read_dir(&object_path) { for entry in entries.flatten() { let entry_path = entry.path(); if entry_path.is_dir() { // This is likely the data_dir, look for part files inside let part_file_path = entry_path.join("part.1"); if part_file_path.exists() { match fs::remove_file(&part_file_path) { Ok(_) => { println!("Deleted part file: {:?}", part_file_path); deleted_part_paths.push(part_file_path); // Store path for verification deleted_parts += 1; } Err(e) => { println!("Failed to delete part file {:?}: {}", part_file_path, e); } } } } } } } println!("Deleted {} part files to simulate data loss", deleted_parts); assert!(deleted_parts > 0, "Should have deleted some part files"); // Scan again to detect missing parts println!("=== Scan after data deletion (should detect missing data) ==="); let scan_after_deletion = scanner.scan_cycle().await; // Wait a bit for the heal manager to process the queue tokio::time::sleep(Duration::from_millis(200)).await; // Add debug information println!("=== Debug: Checking heal manager state ==="); let tasks_count = heal_manager.get_active_tasks_count().await; println!("Active heal tasks count: {}", tasks_count); // Check heal statistics to see if any tasks were submitted let heal_stats = heal_manager.get_statistics().await; println!("Heal statistics:"); println!(" - total_tasks: {}", heal_stats.total_tasks); println!(" - successful_tasks: {}", heal_stats.successful_tasks); println!(" - failed_tasks: {}", heal_stats.failed_tasks); println!(" - running_tasks: {}", heal_stats.running_tasks); // Get scanner metrics to see what was scanned let final_metrics = scanner.get_metrics().await; println!("Scanner metrics after deletion scan:"); println!(" - objects_scanned: {}", final_metrics.objects_scanned); println!(" - healthy_objects: {}", final_metrics.healthy_objects); println!(" - corrupted_objects: {}", final_metrics.corrupted_objects); println!(" - objects_with_issues: {}", final_metrics.objects_with_issues); // Try to manually verify the object to see what happens println!("=== Manual object verification ==="); if let Some(ecstore) = rustfs_ecstore::new_object_layer_fn() { match ecstore.verify_object_integrity(bucket_name, object_name, &object_opts).await { Ok(_) => println!("Manual verification: Object is healthy"), Err(e) => println!("Manual verification: Object verification failed: {}", e), } } // Check if a heal task was submitted (check total tasks instead of active tasks) assert!(heal_stats.total_tasks > 0, "Heal task should have been submitted"); println!("{} heal tasks submitted in total", heal_stats.total_tasks); // Scanner should handle missing parts gracefully but may detect errors match scan_after_deletion { Ok(_) => { println!("Scanner completed successfully despite missing data"); } Err(e) => { println!("Scanner detected errors (expected): {}", e); // This is acceptable - scanner may report errors when data is missing } } let final_metrics = scanner.get_metrics().await; println!("Final scan metrics: objects_scanned={}", final_metrics.objects_scanned); // Verify that scanner completed additional cycles assert!( final_metrics.total_cycles > initial_metrics.total_cycles, "Should have completed additional scan cycles" ); // Test object retrieval after data loss println!("=== Testing object retrieval after data loss ==="); let get_result = ecstore.get_object_info(bucket_name, object_name, &object_opts).await; match get_result { Ok(info) => { println!("Object still accessible: size={}", info.size); // EC should allow recovery if enough shards remain } Err(e) => { println!("Object not accessible due to missing data: {}", e); // This is expected if too many shards are missing } } println!("=== Test completed ==="); println!("Scanner successfully handled missing data scenario"); // Verify that deleted part files have been restored by the healing process println!("=== Verifying file recovery ==="); let mut recovered_files = 0; for deleted_path in &deleted_part_paths { assert!(deleted_path.exists(), "Deleted file should have been recovered"); println!("Recovered file: {:?}", deleted_path); recovered_files += 1; } // Assert that at least some files have been recovered // Note: In a real scenario, healing might take longer, but our test setup should allow recovery if heal_stats.successful_tasks > 0 { assert!( recovered_files == deleted_part_paths.len(), "Expected at least some deleted files to be recovered by healing process. \ Deleted {} files, recovered {} files, successful heal tasks: {}", deleted_part_paths.len(), recovered_files, heal_stats.successful_tasks ); println!("Successfully recovered {}/{} deleted files", recovered_files, deleted_part_paths.len()); } else { println!("No successful heal tasks completed yet - healing may still be in progress"); } // Clean up let _ = std::fs::remove_dir_all(std::path::Path::new(TEST_DIR_MISSING_PARTS)); } #[tokio::test(flavor = "multi_thread")] #[serial] async fn test_scanner_detect_missing_xl_meta() { const TEST_DIR_MISSING_META: &str = "/tmp/rustfs_ahm_test_missing_meta"; let (disk_paths, ecstore) = prepare_test_env(Some(TEST_DIR_MISSING_META), Some(9005)).await; // Create test bucket let bucket_name = "test-bucket-meta"; let object_name = "test-object-meta"; ecstore.make_bucket(bucket_name, &Default::default()).await.unwrap(); // Create a test object let test_data = vec![b'B'; 5 * 1024 * 1024]; // 5MB of 'B' characters let mut put_reader = PutObjReader::from_vec(test_data); let object_opts = rustfs_ecstore::store_api::ObjectOptions::default(); println!("=== Creating test object ==="); ecstore .put_object(bucket_name, object_name, &mut put_reader, &object_opts) .await .expect("put_object failed"); // Verify object was created and get its info let obj_info = ecstore .get_object_info(bucket_name, object_name, &object_opts) .await .expect("get_object_info failed"); println!("Object info: size={}, parts={}", obj_info.size, obj_info.parts.len()); // Create HealManager and Scanner with shorter heal interval for testing let heal_storage = Arc::new(crate::heal::storage::ECStoreHealStorage::new(ecstore.clone())); let heal_config = HealConfig { enable_auto_heal: true, heal_interval: Duration::from_millis(100), // 100ms for faster testing max_concurrent_heals: 4, task_timeout: Duration::from_secs(300), queue_size: 1000, }; let heal_manager = Arc::new(crate::heal::HealManager::new(heal_storage, Some(heal_config))); heal_manager.start().await.unwrap(); let scanner = Scanner::new(None, Some(heal_manager.clone())); // Enable healing to detect missing metadata { let mut config = scanner.config.write().await; config.enable_healing = true; config.scan_mode = ScanMode::Deep; } println!("=== Initial scan (all metadata present) ==="); let initial_scan = scanner.scan_cycle().await; assert!(initial_scan.is_ok(), "Initial scan should succeed"); let initial_metrics = scanner.get_metrics().await; println!("Initial scan metrics: objects_scanned={}", initial_metrics.objects_scanned); // Simulate xl.meta file loss by deleting xl.meta files from some disks println!("=== Simulating xl.meta file loss ==="); let mut deleted_meta_files = 0; let mut deleted_meta_paths = Vec::new(); // Track deleted file paths for later verification for (disk_idx, disk_path) in disk_paths.iter().enumerate() { if disk_idx >= 2 { // Only delete from first two disks to ensure some copies remain for recovery break; } let bucket_path = disk_path.join(bucket_name); let object_path = bucket_path.join(object_name); if !object_path.exists() { continue; } // Delete xl.meta file let xl_meta_path = object_path.join("xl.meta"); if xl_meta_path.exists() { match fs::remove_file(&xl_meta_path) { Ok(_) => { println!("Deleted xl.meta file: {:?}", xl_meta_path); deleted_meta_paths.push(xl_meta_path); deleted_meta_files += 1; } Err(e) => { println!("Failed to delete xl.meta file {:?}: {}", xl_meta_path, e); } } } } println!("Deleted {} xl.meta files to simulate metadata loss", deleted_meta_files); assert!(deleted_meta_files > 0, "Should have deleted some xl.meta files"); // Scan again to detect missing metadata println!("=== Scan after xl.meta deletion (should detect missing metadata) ==="); let scan_after_deletion = scanner.scan_cycle().await; // Wait a bit for the heal manager to process the queue tokio::time::sleep(Duration::from_millis(500)).await; // Add debug information println!("=== Debug: Checking heal manager state ==="); let tasks_count = heal_manager.get_active_tasks_count().await; println!("Active heal tasks count: {}", tasks_count); // Check heal statistics to see if any tasks were submitted let heal_stats = heal_manager.get_statistics().await; println!("Heal statistics:"); println!(" - total_tasks: {}", heal_stats.total_tasks); println!(" - successful_tasks: {}", heal_stats.successful_tasks); println!(" - failed_tasks: {}", heal_stats.failed_tasks); println!(" - running_tasks: {}", heal_stats.running_tasks); // Get scanner metrics to see what was scanned let final_metrics = scanner.get_metrics().await; println!("Scanner metrics after deletion scan:"); println!(" - objects_scanned: {}", final_metrics.objects_scanned); println!(" - healthy_objects: {}", final_metrics.healthy_objects); println!(" - corrupted_objects: {}", final_metrics.corrupted_objects); println!(" - objects_with_issues: {}", final_metrics.objects_with_issues); // Try to manually verify the object to see what happens println!("=== Manual object verification ==="); if let Some(ecstore) = rustfs_ecstore::new_object_layer_fn() { match ecstore.verify_object_integrity(bucket_name, object_name, &object_opts).await { Ok(_) => println!("Manual verification: Object is healthy"), Err(e) => println!("Manual verification: Object verification failed: {}", e), } } // Check if a heal task was submitted for metadata recovery assert!(heal_stats.total_tasks > 0, "Heal task should have been submitted for missing xl.meta"); println!("{} heal tasks submitted in total", heal_stats.total_tasks); // Scanner should handle missing metadata gracefully but may detect errors match scan_after_deletion { Ok(_) => { println!("Scanner completed successfully despite missing metadata"); } Err(e) => { println!("Scanner detected errors (expected): {}", e); // This is acceptable - scanner may report errors when metadata is missing } } let final_metrics = scanner.get_metrics().await; println!("Final scan metrics: objects_scanned={}", final_metrics.objects_scanned); // Verify that scanner completed additional cycles assert!( final_metrics.total_cycles > initial_metrics.total_cycles, "Should have completed additional scan cycles" ); // Test object retrieval after metadata loss println!("=== Testing object retrieval after metadata loss ==="); let get_result = ecstore.get_object_info(bucket_name, object_name, &object_opts).await; match get_result { Ok(info) => { println!("Object still accessible: size={}", info.size); // Object should still be accessible if enough metadata copies remain } Err(e) => { println!("Object not accessible due to missing metadata: {}", e); // This might happen if too many metadata files are missing } } // Wait a bit more for healing to complete tokio::time::sleep(Duration::from_millis(1000)).await; // Check heal statistics again after waiting let final_heal_stats = heal_manager.get_statistics().await; println!("Final heal statistics:"); println!(" - total_tasks: {}", final_heal_stats.total_tasks); println!(" - successful_tasks: {}", final_heal_stats.successful_tasks); println!(" - failed_tasks: {}", final_heal_stats.failed_tasks); // Verify that deleted xl.meta files have been restored by the healing process println!("=== Verifying xl.meta file recovery ==="); let mut recovered_files = 0; for deleted_path in &deleted_meta_paths { if deleted_path.exists() { println!("Recovered xl.meta file: {:?}", deleted_path); recovered_files += 1; } else { println!("xl.meta file still missing: {:?}", deleted_path); } } // Assert that healing was attempted assert!( final_heal_stats.total_tasks > 0, "Heal tasks should have been submitted for missing xl.meta files" ); // Check if any heal tasks were successful if final_heal_stats.successful_tasks > 0 { println!("Healing completed successfully, checking file recovery..."); if recovered_files > 0 { println!("Successfully recovered {}/{} deleted xl.meta files", recovered_files, deleted_meta_paths.len()); } else { println!("No xl.meta files recovered yet - healing may have recreated metadata elsewhere"); } } else { println!("No successful heal tasks completed yet - healing may still be in progress or failed"); // If healing failed, this is acceptable for this test scenario // The important thing is that the scanner detected the issue and submitted heal tasks if final_heal_stats.failed_tasks > 0 { println!("Heal tasks failed - this is acceptable for missing xl.meta scenario"); println!("The scanner correctly detected missing metadata and submitted heal requests"); } } // The key success criteria for this test is that: // 1. Scanner detected missing xl.meta files // 2. Scanner submitted heal tasks for the missing metadata // 3. Scanner handled the situation gracefully without crashing println!("=== Test completed ==="); println!("Scanner successfully handled missing xl.meta scenario"); println!("Key achievements:"); println!(" - Scanner detected missing xl.meta files"); println!(" - Scanner submitted {} heal tasks", final_heal_stats.total_tasks); println!(" - Scanner handled the situation gracefully"); if recovered_files > 0 { println!(" - Successfully recovered {}/{} xl.meta files", recovered_files, deleted_meta_paths.len()); } else { println!(" - Note: xl.meta file recovery may require additional time or manual intervention"); } // Clean up let _ = std::fs::remove_dir_all(std::path::Path::new(TEST_DIR_MISSING_META)); } }