fix: Refact heal and scanner design

Signed-off-by: junxiang Mu <1948535941@qq.com>

crates/ahm/src/scanner/bandwidth_limiter.rs

@@ -0,0 +1,353 @@
+// 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::{
+    sync::{
+        atomic::{AtomicU64, Ordering},
+        Arc,
+    },
+    time::{Duration, Instant},
+};
+
+use tokio::{
+    sync::RwLock,
+    time::{sleep, sleep_until, Instant as TokioInstant},
+};
+use tracing::{debug, info, warn};
+
+use crate::error::Result;
+
+/// Configuration for bandwidth limiting
+#[derive(Debug, Clone)]
+pub struct BandwidthConfig {
+    /// Maximum bytes per second
+    pub bytes_per_second: u64,
+    /// Maximum operations per second
+    pub operations_per_second: u64,
+    /// Burst allowance multiplier
+    pub burst_multiplier: f64,
+    /// Whether to enable adaptive throttling
+    pub adaptive_throttling: bool,
+    /// Minimum sleep duration between operations
+    pub min_sleep_duration: Duration,
+    /// Maximum sleep duration between operations
+    pub max_sleep_duration: Duration,
+}
+
+impl Default for BandwidthConfig {
+    fn default() -> Self {
+        Self {
+            bytes_per_second: 100 * 1024 * 1024, // 100 MB/s
+            operations_per_second: 1000, // 1000 ops/s
+            burst_multiplier: 2.0,
+            adaptive_throttling: true,
+            min_sleep_duration: Duration::from_micros(100),
+            max_sleep_duration: Duration::from_millis(100),
+        }
+    }
+}
+
+/// Bandwidth limiter for controlling scan I/O rates
+pub struct BandwidthLimiter {
+    config: BandwidthConfig,
+    bytes_this_second: Arc<AtomicU64>,
+    operations_this_second: Arc<AtomicU64>,
+    last_reset: Arc<RwLock<Instant>>,
+    adaptive_sleep_duration: Arc<RwLock<Duration>>,
+    total_bytes_processed: Arc<AtomicU64>,
+    total_operations_processed: Arc<AtomicU64>,
+    start_time: Instant,
+}
+
+impl BandwidthLimiter {
+    /// Create a new bandwidth limiter
+    pub fn new(config: BandwidthConfig) -> Self {
+        let adaptive_sleep = if config.adaptive_throttling {
+            config.min_sleep_duration
+        } else {
+            Duration::from_micros(1000) // 1ms default
+        };
+
+        Self {
+            config,
+            bytes_this_second: Arc::new(AtomicU64::new(0)),
+            operations_this_second: Arc::new(AtomicU64::new(0)),
+            last_reset: Arc::new(RwLock::new(Instant::now())),
+            adaptive_sleep_duration: Arc::new(RwLock::new(adaptive_sleep)),
+            total_bytes_processed: Arc::new(AtomicU64::new(0)),
+            total_operations_processed: Arc::new(AtomicU64::new(0)),
+            start_time: Instant::now(),
+        }
+    }
+
+    /// Wait for bandwidth allowance before processing bytes
+    pub async fn wait_for_bytes(&self, bytes: u64) -> Result<()> {
+        if self.config.bytes_per_second == 0 {
+            return Ok(());
+        }
+
+        let mut total_wait_time = Duration::ZERO;
+        let mut remaining_bytes = bytes;
+
+        while remaining_bytes > 0 {
+            // Reset counters if a second has passed
+            self.reset_counters_if_needed().await;
+
+            let current_bytes = self.bytes_this_second.load(Ordering::Relaxed);
+            let burst_limit = (self.config.bytes_per_second as f64 * self.config.burst_multiplier) as u64;
+
+            if current_bytes >= burst_limit {
+                // We're over the burst limit, wait
+                let wait_time = self.calculate_wait_time(current_bytes, self.config.bytes_per_second).await;
+                sleep(wait_time).await;
+                total_wait_time += wait_time;
+                continue;
+            }
+
+            let bytes_to_process = std::cmp::min(remaining_bytes, burst_limit - current_bytes);
+            self.bytes_this_second.fetch_add(bytes_to_process, Ordering::Relaxed);
+            self.total_bytes_processed.fetch_add(bytes_to_process, Ordering::Relaxed);
+            remaining_bytes -= bytes_to_process;
+
+            // Adaptive throttling
+            if self.config.adaptive_throttling {
+                self.update_adaptive_sleep(bytes_to_process).await;
+            }
+        }
+
+        if total_wait_time > Duration::ZERO {
+            debug!("Bandwidth limiter waited {:?} for {} bytes", total_wait_time, bytes);
+        }
+
+        Ok(())
+    }
+
+    /// Wait for bandwidth allowance before processing an operation
+    pub async fn wait_for_operation(&self) -> Result<()> {
+        if self.config.operations_per_second == 0 {
+            return Ok(());
+        }
+
+        // Reset counters if a second has passed
+        self.reset_counters_if_needed().await;
+
+        let current_ops = self.operations_this_second.load(Ordering::Relaxed);
+        let burst_limit = (self.config.operations_per_second as f64 * self.config.burst_multiplier) as u64;
+
+        if current_ops >= burst_limit {
+            // We're over the burst limit, wait
+            let wait_time = self.calculate_wait_time(current_ops, self.config.operations_per_second).await;
+            sleep(wait_time).await;
+            debug!("Bandwidth limiter waited {:?} for operation", wait_time);
+        }
+
+        self.operations_this_second.fetch_add(1, Ordering::Relaxed);
+        self.total_operations_processed.fetch_add(1, Ordering::Relaxed);
+
+        Ok(())
+    }
+
+    /// Wait for bandwidth allowance before processing both bytes and operations
+    pub async fn wait_for_bytes_and_operation(&self, bytes: u64) -> Result<()> {
+        self.wait_for_bytes(bytes).await?;
+        self.wait_for_operation().await?;
+        Ok(())
+    }
+
+    /// Reset counters if a second has passed
+    async fn reset_counters_if_needed(&self) {
+        let mut last_reset = self.last_reset.write().await;
+        let now = Instant::now();
+        
+        if now.duration_since(*last_reset) >= Duration::from_secs(1) {
+            self.bytes_this_second.store(0, Ordering::Relaxed);
+            self.operations_this_second.store(0, Ordering::Relaxed);
+            *last_reset = now;
+        }
+    }
+
+    /// Calculate wait time based on current usage and limit
+    async fn calculate_wait_time(&self, current: u64, limit: u64) -> Duration {
+        if current == 0 || limit == 0 {
+            return self.config.min_sleep_duration;
+        }
+
+        let utilization = current as f64 / limit as f64;
+        let base_sleep = self.config.min_sleep_duration.as_micros() as f64;
+        let max_sleep = self.config.max_sleep_duration.as_micros() as f64;
+
+        // Exponential backoff based on utilization
+        let sleep_micros = base_sleep * (utilization * utilization);
+        let sleep_micros = sleep_micros.min(max_sleep).max(base_sleep);
+
+        Duration::from_micros(sleep_micros as u64)
+    }
+
+    /// Update adaptive sleep duration based on recent activity
+    async fn update_adaptive_sleep(&self, bytes_processed: u64) {
+        let mut sleep_duration = self.adaptive_sleep_duration.write().await;
+        
+        // Simple adaptive algorithm: increase sleep if we're processing too much
+        let current_rate = bytes_processed as f64 / sleep_duration.as_secs_f64();
+        let target_rate = self.config.bytes_per_second as f64;
+        
+        if current_rate > target_rate * 1.1 {
+            // We're going too fast, increase sleep
+            *sleep_duration = Duration::from_micros(
+                (sleep_duration.as_micros() as f64 * 1.1) as u64
+            ).min(self.config.max_sleep_duration);
+        } else if current_rate < target_rate * 0.9 {
+            // We're going too slow, decrease sleep
+            *sleep_duration = Duration::from_micros(
+                (sleep_duration.as_micros() as f64 * 0.9) as u64
+            ).max(self.config.min_sleep_duration);
+        }
+    }
+
+    /// Get current bandwidth statistics
+    pub async fn statistics(&self) -> BandwidthStatistics {
+        let elapsed = self.start_time.elapsed();
+        let total_bytes = self.total_bytes_processed.load(Ordering::Relaxed);
+        let total_ops = self.total_operations_processed.load(Ordering::Relaxed);
+        let current_bytes = self.bytes_this_second.load(Ordering::Relaxed);
+        let current_ops = self.operations_this_second.load(Ordering::Relaxed);
+        let adaptive_sleep = *self.adaptive_sleep_duration.read().await;
+
+        BandwidthStatistics {
+            total_bytes_processed: total_bytes,
+            total_operations_processed: total_ops,
+            current_bytes_per_second: current_bytes,
+            current_operations_per_second: current_ops,
+            average_bytes_per_second: if elapsed.as_secs() > 0 {
+                total_bytes / elapsed.as_secs()
+            } else {
+                0
+            },
+            average_operations_per_second: if elapsed.as_secs() > 0 {
+                total_ops / elapsed.as_secs()
+            } else {
+                0
+            },
+            adaptive_sleep_duration: adaptive_sleep,
+            uptime: elapsed,
+        }
+    }
+
+    /// Reset all statistics
+    pub async fn reset_statistics(&self) {
+        self.total_bytes_processed.store(0, Ordering::Relaxed);
+        self.total_operations_processed.store(0, Ordering::Relaxed);
+        self.bytes_this_second.store(0, Ordering::Relaxed);
+        self.operations_this_second.store(0, Ordering::Relaxed);
+        *self.last_reset.write().await = Instant::now();
+        *self.adaptive_sleep_duration.write().await = self.config.min_sleep_duration;
+    }
+
+    /// Update configuration
+    pub async fn update_config(&self, new_config: BandwidthConfig) {
+        info!("Updating bandwidth limiter config: {:?}", new_config);
+        
+        // Reset adaptive sleep if adaptive throttling is disabled
+        if !new_config.adaptive_throttling {
+            *self.adaptive_sleep_duration.write().await = new_config.min_sleep_duration;
+        }
+        
+        // Note: We can't update the config struct itself since it's not wrapped in Arc<RwLock>
+        // In a real implementation, you might want to wrap the config in Arc<RwLock> as well
+        warn!("Config update not fully implemented - config struct is not mutable");
+    }
+}
+
+/// Statistics for bandwidth limiting
+#[derive(Debug, Clone)]
+pub struct BandwidthStatistics {
+    pub total_bytes_processed: u64,
+    pub total_operations_processed: u64,
+    pub current_bytes_per_second: u64,
+    pub current_operations_per_second: u64,
+    pub average_bytes_per_second: u64,
+    pub average_operations_per_second: u64,
+    pub adaptive_sleep_duration: Duration,
+    pub uptime: Duration,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tokio::time::Instant as TokioInstant;
+
+    #[tokio::test]
+    async fn test_bandwidth_limiter_creation() {
+        let config = BandwidthConfig::default();
+        let limiter = BandwidthLimiter::new(config);
+        let stats = limiter.statistics().await;
+        assert_eq!(stats.total_bytes_processed, 0);
+        assert_eq!(stats.total_operations_processed, 0);
+    }
+
+    #[tokio::test]
+    async fn test_bytes_limiting() {
+        let config = BandwidthConfig {
+            bytes_per_second: 1000, // 1KB/s
+            operations_per_second: 1000,
+            ..Default::default()
+        };
+        let limiter = BandwidthLimiter::new(config);
+        
+        let start = TokioInstant::now();
+        
+        // Process 500 bytes (should not be limited)
+        limiter.wait_for_bytes(500).await.unwrap();
+        
+        // Process another 600 bytes (should be limited)
+        limiter.wait_for_bytes(600).await.unwrap();
+        
+        let elapsed = start.elapsed();
+        assert!(elapsed >= Duration::from_millis(100)); // Should take some time due to limiting
+    }
+
+    #[tokio::test]
+    async fn test_operation_limiting() {
+        let config = BandwidthConfig {
+            bytes_per_second: 1000000, // 1MB/s
+            operations_per_second: 10, // 10 ops/s
+            ..Default::default()
+        };
+        let limiter = BandwidthLimiter::new(config);
+        
+        let start = TokioInstant::now();
+        
+        // Process 15 operations (should be limited)
+        for _ in 0..15 {
+            limiter.wait_for_operation().await.unwrap();
+        }
+        
+        let elapsed = start.elapsed();
+        assert!(elapsed >= Duration::from_millis(500)); // Should take some time due to limiting
+    }
+
+    #[tokio::test]
+    async fn test_statistics() {
+        let config = BandwidthConfig::default();
+        let limiter = BandwidthLimiter::new(config);
+        
+        limiter.wait_for_bytes(1000).await.unwrap();
+        limiter.wait_for_operation().await.unwrap();
+        
+        let stats = limiter.statistics().await;
+        assert_eq!(stats.total_bytes_processed, 1000);
+        assert_eq!(stats.total_operations_processed, 1);
+        assert!(stats.uptime > Duration::ZERO);
+    }
+} 

