mirror of
https://github.com/rustfs/rustfs.git
synced 2026-07-27 08:38:58 +00:00
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>
This commit is contained in:
dandan
@@ -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::{
|
||||
|
||||
@@ -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<SystemTime>,
|
||||
/// Current scan cycle
|
||||
pub current_cycle: u64,
|
||||
/// Total scan cycles completed
|
||||
pub total_cycles: u64,
|
||||
/// Current scan duration
|
||||
pub current_scan_duration: Option<Duration>,
|
||||
/// 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<String, BucketMetrics>,
|
||||
/// Disk metrics
|
||||
pub disk_metrics: HashMap<String, DiskMetrics>,
|
||||
}
|
||||
|
||||
/// 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<SystemTime>,
|
||||
/// Scan duration
|
||||
pub scan_duration: Option<Duration>,
|
||||
/// 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<SystemTime>,
|
||||
/// 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);
|
||||
}
|
||||
}
|
||||
@@ -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<SystemTime>,
|
||||
/// Last scan end time
|
||||
pub last_scan_end: Option<SystemTime>,
|
||||
/// Current scan duration
|
||||
pub current_scan_duration: Option<Duration>,
|
||||
/// Last deep scan time
|
||||
pub last_deep_scan_time: Option<SystemTime>,
|
||||
/// Buckets being scanned
|
||||
pub scanning_buckets: Vec<String>,
|
||||
/// Disks being scanned
|
||||
pub scanning_disks: Vec<String>,
|
||||
}
|
||||
|
||||
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<RwLock<ScannerConfig>>,
|
||||
/// Scanner state
|
||||
state: Arc<RwLock<ScannerState>>,
|
||||
/// Metrics collector
|
||||
metrics: Arc<MetricsCollector>,
|
||||
/// Bucket metrics cache
|
||||
bucket_metrics: Arc<RwLock<HashMap<String, BucketMetrics>>>,
|
||||
/// Disk metrics cache
|
||||
disk_metrics: Arc<RwLock<HashMap<String, DiskMetrics>>>,
|
||||
/// EC Set disks - represents a complete erasure coding set
|
||||
set_disks: Arc<SetDisks>,
|
||||
/// Cancellation token for graceful shutdown
|
||||
cancel_token: CancellationToken,
|
||||
}
|
||||
|
||||
impl Scanner {
|
||||
/// Create a new scanner
|
||||
pub fn new(set_disks: Arc<SetDisks>, config: Option<ScannerConfig>) -> 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<HashMap<String, HashMap<String, rustfs_filemeta::FileMeta>>> {
|
||||
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<HashMap<String, rustfs_filemeta::FileMeta>> {
|
||||
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<String, HashMap<String, rustfs_filemeta::FileMeta>>],
|
||||
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<object_name>
|
||||
let mut object_locations = HashMap::new(); // (bucket, object) -> Vec<disk_index>
|
||||
|
||||
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::<usize>());
|
||||
|
||||
// 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<usize> = (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);
|
||||
// }
|
||||
}
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user