From 2aa7a631efa192b574a392d95b7ab0fb40d5c1ff Mon Sep 17 00:00:00 2001 From: dandan Date: Thu, 10 Jul 2025 15:44:37 +0800 Subject: [PATCH] feat: refactor scanner module and add data usage statistics - Move scanner code to scanner/ subdirectory for better organization - Add data usage statistics collection and persistence - Implement histogram support for size and version distribution - Add global cancel token management for scanner operations - Integrate scanner with ECStore for comprehensive data analysis - Update error handling and improve test isolation - Add data usage API endpoints and backend integration Signed-off-by: junxiang Mu <1948535941@qq.com> --- crates/ahm/src/lib.rs | 1 - crates/ahm/src/metrics.rs | 284 ------------ crates/ahm/src/scanner.rs | 902 -------------------------------------- 3 files changed, 1187 deletions(-) delete mode 100644 crates/ahm/src/metrics.rs delete mode 100644 crates/ahm/src/scanner.rs diff --git a/crates/ahm/src/lib.rs b/crates/ahm/src/lib.rs index d3f6e151b..d3d656194 100644 --- a/crates/ahm/src/lib.rs +++ b/crates/ahm/src/lib.rs @@ -17,7 +17,6 @@ use tokio_util::sync::CancellationToken; pub mod error; pub mod scanner; -pub mod metrics; pub use error::{Error, Result}; pub use scanner::{ diff --git a/crates/ahm/src/metrics.rs b/crates/ahm/src/metrics.rs deleted file mode 100644 index b541b2fc3..000000000 --- a/crates/ahm/src/metrics.rs +++ /dev/null @@ -1,284 +0,0 @@ -// 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::atomic::{AtomicU64, Ordering}, - time::{Duration, SystemTime}, -}; - -use serde::{Deserialize, Serialize}; -use tracing::info; - -/// Scanner metrics similar to MinIO's scanner metrics -#[derive(Debug, Clone, Default, Serialize, Deserialize)] -pub struct ScannerMetrics { - /// Total objects scanned since server start - pub objects_scanned: u64, - /// Total object versions scanned since server start - pub versions_scanned: u64, - /// Total directories scanned since server start - pub directories_scanned: u64, - /// Total bucket scans started since server start - pub bucket_scans_started: u64, - /// Total bucket scans finished since server start - pub bucket_scans_finished: u64, - /// Total objects with health issues found - pub objects_with_issues: u64, - /// Total heal tasks queued - pub heal_tasks_queued: u64, - /// Total heal tasks completed - pub heal_tasks_completed: u64, - /// Total heal tasks failed - pub heal_tasks_failed: u64, - /// Last scan activity time - pub last_activity: Option, - /// Current scan cycle - pub current_cycle: u64, - /// Total scan cycles completed - pub total_cycles: u64, - /// Current scan duration - pub current_scan_duration: Option, - /// Average scan duration - pub avg_scan_duration: Duration, - /// Objects scanned per second - pub objects_per_second: f64, - /// Buckets scanned per second - pub buckets_per_second: f64, - /// Storage metrics by bucket - pub bucket_metrics: HashMap, - /// Disk metrics - pub disk_metrics: HashMap, -} - -/// Bucket-specific metrics -#[derive(Debug, Clone, Default, Serialize, Deserialize)] -pub struct BucketMetrics { - /// Bucket name - pub bucket: String, - /// Total objects in bucket - pub total_objects: u64, - /// Total size of objects in bucket (bytes) - pub total_size: u64, - /// Objects with health issues - pub objects_with_issues: u64, - /// Last scan time - pub last_scan_time: Option, - /// Scan duration - pub scan_duration: Option, - /// Heal tasks queued for this bucket - pub heal_tasks_queued: u64, - /// Heal tasks completed for this bucket - pub heal_tasks_completed: u64, - /// Heal tasks failed for this bucket - pub heal_tasks_failed: u64, -} - -/// Disk-specific metrics -#[derive(Debug, Clone, Default, Serialize, Deserialize)] -pub struct DiskMetrics { - /// Disk path - pub disk_path: String, - /// Total disk space (bytes) - pub total_space: u64, - /// Used disk space (bytes) - pub used_space: u64, - /// Free disk space (bytes) - pub free_space: u64, - /// Objects scanned on this disk - pub objects_scanned: u64, - /// Objects with issues on this disk - pub objects_with_issues: u64, - /// Last scan time - pub last_scan_time: Option, - /// Whether disk is online - pub is_online: bool, - /// Whether disk is being scanned - pub is_scanning: bool, -} - -/// Thread-safe metrics collector -pub struct MetricsCollector { - /// Atomic counters for real-time metrics - objects_scanned: AtomicU64, - versions_scanned: AtomicU64, - directories_scanned: AtomicU64, - bucket_scans_started: AtomicU64, - bucket_scans_finished: AtomicU64, - objects_with_issues: AtomicU64, - heal_tasks_queued: AtomicU64, - heal_tasks_completed: AtomicU64, - heal_tasks_failed: AtomicU64, - current_cycle: AtomicU64, - total_cycles: AtomicU64, -} - -impl MetricsCollector { - /// Create a new metrics collector - pub fn new() -> Self { - Self { - objects_scanned: AtomicU64::new(0), - versions_scanned: AtomicU64::new(0), - directories_scanned: AtomicU64::new(0), - bucket_scans_started: AtomicU64::new(0), - bucket_scans_finished: AtomicU64::new(0), - objects_with_issues: AtomicU64::new(0), - heal_tasks_queued: AtomicU64::new(0), - heal_tasks_completed: AtomicU64::new(0), - heal_tasks_failed: AtomicU64::new(0), - current_cycle: AtomicU64::new(0), - total_cycles: AtomicU64::new(0), - } - } - - /// Increment objects scanned count - pub fn increment_objects_scanned(&self, count: u64) { - self.objects_scanned.fetch_add(count, Ordering::Relaxed); - } - - /// Increment versions scanned count - pub fn increment_versions_scanned(&self, count: u64) { - self.versions_scanned.fetch_add(count, Ordering::Relaxed); - } - - /// Increment directories scanned count - pub fn increment_directories_scanned(&self, count: u64) { - self.directories_scanned.fetch_add(count, Ordering::Relaxed); - } - - /// Increment bucket scans started count - pub fn increment_bucket_scans_started(&self, count: u64) { - self.bucket_scans_started.fetch_add(count, Ordering::Relaxed); - } - - /// Increment bucket scans finished count - pub fn increment_bucket_scans_finished(&self, count: u64) { - self.bucket_scans_finished.fetch_add(count, Ordering::Relaxed); - } - - /// Increment objects with issues count - pub fn increment_objects_with_issues(&self, count: u64) { - self.objects_with_issues.fetch_add(count, Ordering::Relaxed); - } - - /// Increment heal tasks queued count - pub fn increment_heal_tasks_queued(&self, count: u64) { - self.heal_tasks_queued.fetch_add(count, Ordering::Relaxed); - } - - /// Increment heal tasks completed count - pub fn increment_heal_tasks_completed(&self, count: u64) { - self.heal_tasks_completed.fetch_add(count, Ordering::Relaxed); - } - - /// Increment heal tasks failed count - pub fn increment_heal_tasks_failed(&self, count: u64) { - self.heal_tasks_failed.fetch_add(count, Ordering::Relaxed); - } - - /// Set current cycle - pub fn set_current_cycle(&self, cycle: u64) { - self.current_cycle.store(cycle, Ordering::Relaxed); - } - - /// Increment total cycles - pub fn increment_total_cycles(&self) { - self.total_cycles.fetch_add(1, Ordering::Relaxed); - } - - /// Get current metrics snapshot - pub fn get_metrics(&self) -> ScannerMetrics { - ScannerMetrics { - objects_scanned: self.objects_scanned.load(Ordering::Relaxed), - versions_scanned: self.versions_scanned.load(Ordering::Relaxed), - directories_scanned: self.directories_scanned.load(Ordering::Relaxed), - bucket_scans_started: self.bucket_scans_started.load(Ordering::Relaxed), - bucket_scans_finished: self.bucket_scans_finished.load(Ordering::Relaxed), - objects_with_issues: self.objects_with_issues.load(Ordering::Relaxed), - heal_tasks_queued: self.heal_tasks_queued.load(Ordering::Relaxed), - heal_tasks_completed: self.heal_tasks_completed.load(Ordering::Relaxed), - heal_tasks_failed: self.heal_tasks_failed.load(Ordering::Relaxed), - last_activity: Some(SystemTime::now()), - current_cycle: self.current_cycle.load(Ordering::Relaxed), - total_cycles: self.total_cycles.load(Ordering::Relaxed), - current_scan_duration: None, // Will be set by scanner - avg_scan_duration: Duration::ZERO, // Will be calculated - objects_per_second: 0.0, // Will be calculated - buckets_per_second: 0.0, // Will be calculated - bucket_metrics: HashMap::new(), // Will be populated by scanner - disk_metrics: HashMap::new(), // Will be populated by scanner - } - } - - /// Reset all metrics - pub fn reset(&self) { - self.objects_scanned.store(0, Ordering::Relaxed); - self.versions_scanned.store(0, Ordering::Relaxed); - self.directories_scanned.store(0, Ordering::Relaxed); - self.bucket_scans_started.store(0, Ordering::Relaxed); - self.bucket_scans_finished.store(0, Ordering::Relaxed); - self.objects_with_issues.store(0, Ordering::Relaxed); - self.heal_tasks_queued.store(0, Ordering::Relaxed); - self.heal_tasks_completed.store(0, Ordering::Relaxed); - self.heal_tasks_failed.store(0, Ordering::Relaxed); - self.current_cycle.store(0, Ordering::Relaxed); - self.total_cycles.store(0, Ordering::Relaxed); - - info!("Scanner metrics reset"); - } -} - -impl Default for MetricsCollector { - fn default() -> Self { - Self::new() - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn test_metrics_collector_creation() { - let collector = MetricsCollector::new(); - let metrics = collector.get_metrics(); - assert_eq!(metrics.objects_scanned, 0); - assert_eq!(metrics.versions_scanned, 0); - } - - #[test] - fn test_metrics_increment() { - let collector = MetricsCollector::new(); - - collector.increment_objects_scanned(10); - collector.increment_versions_scanned(5); - collector.increment_objects_with_issues(2); - - let metrics = collector.get_metrics(); - assert_eq!(metrics.objects_scanned, 10); - assert_eq!(metrics.versions_scanned, 5); - assert_eq!(metrics.objects_with_issues, 2); - } - - #[test] - fn test_metrics_reset() { - let collector = MetricsCollector::new(); - - collector.increment_objects_scanned(10); - collector.reset(); - - let metrics = collector.get_metrics(); - assert_eq!(metrics.objects_scanned, 0); - } -} \ No newline at end of file diff --git a/crates/ahm/src/scanner.rs b/crates/ahm/src/scanner.rs deleted file mode 100644 index 52f92bba3..000000000 --- a/crates/ahm/src/scanner.rs +++ /dev/null @@ -1,902 +0,0 @@ -// 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 rustfs_ecstore as ecstore; -use ecstore::{ - disk::{DiskAPI, DiskStore, WalkDirOptions}, - set_disk::SetDisks, -}; -use rustfs_filemeta::MetacacheReader; -use tokio::sync::RwLock; -use tokio_util::sync::CancellationToken; -use tracing::{debug, error, info, warn}; - -use crate::{ - error::{Error, Result}, - metrics::{BucketMetrics, DiskMetrics, MetricsCollector, ScannerMetrics}, -}; - -/// Custom scan mode enum for AHM scanner -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum ScanMode { - /// Normal scan - basic object discovery and metadata collection - Normal, - /// Deep scan - includes EC verification and integrity checks - Deep, -} - -impl Default for ScanMode { - fn default() -> Self { - ScanMode::Normal - } -} - -/// 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, -} - -impl Default for ScannerConfig { - fn default() -> Self { - Self { - scan_interval: Duration::from_secs(3600), // 1 hour - deep_scan_interval: Duration::from_secs(3600), // 1 hour - max_concurrent_scans: 20, - enable_healing: true, - enable_metrics: true, - scan_mode: ScanMode::Normal, - } - } -} - -/// Scanner state -#[derive(Debug)] -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, -} - -impl Default for ScannerState { - fn default() -> Self { - Self { - is_running: false, - current_cycle: 0, - last_scan_start: None, - last_scan_end: None, - current_scan_duration: None, - last_deep_scan_time: None, - scanning_buckets: Vec::new(), - scanning_disks: Vec::new(), - } - } -} - -/// AHM Scanner - Automatic Health Management Scanner -/// -/// This scanner monitors the health of objects in the RustFS storage system. -/// It integrates with ECStore's SetDisks to perform real data scanning and -/// collects metrics similar to MinIO's scanner. -/// -/// The scanner operates on 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>, - /// Metrics collector - metrics: Arc, - /// Bucket metrics cache - bucket_metrics: Arc>>, - /// Disk metrics cache - disk_metrics: Arc>>, - /// EC Set disks - represents a complete erasure coding set - set_disks: Arc, - /// Cancellation token for graceful shutdown - cancel_token: CancellationToken, -} - -impl Scanner { - /// Create a new scanner - pub fn new(set_disks: Arc, config: Option) -> Self { - let config = config.unwrap_or_default(); - let cancel_token = CancellationToken::new(); - - info!("Creating AHM scanner for EC set with {} disks", set_disks.set_drive_count); - - Self { - config: Arc::new(RwLock::new(config)), - state: Arc::new(RwLock::new(ScannerState::default())), - metrics: Arc::new(MetricsCollector::new()), - bucket_metrics: Arc::new(RwLock::new(HashMap::new())), - disk_metrics: Arc::new(RwLock::new(HashMap::new())), - set_disks, - cancel_token, - } - } - - /// 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 - self.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 a clone of the cancellation token (for external graceful shutdown) - pub fn cancellation_token(&self) -> CancellationToken { - self.cancel_token.clone() - } - - /// Get current scanner metrics - pub async fn get_metrics(&self) -> ScannerMetrics { - let mut metrics = self.metrics.get_metrics(); - - // Add bucket metrics - let bucket_metrics = self.bucket_metrics.read().await; - metrics.bucket_metrics = bucket_metrics.clone(); - - // Add disk metrics - let disk_metrics = self.disk_metrics.read().await; - metrics.disk_metrics = disk_metrics.clone(); - - // Add current scan duration - let state = self.state.read().await; - metrics.current_scan_duration = state.current_scan_duration; - - metrics - } - - /// Perform a single scan cycle - pub async fn scan_cycle(&self) -> Result<()> { - let start_time = SystemTime::now(); - - info!("Starting scan cycle {}", 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(); - } - - self.metrics.set_current_cycle(self.state.read().await.current_cycle); - self.metrics.increment_total_cycles(); - - // Get online disks from the EC set - let (disks, _) = self.set_disks.get_online_disks_with_healing(false).await; - - if disks.is_empty() { - warn!("No online disks available for scanning"); - return Ok(()); - } - - info!("Scanning {} online disks", disks.len()); - - // Phase 1: Scan all disks concurrently to collect object metadata - 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.clone() { - 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: {}", e); - // Add empty map for failed disk - all_disk_objects.push(HashMap::new()); - } - } - - } - - // Phase 2: Analyze object distribution and perform EC verification - if successful_scans > 0 { - if let Err(e) = self.analyze_object_distribution(&all_disk_objects, &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); - } - - info!("Completed scan cycle in {:?} ({} successful, {} failed)", - scan_duration, successful_scans, failed_scans); - Ok(()) - } - - /// Scan a single disk - async fn scan_disk(&self, disk: &DiskStore) -> Result>> { - let disk_path = disk.path().to_string_lossy().to_string(); - - info!("Scanning disk: {}", disk_path); - - // Update disk metrics - { - let mut disk_metrics = self.disk_metrics.write().await; - let metrics = disk_metrics.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 - 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; - } - } - - // 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); - return Err(Error::Storage(e.into())); - } - }; - - // Scan each volume and collect object metadata - let mut disk_objects = HashMap::new(); - for volume in volumes { - // 检查取消信号 - if self.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 = self.disk_metrics.write().await; - if let Some(metrics) = disk_metrics.get_mut(&disk_path) { - metrics.is_scanning = false; - } - } - - // Update state - { - let mut state = self.state.write().await; - state.scanning_disks.retain(|d| d != &disk_path); - } - - 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> { - info!("Scanning bucket: {} on disk: {}", bucket, disk.to_string()); - - // Update bucket metrics - { - let mut bucket_metrics = self.bucket_metrics.write().await; - let metrics = bucket_metrics.entry(bucket.to_string()).or_insert_with(|| BucketMetrics { - bucket: bucket.to_string(), - ..Default::default() - }); - - metrics.last_scan_time = Some(SystemTime::now()); - } - - // 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); - } else { - // Store object metadata for later analysis - object_metadata.insert(entry.name.clone(), file_meta); - } - } else { - objects_with_issues += 1; - warn!("Failed to parse metadata for object {}", entry.name); - } - } - } - - // 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 = self.bucket_metrics.write().await; - if let Some(metrics) = bucket_metrics.get_mut(bucket) { - metrics.total_objects = objects_scanned; - metrics.objects_with_issues = objects_with_issues; - metrics.scan_duration = Some( - SystemTime::now() - .duration_since(scan_start) - .unwrap_or(Duration::ZERO) - ); - } - } - - // Update state - { - let mut state = self.state.write().await; - state.scanning_buckets.retain(|b| b != bucket); - } - - 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 { - // 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 { - 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 { - 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); - println!("Object {}/{} missing from disks: {:?}", bucket, object_name, missing_disks); - // TODO: Trigger heal for this object - } - - // Step 3: Deep scan EC verification - let config = self.config.read().await; - if config.scan_mode == ScanMode::Deep { - // Find the first disk that has this object to get metadata - if let Some(&first_disk_idx) = locations.first() { - if let Some(file_meta) = all_disk_objects[first_disk_idx] - .get(bucket) - .and_then(|objects| objects.get(object_name)) - { - if let Err(e) = self.verify_ec_decode_with_locations( - bucket, object_name, file_meta, locations, disks - ).await { - objects_with_ec_issues += 1; - warn!("EC decode 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); - - Ok(()) - } - - /// Verify EC decode capability for an object using known disk locations - /// - /// This method is optimized to use the known locations of object copies - /// instead of scanning all disks. - async fn verify_ec_decode_with_locations( - &self, - bucket: &str, - object: &str, - file_meta: &rustfs_filemeta::FileMeta, - locations: &[usize], - all_disks: &[DiskStore], - ) -> Result<()> { - // Get EC parameters from the latest version - let (data_blocks, _parity_blocks) = if let Some(latest_version) = file_meta.versions.last() { - if let Ok(version) = rustfs_filemeta::FileMetaVersion::try_from(latest_version.clone()) { - if let Some(obj) = version.object { - (obj.erasure_m, obj.erasure_n) - } else { - // Not an object version, skip EC verification - return Ok(()); - } - } else { - // Cannot parse version, skip EC verification - return Ok(()); - } - } else { - // No versions, skip EC verification - return Ok(()); - }; - - let read_quorum = data_blocks; // Need at least data_blocks to decode - - if locations.len() < read_quorum { - return Err(Error::Scanner(format!( - "Insufficient object copies for EC decode: need {}, have {}", - read_quorum, locations.len() - ))); - } - - // Try to read object metadata from the known locations - let mut successful_reads = 0; - let mut errors = Vec::new(); - - for &disk_idx in locations { - if successful_reads >= read_quorum { - break; // We have enough copies for EC decode - } - - let disk = &all_disks[disk_idx]; - match disk.read_xl(bucket, object, false).await { - Ok(_) => { - successful_reads += 1; - debug!("Successfully read object {}/{} from disk {} (index: {})", - bucket, object, disk.to_string(), disk_idx); - } - Err(e) => { - let error_msg = format!("{}", e); - errors.push(error_msg); - debug!("Failed to read object {}/{} from disk {} (index: {}): {}", - bucket, object, disk.to_string(), disk_idx, e); - } - } - } - - if successful_reads >= read_quorum { - debug!("EC decode verification passed for object {}/{} ({} successful reads from {} locations)", - bucket, object, successful_reads, locations.len()); - Ok(()) - } else { - Err(Error::Scanner(format!( - "EC decode verification failed for object {}/{}: need {} reads, got {} (errors: {:?})", - bucket, object, read_quorum, successful_reads, errors - ))) - } - } - - /// 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); - let cancel_token = self.cancel_token.clone(); - - 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(()) - } - - /// 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), - set_disks: Arc::clone(&self.set_disks), - cancel_token: self.cancel_token.clone(), - } - } -} - -#[cfg(test)] -mod tests { - use super::*; - use rustfs_ecstore::store::ECStore; - use rustfs_ecstore::disk::endpoint::Endpoint; - use rustfs_ecstore::endpoints::{EndpointServerPools, PoolEndpoints, Endpoints}; - use rustfs_ecstore::{StorageAPI, store_api::{ObjectIO, MakeBucketOptions, PutObjReader}}; - use std::fs; - use std::net::SocketAddr; - - #[tokio::test(flavor = "multi_thread")] - async fn test_scanner_basic_functionality() { - // create temp dir as 4 disks - let temp_dir = std::path::PathBuf::from("/tmp/rustfs_ahm_test"); - 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 - let server_addr: SocketAddr = "127.0.0.1:9000".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; - - // get first SetDisks - let set_disks = ecstore.pools[0].get_disks(0); - - // 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(set_disks, 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 - // if let Err(e) = fs::remove_dir_all(&temp_dir) { - // eprintln!("Warning: Failed to clean up temp directory {:?}: {}", temp_dir, e); - // } - } -} -