crates/ahm/src/scanner/disk_scanner.rs

@@ -0,0 +1,591 @@
+// 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,
+    path::Path,
+    sync::Arc,
+    time::{Duration, Instant, SystemTime},
+};
+
+use tokio::sync::RwLock;
+use tracing::{error, info};
+use anyhow;
+
+use crate::error::Result;
+use super::{HealthIssue, HealthIssueType, Severity};
+
+/// Configuration for disk scanning
+#[derive(Debug, Clone)]
+pub struct DiskScannerConfig {
+    /// Scan interval for disk health checks
+    pub scan_interval: Duration,
+    /// Minimum free space threshold (percentage)
+    pub min_free_space_percent: f64,
+    /// Maximum disk usage threshold (percentage)
+    pub max_disk_usage_percent: f64,
+    /// Minimum inode usage threshold (percentage)
+    pub min_inode_usage_percent: f64,
+    /// Maximum inode usage threshold (percentage)
+    pub max_inode_usage_percent: f64,
+    /// Whether to check disk I/O performance
+    pub check_io_performance: bool,
+    /// Whether to check disk temperature (if available)
+    pub check_temperature: bool,
+    /// Whether to check disk SMART status (if available)
+    pub check_smart_status: bool,
+    /// Timeout for individual disk operations
+    pub operation_timeout: Duration,
+    /// Maximum number of concurrent disk scans
+    pub max_concurrent_scans: usize,
+}
+
+impl Default for DiskScannerConfig {
+    fn default() -> Self {
+        Self {
+            scan_interval: Duration::from_secs(300), // 5 minutes
+            min_free_space_percent: 10.0, // 10% minimum free space
+            max_disk_usage_percent: 90.0, // 90% maximum usage
+            min_inode_usage_percent: 5.0, // 5% minimum inode usage
+            max_inode_usage_percent: 95.0, // 95% maximum inode usage
+            check_io_performance: true,
+            check_temperature: false, // Disabled by default
+            check_smart_status: false, // Disabled by default
+            operation_timeout: Duration::from_secs(30),
+            max_concurrent_scans: 4,
+        }
+    }
+}
+
+/// Disk information and health status
+#[derive(Debug, Clone)]
+pub struct DiskInfo {
+    pub device_path: String,
+    pub mount_point: String,
+    pub filesystem_type: String,
+    pub total_space: u64,
+    pub used_space: u64,
+    pub free_space: u64,
+    pub available_space: u64,
+    pub usage_percent: f64,
+    pub inode_total: Option<u64>,
+    pub inode_used: Option<u64>,
+    pub inode_free: Option<u64>,
+    pub inode_usage_percent: Option<f64>,
+    pub last_scan_time: SystemTime,
+    pub health_status: DiskHealthStatus,
+    pub performance_metrics: Option<DiskPerformanceMetrics>,
+    pub temperature: Option<f64>,
+    pub smart_status: Option<SmartStatus>,
+}
+
+/// Disk health status
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum DiskHealthStatus {
+    Healthy,
+    Warning,
+    Critical,
+    Unknown,
+}
+
+/// Disk performance metrics
+#[derive(Debug, Clone)]
+pub struct DiskPerformanceMetrics {
+    pub read_bytes_per_sec: f64,
+    pub write_bytes_per_sec: f64,
+    pub read_operations_per_sec: f64,
+    pub write_operations_per_sec: f64,
+    pub average_response_time_ms: f64,
+    pub queue_depth: f64,
+    pub utilization_percent: f64,
+    pub last_updated: SystemTime,
+}
+
+/// SMART status information
+#[derive(Debug, Clone)]
+pub struct SmartStatus {
+    pub overall_health: SmartHealthStatus,
+    pub temperature: Option<f64>,
+    pub power_on_hours: Option<u64>,
+    pub reallocated_sectors: Option<u64>,
+    pub pending_sectors: Option<u64>,
+    pub uncorrectable_sectors: Option<u64>,
+    pub attributes: HashMap<String, SmartAttribute>,
+}
+
+/// SMART health status
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum SmartHealthStatus {
+    Passed,
+    Failed,
+    Unknown,
+}
+
+/// SMART attribute
+#[derive(Debug, Clone)]
+pub struct SmartAttribute {
+    pub name: String,
+    pub value: u64,
+    pub worst: u64,
+    pub threshold: u64,
+    pub status: SmartAttributeStatus,
+}
+
+/// SMART attribute status
+#[derive(Debug, Clone, PartialEq, Eq)]
+pub enum SmartAttributeStatus {
+    Good,
+    Warning,
+    Critical,
+    Unknown,
+}
+
+/// Result of scanning a single disk
+#[derive(Debug, Clone)]
+pub struct DiskScanResult {
+    pub disk_info: DiskInfo,
+    pub health_issues: Vec<HealthIssue>,
+    pub scan_duration: Duration,
+    pub success: bool,
+    pub error_message: Option<String>,
+}
+
+/// Disk scanner for monitoring disk health and performance
+pub struct DiskScanner {
+    config: DiskScannerConfig,
+    statistics: Arc<RwLock<DiskScannerStatistics>>,
+    last_scan_results: Arc<RwLock<HashMap<String, DiskScanResult>>>,
+}
+
+/// Statistics for disk scanning
+#[derive(Debug, Clone, Default)]
+pub struct DiskScannerStatistics {
+    pub disks_scanned: u64,
+    pub disks_with_issues: u64,
+    pub total_issues_found: u64,
+    pub total_scan_time: Duration,
+    pub average_scan_time: Duration,
+    pub last_scan_time: Option<SystemTime>,
+    pub scan_cycles_completed: u64,
+    pub scan_cycles_failed: u64,
+}
+
+impl DiskScanner {
+    /// Create a new disk scanner
+    pub fn new(config: DiskScannerConfig) -> Self {
+        Self {
+            config,
+            statistics: Arc::new(RwLock::new(DiskScannerStatistics::default())),
+            last_scan_results: Arc::new(RwLock::new(HashMap::new())),
+        }
+    }
+
+    /// Scan all mounted disks
+    pub async fn scan_all_disks(&self) -> Result<Vec<DiskScanResult>> {
+        let scan_start = Instant::now();
+        let mut results = Vec::new();
+
+        // Get list of mounted filesystems
+        let mount_points = self.get_mount_points().await?;
+        
+        info!("Starting disk scan for {} mount points", mount_points.len());
+
+        // Scan each mount point
+        for mount_point in mount_points {
+            match self.scan_disk(&mount_point).await {
+                Ok(result) => {
+                    results.push(result.clone());
+                    
+                    // Store result for later reference
+                    let mut last_results = self.last_scan_results.write().await;
+                    last_results.insert(mount_point.clone(), result);
+                }
+                Err(e) => {
+                    error!("Failed to scan disk at {}: {}", mount_point, e);
+                    
+                    // Create error result
+                    let error_result = DiskScanResult {
+                        disk_info: DiskInfo {
+                            device_path: "unknown".to_string(),
+                            mount_point: mount_point.clone(),
+                            filesystem_type: "unknown".to_string(),
+                            total_space: 0,
+                            used_space: 0,
+                            free_space: 0,
+                            available_space: 0,
+                            usage_percent: 0.0,
+                            inode_total: None,
+                            inode_used: None,
+                            inode_free: None,
+                            inode_usage_percent: None,
+                            last_scan_time: SystemTime::now(),
+                            health_status: DiskHealthStatus::Unknown,
+                            performance_metrics: None,
+                            temperature: None,
+                            smart_status: None,
+                        },
+                        health_issues: vec![HealthIssue {
+                            issue_type: HealthIssueType::DiskReadError,
+                            severity: Severity::High,
+                            bucket: "system".to_string(),
+                            object: mount_point.clone(),
+                            description: format!("Failed to scan disk: {}", e),
+                            metadata: None,
+                        }],
+                        scan_duration: scan_start.elapsed(),
+                        success: false,
+                        error_message: Some(e.to_string()),
+                    };
+                    
+                    results.push(error_result);
+                }
+            }
+        }
+
+        // Update statistics
+        self.update_statistics(|stats| {
+            stats.disks_scanned += results.len() as u64;
+            stats.disks_with_issues += results.iter().filter(|r| !r.health_issues.is_empty()).count() as u64;
+            stats.total_issues_found += results.iter().map(|r| r.health_issues.len() as u64).sum::<u64>();
+            stats.total_scan_time += scan_start.elapsed();
+            stats.average_scan_time = Duration::from_millis(
+                stats.total_scan_time.as_millis() as u64 / stats.disks_scanned.max(1)
+            );
+            stats.last_scan_time = Some(SystemTime::now());
+            stats.scan_cycles_completed += 1;
+        }).await;
+
+        info!(
+            "Disk scan completed: {} disks, {} issues found in {:?}",
+            results.len(),
+            results.iter().map(|r| r.health_issues.len()).sum::<usize>(),
+            scan_start.elapsed()
+        );
+
+        Ok(results)
+    }
+
+    /// Scan a single disk
+    pub async fn scan_disk(&self, mount_point: &str) -> Result<DiskScanResult> {
+        let scan_start = Instant::now();
+        let mut health_issues = Vec::new();
+
+        // Get disk space information
+        let disk_info = self.get_disk_info(mount_point).await?;
+
+        // Check disk space usage
+        if disk_info.usage_percent > self.config.max_disk_usage_percent {
+            health_issues.push(HealthIssue {
+                issue_type: HealthIssueType::DiskFull,
+                severity: if disk_info.usage_percent > 95.0 { Severity::Critical } else { Severity::High },
+                bucket: "system".to_string(),
+                object: mount_point.to_string(),
+                description: format!("Disk usage is {}%, exceeds threshold of {}%", 
+                    disk_info.usage_percent, self.config.max_disk_usage_percent),
+                metadata: None,
+            });
+        }
+
+        if disk_info.usage_percent < self.config.min_free_space_percent {
+            health_issues.push(HealthIssue {
+                issue_type: HealthIssueType::DiskFull,
+                severity: Severity::Medium,
+                bucket: "system".to_string(),
+                object: mount_point.to_string(),
+                description: format!("Free space is only {}%, below threshold of {}%", 
+                    100.0 - disk_info.usage_percent, self.config.min_free_space_percent),
+                metadata: None,
+            });
+        }
+
+        // Check inode usage if available
+        if let Some(inode_usage) = disk_info.inode_usage_percent {
+            if inode_usage > self.config.max_inode_usage_percent {
+                health_issues.push(HealthIssue {
+                    issue_type: HealthIssueType::DiskFull,
+                    severity: if inode_usage > 95.0 { Severity::Critical } else { Severity::High },
+                    bucket: "system".to_string(),
+                    object: mount_point.to_string(),
+                    description: format!("Inode usage is {}%, exceeds threshold of {}%", 
+                        inode_usage, self.config.max_inode_usage_percent),
+                    metadata: None,
+                });
+            }
+        }
+
+        // Check I/O performance if enabled
+        if self.config.check_io_performance {
+            if let Some(metrics) = &disk_info.performance_metrics {
+                if metrics.utilization_percent > 90.0 {
+                    health_issues.push(HealthIssue {
+                        issue_type: HealthIssueType::DiskReadError,
+                        severity: Severity::Medium,
+                        bucket: "system".to_string(),
+                        object: mount_point.to_string(),
+                        description: format!("High disk utilization: {}%", metrics.utilization_percent),
+                        metadata: None,
+                    });
+                }
+
+                if metrics.average_response_time_ms > 100.0 {
+                    health_issues.push(HealthIssue {
+                        issue_type: HealthIssueType::DiskReadError,
+                        severity: Severity::Medium,
+                        bucket: "system".to_string(),
+                        object: mount_point.to_string(),
+                        description: format!("High disk response time: {}ms", metrics.average_response_time_ms),
+                        metadata: None,
+                    });
+                }
+            }
+        }
+
+        // Check temperature if enabled
+        if self.config.check_temperature {
+            if let Some(temp) = disk_info.temperature {
+                if temp > 60.0 {
+                    health_issues.push(HealthIssue {
+                        issue_type: HealthIssueType::DiskReadError,
+                        severity: if temp > 70.0 { Severity::Critical } else { Severity::High },
+                        bucket: "system".to_string(),
+                        object: mount_point.to_string(),
+                        description: format!("High disk temperature: {}°C", temp),
+                        metadata: None,
+                    });
+                }
+            }
+        }
+
+        // Check SMART status if enabled
+        if self.config.check_smart_status {
+            if let Some(smart) = &disk_info.smart_status {
+                if smart.overall_health == SmartHealthStatus::Failed {
+                    health_issues.push(HealthIssue {
+                        issue_type: HealthIssueType::DiskReadError,
+                        severity: Severity::Critical,
+                        bucket: "system".to_string(),
+                        object: mount_point.to_string(),
+                        description: "SMART health check failed".to_string(),
+                        metadata: None,
+                    });
+                }
+            }
+        }
+
+        let scan_duration = scan_start.elapsed();
+        let success = health_issues.is_empty();
+
+        Ok(DiskScanResult {
+            disk_info,
+            health_issues,
+            scan_duration,
+            success,
+            error_message: None,
+        })
+    }
+
+    /// Get list of mounted filesystems
+    async fn get_mount_points(&self) -> Result<Vec<String>> {
+        // TODO: Implement actual mount point detection
+        // For now, return common mount points
+        Ok(vec![
+            "/".to_string(),
+            "/data".to_string(),
+            "/var".to_string(),
+        ])
+    }
+
+    /// Get disk information for a mount point
+    async fn get_disk_info(&self, mount_point: &str) -> Result<DiskInfo> {
+        let path = Path::new(mount_point);
+        
+        // Get filesystem statistics using std::fs instead of nix for now
+        let _metadata = match std::fs::metadata(path) {
+            Ok(metadata) => metadata,
+            Err(e) => {
+                return Err(crate::error::Error::Other(anyhow::anyhow!("Failed to get filesystem stats: {}", e)));
+            }
+        };
+
+        // For now, use placeholder values since we can't easily get filesystem stats
+        let total_space = 1000000000; // 1GB placeholder
+        let free_space = 500000000;   // 500MB placeholder
+        let available_space = 450000000; // 450MB placeholder
+        let used_space = total_space - free_space;
+        let usage_percent = (used_space as f64 / total_space as f64) * 100.0;
+
+        // Get inode information (placeholder)
+        let inode_total = Some(1000000);
+        let inode_free = Some(500000);
+        let inode_used = Some(500000);
+        let inode_usage_percent = Some(50.0);
+
+        // Get filesystem type
+        let filesystem_type = self.get_filesystem_type(mount_point).await.unwrap_or_else(|_| "unknown".to_string());
+
+        // Get device path
+        let device_path = self.get_device_path(mount_point).await.unwrap_or_else(|_| "unknown".to_string());
+
+        // Get performance metrics if enabled
+        let performance_metrics = if self.config.check_io_performance {
+            self.get_performance_metrics(&device_path).await.ok()
+        } else {
+            None
+        };
+
+        // Get temperature if enabled
+        let temperature = if self.config.check_temperature {
+            self.get_disk_temperature(&device_path).await.ok().flatten()
+        } else {
+            None
+        };
+
+        // Get SMART status if enabled
+        let smart_status = if self.config.check_smart_status {
+            self.get_smart_status(&device_path).await.ok().flatten()
+        } else {
+            None
+        };
+
+        // Determine health status (placeholder - will be set by scan_disk method)
+        let health_status = DiskHealthStatus::Healthy;
+
+        Ok(DiskInfo {
+            device_path,
+            mount_point: mount_point.to_string(),
+            filesystem_type,
+            total_space,
+            used_space,
+            free_space,
+            available_space,
+            usage_percent,
+            inode_total,
+            inode_used,
+            inode_free,
+            inode_usage_percent,
+            last_scan_time: SystemTime::now(),
+            health_status,
+            performance_metrics,
+            temperature,
+            smart_status,
+        })
+    }
+
+    /// Get filesystem type for a mount point
+    async fn get_filesystem_type(&self, _mount_point: &str) -> Result<String> {
+        // TODO: Implement filesystem type detection
+        // For now, return a placeholder
+        Ok("ext4".to_string())
+    }
+
+    /// Get device path for a mount point
+    async fn get_device_path(&self, _mount_point: &str) -> Result<String> {
+        // TODO: Implement device path detection
+        // For now, return a placeholder
+        Ok("/dev/sda1".to_string())
+    }
+
+    /// Get disk performance metrics
+    async fn get_performance_metrics(&self, _device_path: &str) -> Result<DiskPerformanceMetrics> {
+        // TODO: Implement performance metrics collection
+        // For now, return placeholder metrics
+        Ok(DiskPerformanceMetrics {
+            read_bytes_per_sec: 1000000.0, // 1MB/s
+            write_bytes_per_sec: 500000.0, // 500KB/s
+            read_operations_per_sec: 100.0,
+            write_operations_per_sec: 50.0,
+            average_response_time_ms: 5.0,
+            queue_depth: 1.0,
+            utilization_percent: 10.0,
+            last_updated: SystemTime::now(),
+        })
+    }
+
+    /// Get disk temperature
+    async fn get_disk_temperature(&self, _device_path: &str) -> Result<Option<f64>> {
+        // TODO: Implement temperature monitoring
+        // For now, return None (temperature not available)
+        Ok(None)
+    }
+
+    /// Get SMART status
+    async fn get_smart_status(&self, _device_path: &str) -> Result<Option<SmartStatus>> {
+        // TODO: Implement SMART status checking
+        // For now, return None (SMART not available)
+        Ok(None)
+    }
+
+    /// Update scanner statistics
+    async fn update_statistics<F>(&self, update_fn: F)
+    where
+        F: FnOnce(&mut DiskScannerStatistics),
+    {
+        let mut stats = self.statistics.write().await;
+        update_fn(&mut stats);
+    }
+
+    /// Get current statistics
+    pub async fn statistics(&self) -> DiskScannerStatistics {
+        self.statistics.read().await.clone()
+    }
+
+    /// Get last scan results
+    pub async fn last_scan_results(&self) -> HashMap<String, DiskScanResult> {
+        self.last_scan_results.read().await.clone()
+    }
+
+    /// Reset statistics
+    pub async fn reset_statistics(&self) {
+        let mut stats = self.statistics.write().await;
+        *stats = DiskScannerStatistics::default();
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn test_disk_scanner_creation() {
+        let config = DiskScannerConfig::default();
+        let scanner = DiskScanner::new(config);
+        assert_eq!(scanner.statistics().await.disks_scanned, 0);
+    }
+
+    #[tokio::test]
+    async fn test_disk_info_creation() {
+        let disk_info = DiskInfo {
+            device_path: "/dev/sda1".to_string(),
+            mount_point: "/".to_string(),
+            filesystem_type: "ext4".to_string(),
+            total_space: 1000000000,
+            used_space: 500000000,
+            free_space: 500000000,
+            available_space: 450000000,
+            usage_percent: 50.0,
+            inode_total: Some(1000000),
+            inode_used: Some(500000),
+            inode_free: Some(500000),
+            inode_usage_percent: Some(50.0),
+            last_scan_time: SystemTime::now(),
+            health_status: DiskHealthStatus::Healthy,
+            performance_metrics: None,
+            temperature: None,
+            smart_status: None,
+        };
+
+        assert_eq!(disk_info.usage_percent, 50.0);
+        assert_eq!(disk_info.health_status, DiskHealthStatus::Healthy);
+    }
+} 

crates/ahm/src/scanner/engine.rs

@@ -0,0 +1,536 @@
+// 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,
+    path::{Path, PathBuf},
+    sync::Arc,
+    time::{Duration, Instant, SystemTime},
+};
+
+use tokio::{
+    sync::{broadcast, RwLock},
+    time::sleep,
+};
+use tracing::{error, info, warn};
+use tokio_util::sync::CancellationToken;
+
+use crate::{core, error::Result, metrics, SystemEvent};
+use crate::core::Status;
+use super::{HealthIssue, HealthIssueType, Severity};
+
+/// Represents a discovered object during scanning
+#[derive(Debug, Clone)]
+pub struct ScannedObject {
+    pub bucket: String,
+    pub object: String,
+    pub version_id: Option<String>,
+    pub path: PathBuf,
+    pub size: u64,
+    pub modified_time: SystemTime,
+    pub metadata: HashMap<String, String>,
+    pub health_issues: Vec<HealthIssue>,
+}
+
+/// Configuration for the scanner engine
+#[derive(Debug, Clone)]
+pub struct EngineConfig {
+    /// Root directory to scan
+    pub root_path: String,
+    /// Maximum number of concurrent scan workers
+    pub max_workers: usize,
+    /// Scan interval between cycles
+    pub scan_interval: Duration,
+    /// Bandwidth limit for scanning (bytes per second)
+    pub bandwidth_limit: Option<u64>,
+    /// Whether to enable deep scanning (bitrot detection)
+    pub enable_deep_scan: bool,
+    /// Probability of healing objects during scan (1 in N)
+    pub heal_probability: u32,
+    /// Maximum folders to scan before compacting
+    pub max_folders_before_compact: u64,
+    /// Sleep duration between folder scans
+    pub folder_sleep_duration: Duration,
+}
+
+impl Default for EngineConfig {
+    fn default() -> Self {
+        Self {
+            root_path: "/data".to_string(),
+            max_workers: 4,
+            scan_interval: Duration::from_secs(300), // 5 minutes
+            bandwidth_limit: None,
+            enable_deep_scan: false,
+            heal_probability: 1024, // 1 in 1024 objects
+            max_folders_before_compact: 10000,
+            folder_sleep_duration: Duration::from_millis(1),
+        }
+    }
+}
+
+/// Scanner statistics
+#[derive(Debug, Clone, Default)]
+pub struct ScannerStatistics {
+    pub objects_scanned: u64,
+    pub bytes_scanned: u64,
+    pub issues_found: u64,
+    pub scan_duration: Duration,
+    pub scan_rate_objects_per_sec: f64,
+    pub scan_rate_bytes_per_sec: f64,
+    pub folders_scanned: u64,
+    pub objects_with_issues: u64,
+}
+
+/// Main scanner engine
+pub struct Engine {
+    config: EngineConfig,
+    coordinator: Arc<core::Coordinator>,
+    metrics: Arc<metrics::Collector>,
+    cancel_token: CancellationToken,
+    status: Arc<RwLock<Status>>,
+    statistics: Arc<RwLock<ScannerStatistics>>,
+    scan_cycle: Arc<RwLock<u64>>,
+}
+
+impl Engine {
+    /// Create a new scanner engine
+    pub async fn new(
+        config: EngineConfig,
+        coordinator: Arc<core::Coordinator>,
+        metrics: Arc<metrics::Collector>,
+        cancel_token: CancellationToken,
+    ) -> Result<Self> {
+        let engine = Self {
+            config,
+            coordinator,
+            metrics,
+            cancel_token,
+            status: Arc::new(RwLock::new(Status::Initializing)),
+            statistics: Arc::new(RwLock::new(ScannerStatistics::default())),
+            scan_cycle: Arc::new(RwLock::new(0)),
+        };
+
+        info!("Scanner engine created with config: {:?}", engine.config);
+        Ok(engine)
+    }
+
+    /// Start the scanner engine
+    pub async fn start(&self) -> Result<()> {
+        info!("Starting scanner engine");
+        *self.status.write().await = Status::Running;
+
+        let engine = self.clone_for_background();
+        tokio::spawn(async move {
+            if let Err(e) = engine.run_scan_loop().await {
+                error!("Scanner engine error: {}", e);
+            }
+        });
+
+        Ok(())
+    }
+
+    /// Stop the scanner engine
+    pub async fn stop(&self) -> Result<()> {
+        info!("Stopping scanner engine");
+        *self.status.write().await = Status::Stopping;
+        self.cancel_token.cancel();
+        *self.status.write().await = Status::Stopped;
+        Ok(())
+    }
+
+    /// Get current status
+    pub async fn status(&self) -> Status {
+        self.status.read().await.clone()
+    }
+
+    /// Get current statistics
+    pub async fn statistics(&self) -> ScannerStatistics {
+        self.statistics.read().await.clone()
+    }
+
+    /// Clone the engine for background tasks
+    fn clone_for_background(&self) -> Arc<Self> {
+        Arc::new(Self {
+            config: self.config.clone(),
+            coordinator: self.coordinator.clone(),
+            metrics: self.metrics.clone(),
+            cancel_token: self.cancel_token.clone(),
+            status: self.status.clone(),
+            statistics: self.statistics.clone(),
+            scan_cycle: self.scan_cycle.clone(),
+        })
+    }
+
+    /// Main scan loop
+    async fn run_scan_loop(&self) -> Result<()> {
+        info!("Scanner engine loop started");
+
+        loop {
+            tokio::select! {
+                _ = self.cancel_token.cancelled() => {
+                    info!("Scanner engine received cancellation signal");
+                    break;
+                }
+                _ = sleep(self.config.scan_interval) => {
+                    if let Err(e) = self.run_scan_cycle().await {
+                        error!("Scan cycle failed: {}", e);
+                    }
+                }
+            }
+        }
+
+        Ok(())
+    }
+
+    /// Run a single scan cycle
+    async fn run_scan_cycle(&self) -> Result<()> {
+        let cycle_start = Instant::now();
+        let cycle = {
+            let mut cycle_guard = self.scan_cycle.write().await;
+            *cycle_guard += 1;
+            *cycle_guard
+        };
+
+        info!("Starting scan cycle {}", cycle);
+
+        // Reset statistics for new cycle
+        {
+            let mut stats = self.statistics.write().await;
+            *stats = ScannerStatistics::default();
+        }
+
+        // Scan the root directory
+        let scan_result = self.scan_directory(&self.config.root_path).await?;
+
+        // Update statistics
+        {
+            let mut stats = self.statistics.write().await;
+            stats.scan_duration = cycle_start.elapsed();
+            stats.objects_scanned = scan_result.objects.len() as u64;
+            stats.bytes_scanned = scan_result.total_size;
+            stats.issues_found = scan_result.total_issues;
+            stats.folders_scanned = scan_result.folders_scanned;
+            stats.objects_with_issues = scan_result.objects_with_issues;
+
+            if stats.scan_duration.as_secs() > 0 {
+                stats.scan_rate_objects_per_sec = stats.objects_scanned as f64 / stats.scan_duration.as_secs() as f64;
+                stats.scan_rate_bytes_per_sec = stats.bytes_scanned as f64 / stats.scan_duration.as_secs() as f64;
+            }
+        }
+
+        // Publish scan completion event
+        let scan_report = crate::scanner::ScanReport {
+            scan_id: cycle.to_string(),
+            status: "completed".to_string(),
+            summary: format!("Scanned {} objects, found {} issues", scan_result.objects.len(), scan_result.total_issues),
+            issues_found: scan_result.total_issues,
+        };
+
+        self.coordinator.publish_event(SystemEvent::ScanCompleted(scan_report)).await?;
+
+        info!(
+            "Scan cycle {} completed: {} objects, {} bytes, {} issues in {:?}",
+            cycle,
+            scan_result.objects.len(),
+            scan_result.total_size,
+            scan_result.total_issues,
+            cycle_start.elapsed()
+        );
+
+        Ok(())
+    }
+
+    /// Scan a directory recursively
+    async fn scan_directory(&self, path: &str) -> Result<ScanResult> {
+        let mut result = ScanResult::default();
+        let path_buf = PathBuf::from(path);
+
+        if !path_buf.exists() {
+            warn!("Scan path does not exist: {}", path);
+            return Ok(result);
+        }
+
+        if !path_buf.is_dir() {
+            warn!("Scan path is not a directory: {}", path);
+            return Ok(result);
+        }
+
+        self.scan_directory_recursive(&path_buf, &mut result).await?;
+        Ok(result)
+    }
+
+    /// Recursively scan a directory
+    async fn scan_directory_recursive(&self, dir_path: &Path, result: &mut ScanResult) -> Result<()> {
+        result.folders_scanned += 1;
+
+        // Check for cancellation
+        if self.cancel_token.is_cancelled() {
+            return Ok(());
+        }
+
+        let entries = match std::fs::read_dir(dir_path) {
+            Ok(entries) => entries,
+            Err(e) => {
+                warn!("Failed to read directory {}: {}", dir_path.display(), e);
+                return Ok(());
+            }
+        };
+
+        for entry in entries {
+            if self.cancel_token.is_cancelled() {
+                break;
+            }
+
+            let entry = match entry {
+                Ok(entry) => entry,
+                Err(e) => {
+                    warn!("Failed to read directory entry: {}", e);
+                    continue;
+                }
+            };
+
+            let file_path = entry.path();
+            let _path_str = file_path.to_string_lossy();
+            let entry_name = file_path.file_name()
+                .and_then(|n| n.to_str())
+                .unwrap_or("unknown");
+
+            // Skip hidden files and system files
+            if entry_name.starts_with('.') || entry_name == ".." || entry_name == "." {
+                continue;
+            }
+
+            if file_path.is_dir() {
+                // Recursively scan subdirectories
+                Box::pin(self.scan_directory_recursive(&file_path, result)).await?;
+            } else if file_path.is_file() {
+                // Scan individual file
+                if let Some(scanned_object) = self.scan_object(&file_path).await? {
+                    result.objects.push(scanned_object.clone());
+                    result.total_size += scanned_object.size;
+
+                    if !scanned_object.health_issues.is_empty() {
+                        result.objects_with_issues += 1;
+                        result.total_issues += scanned_object.health_issues.len() as u64;
+
+                        // Publish health issues
+                        for issue in &scanned_object.health_issues {
+                            let health_issue = crate::scanner::HealthIssue {
+                                issue_type: issue.issue_type.clone(),
+                                severity: issue.severity,
+                                bucket: scanned_object.bucket.clone(),
+                                object: scanned_object.object.clone(),
+                                description: issue.description.clone(),
+                                metadata: None, // TODO: Convert HashMap to ObjectMetadata
+                            };
+
+                            self.coordinator.publish_event(SystemEvent::HealthIssueDetected(health_issue)).await?;
+                        }
+                    }
+
+                    // Publish object discovered event
+                    let metadata = crate::ObjectMetadata {
+                        size: scanned_object.size,
+                        mod_time: scanned_object.modified_time.duration_since(SystemTime::UNIX_EPOCH)
+                            .unwrap_or_default()
+                            .as_secs() as i64,
+                        content_type: "application/octet-stream".to_string(),
+                        etag: "".to_string(), // TODO: Calculate actual ETag
+                    };
+
+                    self.coordinator.publish_event(SystemEvent::ObjectDiscovered {
+                        bucket: scanned_object.bucket.clone(),
+                        object: scanned_object.object.clone(),
+                        version_id: scanned_object.version_id.clone(),
+                        metadata,
+                    }).await?;
+                }
+            }
+
+            // Sleep between items to avoid overwhelming the system
+            sleep(self.config.folder_sleep_duration).await;
+        }
+
+        Ok(())
+    }
+
+    /// Scan a single object file
+    async fn scan_object(&self, file_path: &Path) -> Result<Option<ScannedObject>> {
+        let metadata = match std::fs::metadata(file_path) {
+            Ok(metadata) => metadata,
+            Err(e) => {
+                warn!("Failed to read file metadata {}: {}", file_path.display(), e);
+                return Ok(None);
+            }
+        };
+
+        // Extract bucket and object from path
+        let (bucket, object) = self.extract_bucket_object_from_path(file_path)?;
+        if bucket.is_empty() || object.is_empty() {
+            return Ok(None);
+        }
+
+        // Check for health issues
+        let health_issues = self.check_object_health(file_path, &metadata).await?;
+
+        let scanned_object = ScannedObject {
+            bucket,
+            object,
+            version_id: None, // TODO: Extract version ID from path
+            path: file_path.to_path_buf(),
+            size: metadata.len(),
+            modified_time: metadata.modified().unwrap_or(SystemTime::now()),
+            metadata: HashMap::new(), // TODO: Extract metadata
+            health_issues,
+        };
+
+        Ok(Some(scanned_object))
+    }
+
+    /// Extract bucket and object name from file path
+    fn extract_bucket_object_from_path(&self, file_path: &Path) -> Result<(String, String)> {
+        let _path_str = file_path.to_string_lossy();
+        let root_path = Path::new(&self.config.root_path);
+        
+        if let Ok(relative_path) = file_path.strip_prefix(root_path) {
+            let components: Vec<&str> = relative_path.components()
+                .filter_map(|c| c.as_os_str().to_str())
+                .collect();
+
+            if components.len() >= 2 {
+                let bucket = components[0].to_string();
+                let object = components[1..].join("/");
+                return Ok((bucket, object));
+            }
+        }
+
+        Ok((String::new(), String::new()))
+    }
+
+    /// Check object health and detect issues
+    async fn check_object_health(&self, file_path: &Path, metadata: &std::fs::Metadata) -> Result<Vec<HealthIssue>> {
+        let mut issues = Vec::new();
+
+        // Extract bucket and object from path for health issues
+        let (bucket, object) = self.extract_bucket_object_from_path(file_path)?;
+
+        // Check file size
+        if metadata.len() == 0 {
+            issues.push(HealthIssue {
+                issue_type: HealthIssueType::ObjectTooSmall,
+                severity: Severity::Low,
+                bucket: bucket.clone(),
+                object: object.clone(),
+                description: "Object has zero size".to_string(),
+                metadata: None,
+            });
+        }
+
+        // Check file permissions
+        if !metadata.permissions().readonly() {
+            issues.push(HealthIssue {
+                issue_type: HealthIssueType::PolicyViolation,
+                severity: Severity::Medium,
+                bucket: bucket.clone(),
+                object: object.clone(),
+                description: "Object is not read-only".to_string(),
+                metadata: None,
+            });
+        }
+
+        // TODO: Add more health checks:
+        // - Checksum verification
+        // - Replication status
+        // - Encryption status
+        // - Metadata consistency
+        // - Disk health
+
+        Ok(issues)
+    }
+
+    /// Start scanning operations
+    pub async fn start_scan(&self) -> Result<()> {
+        let mut status = self.status.write().await;
+        *status = Status::Running;
+        info!("Scanning operations started");
+        Ok(())
+    }
+
+    /// Stop scanning operations
+    pub async fn stop_scan(&self) -> Result<()> {
+        let mut status = self.status.write().await;
+        *status = Status::Stopped;
+        info!("Scanning operations stopped");
+        Ok(())
+    }
+
+    /// Get engine configuration
+    pub async fn get_config(&self) -> ScanConfig {
+        self.config.clone()
+    }
+}
+
+/// Result of a scan operation
+#[derive(Debug, Clone, Default)]
+pub struct ScanResult {
+    pub objects: Vec<ScannedObject>,
+    pub total_size: u64,
+    pub total_issues: u64,
+    pub folders_scanned: u64,
+    pub objects_with_issues: u64,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tokio::time::Duration;
+
+    #[tokio::test]
+    async fn test_engine_creation() {
+        let config = EngineConfig::default();
+        let coordinator = Arc::new(core::Coordinator::new(
+            core::CoordinatorConfig::default(),
+            Arc::new(metrics::Collector::new(metrics::CollectorConfig::default()).await.unwrap()),
+            CancellationToken::new(),
+        ).await.unwrap());
+        let metrics = Arc::new(metrics::Collector::new(metrics::CollectorConfig::default()).await.unwrap());
+        let cancel_token = CancellationToken::new();
+
+        let engine = Engine::new(config, coordinator, metrics, cancel_token).await;
+        assert!(engine.is_ok());
+    }
+
+    #[tokio::test]
+    async fn test_path_extraction() {
+        let config = EngineConfig {
+            root_path: "/data".to_string(),
+            ..Default::default()
+        };
+        let coordinator = Arc::new(core::Coordinator::new(
+            core::CoordinatorConfig::default(),
+            Arc::new(metrics::Collector::new(metrics::CollectorConfig::default()).await.unwrap()),
+            CancellationToken::new(),
+        ).await.unwrap());
+        let metrics = Arc::new(metrics::Collector::new(metrics::CollectorConfig::default()).await.unwrap());
+        let cancel_token = CancellationToken::new();
+
+        let engine = Engine::new(config, coordinator, metrics, cancel_token).await.unwrap();
+        
+        let test_path = Path::new("/data/bucket1/object1.txt");
+        let (bucket, object) = engine.extract_bucket_object_from_path(test_path).unwrap();
+        
+        assert_eq!(bucket, "bucket1");
+        assert_eq!(object, "object1.txt");
+    }
+} 

crates/ahm/src/scanner/metrics_collector.rs

@@ -0,0 +1,526 @@
+// 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, Instant, SystemTime},
+};
+
+use tokio::sync::RwLock;
+use tracing::{debug, error, info, warn};
+
+use crate::error::Result;
+use super::{HealthIssue, HealthIssueType, Severity};
+
+/// Configuration for metrics collection
+#[derive(Debug, Clone)]
+pub struct MetricsConfig {
+    /// Collection interval for metrics
+    pub collection_interval: Duration,
+    /// Retention period for historical metrics
+    pub retention_period: Duration,
+    /// Maximum number of data points to keep in memory
+    pub max_data_points: usize,
+    /// Whether to enable detailed metrics collection
+    pub enable_detailed_metrics: bool,
+    /// Whether to enable performance profiling
+    pub enable_profiling: bool,
+    /// Whether to enable resource usage tracking
+    pub enable_resource_tracking: bool,
+}
+
+impl Default for MetricsConfig {
+    fn default() -> Self {
+        Self {
+            collection_interval: Duration::from_secs(60), // 1 minute
+            retention_period: Duration::from_secs(3600 * 24), // 24 hours
+            max_data_points: 1440, // 24 hours worth of minute-level data
+            enable_detailed_metrics: true,
+            enable_profiling: false,
+            enable_resource_tracking: true,
+        }
+    }
+}
+
+/// Scanner performance metrics
+#[derive(Debug, Clone)]
+pub struct ScannerMetrics {
+    /// Objects scanned per second
+    pub objects_per_second: f64,
+    /// Bytes scanned per second
+    pub bytes_per_second: f64,
+    /// Average scan time per object
+    pub avg_scan_time_per_object: Duration,
+    /// Total objects scanned in current cycle
+    pub total_objects_scanned: u64,
+    /// Total bytes scanned in current cycle
+    pub total_bytes_scanned: u64,
+    /// Number of health issues detected
+    pub health_issues_detected: u64,
+    /// Scan success rate (percentage)
+    pub success_rate: f64,
+    /// Current scan cycle duration
+    pub current_cycle_duration: Duration,
+    /// Average scan cycle duration
+    pub avg_cycle_duration: Duration,
+    /// Last scan completion time
+    pub last_scan_completion: Option<SystemTime>,
+}
+
+/// Resource usage metrics
+#[derive(Debug, Clone)]
+pub struct ResourceMetrics {
+    /// CPU usage percentage
+    pub cpu_usage_percent: f64,
+    /// Memory usage in bytes
+    pub memory_usage_bytes: u64,
+    /// Memory usage percentage
+    pub memory_usage_percent: f64,
+    /// Disk I/O operations per second
+    pub disk_io_ops_per_sec: f64,
+    /// Disk I/O bytes per second
+    pub disk_io_bytes_per_sec: f64,
+    /// Network I/O bytes per second
+    pub network_io_bytes_per_sec: f64,
+    /// Number of active threads
+    pub active_threads: u32,
+    /// Number of open file descriptors
+    pub open_file_descriptors: u32,
+}
+
+/// Health metrics summary
+#[derive(Debug, Clone)]
+pub struct HealthMetrics {
+    /// Total health issues by severity
+    pub issues_by_severity: HashMap<Severity, u64>,
+    /// Total health issues by type
+    pub issues_by_type: HashMap<HealthIssueType, u64>,
+    /// Objects with health issues
+    pub objects_with_issues: u64,
+    /// Percentage of objects with issues
+    pub objects_with_issues_percent: f64,
+    /// Last health check time
+    pub last_health_check: SystemTime,
+    /// Health score (0-100, higher is better)
+    pub health_score: f64,
+}
+
+/// Historical metrics data point
+#[derive(Debug, Clone)]
+pub struct MetricsDataPoint {
+    pub timestamp: SystemTime,
+    pub scanner_metrics: ScannerMetrics,
+    pub resource_metrics: ResourceMetrics,
+    pub health_metrics: HealthMetrics,
+}
+
+/// Metrics collector for scanner system
+pub struct MetricsCollector {
+    config: MetricsConfig,
+    current_metrics: Arc<RwLock<CurrentMetrics>>,
+    historical_data: Arc<RwLock<Vec<MetricsDataPoint>>>,
+    collection_start_time: Instant,
+}
+
+/// Current metrics state
+#[derive(Debug, Clone)]
+pub struct CurrentMetrics {
+    pub scanner_metrics: ScannerMetrics,
+    pub resource_metrics: ResourceMetrics,
+    pub health_metrics: HealthMetrics,
+    pub last_update: SystemTime,
+}
+
+impl MetricsCollector {
+    /// Create a new metrics collector
+    pub fn new(config: MetricsConfig) -> Self {
+        let collector = Self {
+            config,
+            current_metrics: Arc::new(RwLock::new(CurrentMetrics {
+                scanner_metrics: ScannerMetrics {
+                    objects_per_second: 0.0,
+                    bytes_per_second: 0.0,
+                    avg_scan_time_per_object: Duration::ZERO,
+                    total_objects_scanned: 0,
+                    total_bytes_scanned: 0,
+                    health_issues_detected: 0,
+                    success_rate: 100.0,
+                    current_cycle_duration: Duration::ZERO,
+                    avg_cycle_duration: Duration::ZERO,
+                    last_scan_completion: None,
+                },
+                resource_metrics: ResourceMetrics {
+                    cpu_usage_percent: 0.0,
+                    memory_usage_bytes: 0,
+                    memory_usage_percent: 0.0,
+                    disk_io_ops_per_sec: 0.0,
+                    disk_io_bytes_per_sec: 0.0,
+                    network_io_bytes_per_sec: 0.0,
+                    active_threads: 0,
+                    open_file_descriptors: 0,
+                },
+                health_metrics: HealthMetrics {
+                    issues_by_severity: HashMap::new(),
+                    issues_by_type: HashMap::new(),
+                    objects_with_issues: 0,
+                    objects_with_issues_percent: 0.0,
+                    last_health_check: SystemTime::now(),
+                    health_score: 100.0,
+                },
+                last_update: SystemTime::now(),
+            })),
+            historical_data: Arc::new(RwLock::new(Vec::new())),
+            collection_start_time: Instant::now(),
+        };
+
+        info!("Metrics collector created with config: {:?}", collector.config);
+        collector
+    }
+
+    /// Start metrics collection
+    pub async fn start_collection(&self) -> Result<()> {
+        info!("Starting metrics collection");
+        
+        let collector = self.clone_for_background();
+        tokio::spawn(async move {
+            if let Err(e) = collector.run_collection_loop().await {
+                error!("Metrics collection error: {}", e);
+            }
+        });
+
+        Ok(())
+    }
+
+    /// Stop metrics collection
+    pub async fn stop_collection(&self) -> Result<()> {
+        info!("Stopping metrics collection");
+        Ok(())
+    }
+
+    /// Update scanner metrics
+    pub async fn update_scanner_metrics(&self, metrics: ScannerMetrics) -> Result<()> {
+        let mut current = self.current_metrics.write().await;
+        current.scanner_metrics = metrics;
+        current.last_update = SystemTime::now();
+        Ok(())
+    }
+
+    /// Update resource metrics
+    pub async fn update_resource_metrics(&self, metrics: ResourceMetrics) -> Result<()> {
+        let mut current = self.current_metrics.write().await;
+        current.resource_metrics = metrics;
+        current.last_update = SystemTime::now();
+        Ok(())
+    }
+
+    /// Update health metrics
+    pub async fn update_health_metrics(&self, metrics: HealthMetrics) -> Result<()> {
+        let mut current = self.current_metrics.write().await;
+        current.health_metrics = metrics;
+        current.last_update = SystemTime::now();
+        Ok(())
+    }
+
+    /// Record a health issue
+    pub async fn record_health_issue(&self, issue: &HealthIssue) -> Result<()> {
+        let mut current = self.current_metrics.write().await;
+        
+        // Update severity count
+        *current.health_metrics.issues_by_severity.entry(issue.severity).or_insert(0) += 1;
+        
+        // Update type count
+        *current.health_metrics.issues_by_type.entry(issue.issue_type.clone()).or_insert(0) += 1;
+        
+        // Update scanner metrics
+        current.scanner_metrics.health_issues_detected += 1;
+        
+        current.last_update = SystemTime::now();
+        Ok(())
+    }
+
+    /// Get current metrics
+    pub async fn current_metrics(&self) -> CurrentMetrics {
+        self.current_metrics.read().await.clone()
+    }
+
+    /// Get historical metrics
+    pub async fn historical_metrics(&self, duration: Duration) -> Vec<MetricsDataPoint> {
+        let historical = self.historical_data.read().await;
+        let cutoff_time = SystemTime::now() - duration;
+        
+        historical.iter()
+            .filter(|point| point.timestamp >= cutoff_time)
+            .cloned()
+            .collect()
+    }
+
+    /// Get metrics summary
+    pub async fn metrics_summary(&self) -> MetricsSummary {
+        let current = self.current_metrics.read().await;
+        let historical = self.historical_data.read().await;
+        
+        let uptime = self.collection_start_time.elapsed();
+        let total_data_points = historical.len();
+        
+        // Calculate averages from historical data
+        let avg_objects_per_sec = if !historical.is_empty() {
+            historical.iter()
+                .map(|point| point.scanner_metrics.objects_per_second)
+                .sum::<f64>() / historical.len() as f64
+        } else {
+            0.0
+        };
+        
+        let avg_bytes_per_sec = if !historical.is_empty() {
+            historical.iter()
+                .map(|point| point.scanner_metrics.bytes_per_second)
+                .sum::<f64>() / historical.len() as f64
+        } else {
+            0.0
+        };
+        
+        let avg_cpu_usage = if !historical.is_empty() {
+            historical.iter()
+                .map(|point| point.resource_metrics.cpu_usage_percent)
+                .sum::<f64>() / historical.len() as f64
+        } else {
+            0.0
+        };
+        
+        let avg_memory_usage = if !historical.is_empty() {
+            historical.iter()
+                .map(|point| point.resource_metrics.memory_usage_percent)
+                .sum::<f64>() / historical.len() as f64
+        } else {
+            0.0
+        };
+        
+        MetricsSummary {
+            uptime,
+            total_data_points,
+            current_scanner_metrics: current.scanner_metrics.clone(),
+            current_resource_metrics: current.resource_metrics.clone(),
+            current_health_metrics: current.health_metrics.clone(),
+            avg_objects_per_sec,
+            avg_bytes_per_sec,
+            avg_cpu_usage,
+            avg_memory_usage,
+            last_update: current.last_update,
+        }
+    }
+
+    /// Clone the collector for background tasks
+    fn clone_for_background(&self) -> Arc<Self> {
+        Arc::new(Self {
+            config: self.config.clone(),
+            current_metrics: self.current_metrics.clone(),
+            historical_data: self.historical_data.clone(),
+            collection_start_time: self.collection_start_time,
+        })
+    }
+
+    /// Main collection loop
+    async fn run_collection_loop(&self) -> Result<()> {
+        info!("Metrics collection loop started");
+
+        loop {
+            // Collect current metrics
+            self.collect_current_metrics().await?;
+            
+            // Store historical data point
+            self.store_historical_data_point().await?;
+            
+            // Clean up old data
+            self.cleanup_old_data().await?;
+            
+            // Wait for next collection interval
+            tokio::time::sleep(self.config.collection_interval).await;
+        }
+    }
+
+    /// Collect current system metrics
+    async fn collect_current_metrics(&self) -> Result<()> {
+        if self.config.enable_resource_tracking {
+            let resource_metrics = self.collect_resource_metrics().await?;
+            self.update_resource_metrics(resource_metrics).await?;
+        }
+        
+        Ok(())
+    }
+
+    /// Collect resource usage metrics
+    async fn collect_resource_metrics(&self) -> Result<ResourceMetrics> {
+        // TODO: Implement actual resource metrics collection
+        // For now, return placeholder metrics
+        Ok(ResourceMetrics {
+            cpu_usage_percent: 0.0,
+            memory_usage_bytes: 0,
+            memory_usage_percent: 0.0,
+            disk_io_ops_per_sec: 0.0,
+            disk_io_bytes_per_sec: 0.0,
+            network_io_bytes_per_sec: 0.0,
+            active_threads: 0,
+            open_file_descriptors: 0,
+        })
+    }
+
+    /// Store current metrics as historical data point
+    async fn store_historical_data_point(&self) -> Result<()> {
+        let current = self.current_metrics.read().await;
+        let data_point = MetricsDataPoint {
+            timestamp: SystemTime::now(),
+            scanner_metrics: current.scanner_metrics.clone(),
+            resource_metrics: current.resource_metrics.clone(),
+            health_metrics: current.health_metrics.clone(),
+        };
+        
+        let mut historical = self.historical_data.write().await;
+        historical.push(data_point);
+        
+        // Limit the number of data points
+        if historical.len() > self.config.max_data_points {
+            historical.remove(0);
+        }
+        
+        Ok(())
+    }
+
+    /// Clean up old historical data
+    async fn cleanup_old_data(&self) -> Result<()> {
+        let cutoff_time = SystemTime::now() - self.config.retention_period;
+        let mut historical = self.historical_data.write().await;
+        
+        historical.retain(|point| point.timestamp >= cutoff_time);
+        
+        Ok(())
+    }
+
+    /// Reset all metrics
+    pub async fn reset_metrics(&self) -> Result<()> {
+        let mut current = self.current_metrics.write().await;
+        *current = CurrentMetrics {
+            scanner_metrics: ScannerMetrics {
+                objects_per_second: 0.0,
+                bytes_per_second: 0.0,
+                avg_scan_time_per_object: Duration::ZERO,
+                total_objects_scanned: 0,
+                total_bytes_scanned: 0,
+                health_issues_detected: 0,
+                success_rate: 100.0,
+                current_cycle_duration: Duration::ZERO,
+                avg_cycle_duration: Duration::ZERO,
+                last_scan_completion: None,
+            },
+            resource_metrics: ResourceMetrics {
+                cpu_usage_percent: 0.0,
+                memory_usage_bytes: 0,
+                memory_usage_percent: 0.0,
+                disk_io_ops_per_sec: 0.0,
+                disk_io_bytes_per_sec: 0.0,
+                network_io_bytes_per_sec: 0.0,
+                active_threads: 0,
+                open_file_descriptors: 0,
+            },
+            health_metrics: HealthMetrics {
+                issues_by_severity: HashMap::new(),
+                issues_by_type: HashMap::new(),
+                objects_with_issues: 0,
+                objects_with_issues_percent: 0.0,
+                last_health_check: SystemTime::now(),
+                health_score: 100.0,
+            },
+            last_update: SystemTime::now(),
+        };
+        
+        let mut historical = self.historical_data.write().await;
+        historical.clear();
+        
+        Ok(())
+    }
+}
+
+/// Summary of all metrics
+#[derive(Debug, Clone)]
+pub struct MetricsSummary {
+    pub uptime: Duration,
+    pub total_data_points: usize,
+    pub current_scanner_metrics: ScannerMetrics,
+    pub current_resource_metrics: ResourceMetrics,
+    pub current_health_metrics: HealthMetrics,
+    pub avg_objects_per_sec: f64,
+    pub avg_bytes_per_sec: f64,
+    pub avg_cpu_usage: f64,
+    pub avg_memory_usage: f64,
+    pub last_update: SystemTime,
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[tokio::test]
+    async fn test_metrics_collector_creation() {
+        let config = MetricsConfig::default();
+        let collector = MetricsCollector::new(config);
+        let metrics = collector.current_metrics().await;
+        assert_eq!(metrics.scanner_metrics.total_objects_scanned, 0);
+    }
+
+    #[tokio::test]
+    async fn test_metrics_update() {
+        let config = MetricsConfig::default();
+        let collector = MetricsCollector::new(config);
+        
+        let scanner_metrics = ScannerMetrics {
+            objects_per_second: 100.0,
+            bytes_per_second: 1024.0,
+            avg_scan_time_per_object: Duration::from_millis(10),
+            total_objects_scanned: 1000,
+            total_bytes_scanned: 1024000,
+            health_issues_detected: 5,
+            success_rate: 99.5,
+            current_cycle_duration: Duration::from_secs(60),
+            avg_cycle_duration: Duration::from_secs(65),
+            last_scan_completion: Some(SystemTime::now()),
+        };
+        
+        collector.update_scanner_metrics(scanner_metrics).await.unwrap();
+        
+        let current = collector.current_metrics().await;
+        assert_eq!(current.scanner_metrics.total_objects_scanned, 1000);
+        assert_eq!(current.scanner_metrics.health_issues_detected, 5);
+    }
+
+    #[tokio::test]
+    async fn test_health_issue_recording() {
+        let config = MetricsConfig::default();
+        let collector = MetricsCollector::new(config);
+        
+        let issue = HealthIssue {
+            issue_type: HealthIssueType::DiskFull,
+            severity: Severity::High,
+            bucket: "test-bucket".to_string(),
+            object: "test-object".to_string(),
+            description: "Test issue".to_string(),
+            metadata: None,
+        };
+        
+        collector.record_health_issue(&issue).await.unwrap();
+        
+        let current = collector.current_metrics().await;
+        assert_eq!(current.scanner_metrics.health_issues_detected, 1);
+        assert_eq!(current.health_metrics.issues_by_severity.get(&Severity::High), Some(&1));
+    }
+} 

crates/ahm/src/scanner/object_scanner.rs

@@ -0,0 +1,419 @@
+// 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,
+    path::Path,
+    sync::Arc,
+    time::{Duration, SystemTime},
+};
+
+use tokio::sync::RwLock;
+use tracing::info;
+
+use crate::error::Result;
+use super::{HealthIssue, HealthIssueType, Severity};
+
+/// Configuration for object scanning
+#[derive(Debug, Clone)]
+pub struct ObjectScannerConfig {
+    /// Whether to perform checksum verification
+    pub verify_checksum: bool,
+    /// Whether to check replication status
+    pub check_replication: bool,
+    /// Whether to validate metadata consistency
+    pub validate_metadata: bool,
+    /// Maximum object size to scan (bytes)
+    pub max_object_size: u64,
+    /// Minimum object size (bytes)
+    pub min_object_size: u64,
+    /// Timeout for individual object scans
+    pub scan_timeout: Duration,
+    /// Whether to enable deep scanning (bitrot detection)
+    pub enable_deep_scan: bool,
+}
+
+impl Default for ObjectScannerConfig {
+    fn default() -> Self {
+        Self {
+            verify_checksum: true,
+            check_replication: true,
+            validate_metadata: true,
+            max_object_size: 1024 * 1024 * 1024 * 1024, // 1TB
+            min_object_size: 0,
+            scan_timeout: Duration::from_secs(30),
+            enable_deep_scan: false,
+        }
+    }
+}
+
+/// Result of scanning a single object
+#[derive(Debug, Clone)]
+pub struct ObjectScanResult {
+    /// Object identifier
+    pub bucket: String,
+    pub object: String,
+    pub version_id: Option<String>,
+    /// Scan success status
+    pub success: bool,
+    /// Object metadata discovered
+    pub metadata: Option<ObjectMetadata>,
+    /// Health issues detected
+    pub health_issues: Vec<HealthIssue>,
+    /// Time taken to scan this object
+    pub scan_duration: Duration,
+    /// Error message if scan failed
+    pub error_message: Option<String>,
+}
+
+/// Object metadata
+#[derive(Debug, Clone)]
+pub struct ObjectMetadata {
+    pub size: u64,
+    pub modified_time: SystemTime,
+    pub content_type: String,
+    pub etag: String,
+    pub checksum: Option<String>,
+    pub replication_status: Option<String>,
+    pub encryption_status: Option<String>,
+    pub custom_metadata: HashMap<String, String>,
+}
+
+/// Object scanner for individual object health checking
+pub struct ObjectScanner {
+    config: ObjectScannerConfig,
+    statistics: Arc<RwLock<ObjectScannerStatistics>>,
+}
+
+/// Statistics for object scanning
+#[derive(Debug, Clone, Default)]
+pub struct ObjectScannerStatistics {
+    pub objects_scanned: u64,
+    pub objects_with_issues: u64,
+    pub total_issues_found: u64,
+    pub total_scan_time: Duration,
+    pub average_scan_time: Duration,
+    pub checksum_verifications: u64,
+    pub checksum_failures: u64,
+    pub replication_checks: u64,
+    pub replication_failures: u64,
+}
+
+impl ObjectScanner {
+    /// Create a new object scanner
+    pub fn new(config: ObjectScannerConfig) -> Self {
+        Self {
+            config,
+            statistics: Arc::new(RwLock::new(ObjectScannerStatistics::default())),
+        }
+    }
+
+    /// Scan a single object for health issues
+    pub async fn scan_object(&self, bucket: &str, object: &str, version_id: Option<&str>, path: &Path) -> Result<ObjectScanResult> {
+        let scan_start = std::time::Instant::now();
+        let mut health_issues = Vec::new();
+        let mut error_message = None;
+
+        // Check if file exists
+        if !path.exists() {
+            return Ok(ObjectScanResult {
+                bucket: bucket.to_string(),
+                object: object.to_string(),
+                version_id: version_id.map(|v| v.to_string()),
+                success: false,
+                metadata: None,
+                health_issues: vec![HealthIssue {
+                    issue_type: HealthIssueType::MissingReplica,
+                    severity: Severity::Critical,
+                    bucket: bucket.to_string(),
+                    object: object.to_string(),
+                    description: "Object file does not exist".to_string(),
+                    metadata: None,
+                }],
+                scan_duration: scan_start.elapsed(),
+                error_message: Some("Object file not found".to_string()),
+            });
+        }
+
+        // Get file metadata
+        let metadata = match std::fs::metadata(path) {
+            Ok(metadata) => metadata,
+            Err(e) => {
+                error_message = Some(format!("Failed to read file metadata: {}", e));
+                health_issues.push(HealthIssue {
+                    issue_type: HealthIssueType::DiskReadError,
+                    severity: Severity::High,
+                    bucket: bucket.to_string(),
+                    object: object.to_string(),
+                    description: "Failed to read file metadata".to_string(),
+                    metadata: None,
+                });
+                return Ok(ObjectScanResult {
+                    bucket: bucket.to_string(),
+                    object: object.to_string(),
+                    version_id: version_id.map(|v| v.to_string()),
+                    success: false,
+                    metadata: None,
+                    health_issues,
+                    scan_duration: scan_start.elapsed(),
+                    error_message,
+                });
+            }
+        };
+
+        // Check file size
+        let file_size = metadata.len();
+        if file_size < self.config.min_object_size {
+            health_issues.push(HealthIssue {
+                issue_type: HealthIssueType::ObjectTooSmall,
+                severity: Severity::Low,
+                bucket: bucket.to_string(),
+                object: object.to_string(),
+                description: format!("Object size {} is below minimum {}", file_size, self.config.min_object_size),
+                metadata: None,
+            });
+        }
+
+        if file_size > self.config.max_object_size {
+            health_issues.push(HealthIssue {
+                issue_type: HealthIssueType::ObjectTooLarge,
+                severity: Severity::Medium,
+                bucket: bucket.to_string(),
+                object: object.to_string(),
+                description: format!("Object size {} exceeds maximum {}", file_size, self.config.max_object_size),
+                metadata: None,
+            });
+        }
+
+        // Verify checksum if enabled
+        let checksum = if self.config.verify_checksum {
+            match self.verify_checksum(path).await {
+                Ok(cs) => {
+                    self.update_statistics(|stats| stats.checksum_verifications += 1).await;
+                    Some(cs)
+                }
+                Err(_e) => {
+                    self.update_statistics(|stats| stats.checksum_failures += 1).await;
+                    health_issues.push(HealthIssue {
+                        issue_type: HealthIssueType::ChecksumMismatch,
+                        severity: Severity::High,
+                        bucket: bucket.to_string(),
+                        object: object.to_string(),
+                        description: "Checksum verification failed".to_string(),
+                        metadata: None,
+                    });
+                    None
+                }
+            }
+        } else {
+            None
+        };
+
+        // Check replication status if enabled
+        let replication_status = if self.config.check_replication {
+            match self.check_replication_status(bucket, object).await {
+                Ok(status) => {
+                    self.update_statistics(|stats| stats.replication_checks += 1).await;
+                    Some(status)
+                }
+                Err(_e) => {
+                    self.update_statistics(|stats| stats.replication_failures += 1).await;
+                    health_issues.push(HealthIssue {
+                        issue_type: HealthIssueType::MissingReplica,
+                        severity: Severity::High,
+                        bucket: bucket.to_string(),
+                        object: object.to_string(),
+                        description: "Replication status check failed".to_string(),
+                        metadata: None,
+                    });
+                    None
+                }
+            }
+        } else {
+            None
+        };
+
+        // Validate metadata if enabled
+        if self.config.validate_metadata {
+            if let Some(issue) = self.validate_metadata(bucket, object, &metadata).await? {
+                health_issues.push(issue);
+            }
+        }
+
+        // Create object metadata
+        let object_metadata = ObjectMetadata {
+            size: file_size,
+            modified_time: metadata.modified().unwrap_or(SystemTime::now()),
+            content_type: self.detect_content_type(path),
+            etag: self.calculate_etag(path).await?,
+            checksum,
+            replication_status,
+            encryption_status: None, // TODO: Implement encryption status check
+            custom_metadata: HashMap::new(), // TODO: Extract custom metadata
+        };
+
+        let scan_duration = scan_start.elapsed();
+        let success = health_issues.is_empty();
+
+        // Update statistics
+        self.update_statistics(|stats| {
+            stats.objects_scanned += 1;
+            if !health_issues.is_empty() {
+                stats.objects_with_issues += 1;
+                stats.total_issues_found += health_issues.len() as u64;
+            }
+            stats.total_scan_time += scan_duration;
+            stats.average_scan_time = Duration::from_millis(
+                stats.total_scan_time.as_millis() as u64 / stats.objects_scanned.max(1)
+            );
+        }).await;
+
+        Ok(ObjectScanResult {
+            bucket: bucket.to_string(),
+            object: object.to_string(),
+            version_id: version_id.map(|v| v.to_string()),
+            success,
+            metadata: Some(object_metadata),
+            health_issues,
+            scan_duration,
+            error_message,
+        })
+    }
+
+    /// Verify object checksum
+    async fn verify_checksum(&self, _path: &Path) -> Result<String> {
+        // TODO: Implement actual checksum verification
+        // For now, return a placeholder checksum
+        Ok("placeholder_checksum".to_string())
+    }
+
+    /// Check object replication status
+    async fn check_replication_status(&self, _bucket: &str, _object: &str) -> Result<String> {
+        // TODO: Implement actual replication status checking
+        // For now, return a placeholder status
+        Ok("replicated".to_string())
+    }
+
+    /// Validate object metadata
+    async fn validate_metadata(&self, _bucket: &str, _object: &str, _metadata: &std::fs::Metadata) -> Result<Option<HealthIssue>> {
+        // TODO: Implement actual metadata validation
+        // For now, return None (no issues)
+        Ok(None)
+    }
+
+    /// Detect content type from file extension
+    fn detect_content_type(&self, path: &Path) -> String {
+        if let Some(extension) = path.extension() {
+            match extension.to_str().unwrap_or("").to_lowercase().as_str() {
+                "txt" => "text/plain",
+                "json" => "application/json",
+                "xml" => "application/xml",
+                "html" | "htm" => "text/html",
+                "css" => "text/css",
+                "js" => "application/javascript",
+                "png" => "image/png",
+                "jpg" | "jpeg" => "image/jpeg",
+                "gif" => "image/gif",
+                "pdf" => "application/pdf",
+                "zip" => "application/zip",
+                "tar" => "application/x-tar",
+                "gz" => "application/gzip",
+                _ => "application/octet-stream",
+            }.to_string()
+        } else {
+            "application/octet-stream".to_string()
+        }
+    }
+
+    /// Calculate object ETag
+    async fn calculate_etag(&self, _path: &Path) -> Result<String> {
+        // TODO: Implement actual ETag calculation
+        // For now, return a placeholder ETag
+        Ok("placeholder_etag".to_string())
+    }
+
+    /// Update scanner statistics
+    async fn update_statistics<F>(&self, update_fn: F)
+    where
+        F: FnOnce(&mut ObjectScannerStatistics),
+    {
+        let mut stats = self.statistics.write().await;
+        update_fn(&mut stats);
+    }
+
+    /// Get current statistics
+    pub async fn statistics(&self) -> ObjectScannerStatistics {
+        self.statistics.read().await.clone()
+    }
+
+    /// Reset statistics
+    pub async fn reset_statistics(&self) {
+        let mut stats = self.statistics.write().await;
+        *stats = ObjectScannerStatistics::default();
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use tempfile::TempDir;
+    use std::fs::File;
+    use std::io::Write;
+
+    #[tokio::test]
+    async fn test_object_scanner_creation() {
+        let config = ObjectScannerConfig::default();
+        let scanner = ObjectScanner::new(config);
+        assert_eq!(scanner.statistics().await.objects_scanned, 0);
+    }
+
+    #[tokio::test]
+    async fn test_content_type_detection() {
+        let config = ObjectScannerConfig::default();
+        let scanner = ObjectScanner::new(config);
+        
+        let path = Path::new("test.txt");
+        assert_eq!(scanner.detect_content_type(path), "text/plain");
+        
+        let path = Path::new("test.json");
+        assert_eq!(scanner.detect_content_type(path), "application/json");
+        
+        let path = Path::new("test.unknown");
+        assert_eq!(scanner.detect_content_type(path), "application/octet-stream");
+    }
+
+    #[tokio::test]
+    async fn test_object_scanning() {
+        let temp_dir = TempDir::new().unwrap();
+        let test_file = temp_dir.path().join("test.txt");
+        
+        // Create a test file
+        let mut file = File::create(&test_file).unwrap();
+        writeln!(file, "test content").unwrap();
+        
+        let config = ObjectScannerConfig::default();
+        let scanner = ObjectScanner::new(config);
+        
+        let result = scanner.scan_object("test-bucket", "test.txt", None, &test_file).await.unwrap();
+        
+        assert!(result.success);
+        assert_eq!(result.bucket, "test-bucket");
+        assert_eq!(result.object, "test.txt");
+        assert!(result.metadata.is_some());
+        
+        let metadata = result.metadata.unwrap();
+        assert!(metadata.size > 0);
+        assert_eq!(metadata.content_type, "text/plain");
+    }
+}