From 2832f0e089f10157fd99e542032c3c17ebd4f351 Mon Sep 17 00:00:00 2001 From: guojidan <63799833+guojidan@users.noreply.github.com> Date: Thu, 10 Jul 2025 17:10:44 +0800 Subject: [PATCH] Scanner (#156) * feat: integrate CancellationToken for unified background services management - Consolidate data scanner and auto heal cancellation tokens into single unified token - Move GLOBAL_BACKGROUND_SERVICES_CANCEL_TOKEN to global.rs for centralized management - Add graceful shutdown support to MRF heal routine with MinIO-compatible logic - Implement heal_routine_with_cancel method preserving original healing logic - Update main.rs to use unified background services shutdown mechanism - Enhance error handling with proper ecstore Result types - Fix clippy warnings for needless return statements - Maintain backward compatibility while adding modern cancellation support This change provides a cleaner architecture for background service lifecycle management and ensures all healing services can be gracefully shut down through a single token. Signed-off-by: junxiang Mu <1948535941@qq.com> * fix: Refact heal and scanner design Signed-off-by: junxiang Mu <1948535941@qq.com> * refact: step 2 Signed-off-by: junxiang Mu <1948535941@qq.com> * 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> * Chore: fix ref and fix comment Signed-off-by: junxiang Mu <1948535941@qq.com> * fix: fix clippy Signed-off-by: junxiang Mu <1948535941@qq.com> --------- Signed-off-by: junxiang Mu <1948535941@qq.com> Co-authored-by: dandan --- Cargo.lock | 29 + Cargo.toml | 3 + crates/ahm/Cargo.toml | 35 + crates/ahm/src/error.rs | 45 + crates/ahm/src/lib.rs | 54 + crates/ahm/src/scanner/data_scanner.rs | 1247 +++++++++++++++++ crates/ahm/src/scanner/data_usage.rs | 671 +++++++++ crates/ahm/src/scanner/histogram.rs | 277 ++++ crates/ahm/src/scanner/metrics.rs | 284 ++++ crates/ahm/src/scanner/mod.rs | 25 + crates/ecstore/src/global.rs | 28 + .../ecstore/src/heal/background_heal_ops.rs | 110 +- crates/ecstore/src/heal/data_scanner.rs | 29 +- crates/ecstore/src/heal/mrf.rs | 117 +- rustfs/Cargo.toml | 1 + rustfs/src/admin/handlers.rs | 1 + rustfs/src/main.rs | 45 +- 17 files changed, 2885 insertions(+), 116 deletions(-) create mode 100644 crates/ahm/Cargo.toml create mode 100644 crates/ahm/src/error.rs create mode 100644 crates/ahm/src/lib.rs create mode 100644 crates/ahm/src/scanner/data_scanner.rs create mode 100644 crates/ahm/src/scanner/data_usage.rs create mode 100644 crates/ahm/src/scanner/histogram.rs create mode 100644 crates/ahm/src/scanner/metrics.rs create mode 100644 crates/ahm/src/scanner/mod.rs diff --git a/Cargo.lock b/Cargo.lock index b4ee6da36..e1909f7fb 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7857,6 +7857,7 @@ dependencies = [ "pin-project-lite", "reqwest", "rust-embed", + "rustfs-ahm", "rustfs-appauth", "rustfs-common", "rustfs-config", @@ -7897,6 +7898,34 @@ dependencies = [ "zip", ] +[[package]] +name = "rustfs-ahm" +version = "0.0.3" +dependencies = [ + "anyhow", + "async-trait", + "bytes", + "futures", + "lazy_static", + "rmp-serde", + "rustfs-common", + "rustfs-ecstore", + "rustfs-filemeta", + "rustfs-lock", + "rustfs-madmin", + "rustfs-utils", + "serde", + "serde_json", + "thiserror 2.0.12", + "time", + "tokio", + "tokio-test", + "tokio-util", + "tracing", + "url", + "uuid", +] + [[package]] name = "rustfs-appauth" version = "0.0.5" diff --git a/Cargo.toml b/Cargo.toml index 2d4741fc4..b0b33a742 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -36,6 +36,7 @@ members = [ "crates/utils", # Utility functions and helpers "crates/workers", # Worker thread pools and task scheduling "crates/zip", # ZIP file handling and compression + "crates/ahm", ] resolver = "2" @@ -62,6 +63,7 @@ rustfs-filemeta = { path = "crates/filemeta" } rustfs-rio = { path = "crates/rio" } [workspace.dependencies] +rustfs-ahm = { path = "crates/ahm", version = "0.0.3" } rustfs-s3select-api = { path = "crates/s3select-api", version = "0.0.5" } rustfs-appauth = { path = "crates/appauth", version = "0.0.5" } rustfs-common = { path = "crates/common", version = "0.0.5" } @@ -261,6 +263,7 @@ winapi = { version = "0.3.9" } xxhash-rust = { version = "0.8.15", features = ["xxh64", "xxh3"] } zip = "2.4.2" zstd = "0.13.3" +anyhow = "1.0.86" [profile.wasm-dev] inherits = "dev" diff --git a/crates/ahm/Cargo.toml b/crates/ahm/Cargo.toml new file mode 100644 index 000000000..eb2964123 --- /dev/null +++ b/crates/ahm/Cargo.toml @@ -0,0 +1,35 @@ +[package] +name = "rustfs-ahm" +version = "0.0.3" +edition = "2021" +authors = ["RustFS Team"] +license = "Apache-2.0" +description = "RustFS AHM (Automatic Health Management) Scanner" + +[dependencies] +rustfs-ecstore = { workspace = true } +rustfs-common = { workspace = true } +rustfs-filemeta = { workspace = true } +rustfs-madmin = { workspace = true } +rustfs-utils = { workspace = true } +tokio = { workspace = true, features = ["full"] } +tokio-util = { workspace = true } +tracing = { workspace = true } +serde = { workspace = true, features = ["derive"] } +serde_json = { workspace = true } +thiserror = { workspace = true } +bytes = { workspace = true } +time = { workspace = true, features = ["serde"] } +uuid = { workspace = true, features = ["v4", "serde"] } +anyhow = { workspace = true } +async-trait = { workspace = true } +futures = { workspace = true } +url = { workspace = true } +rustfs-lock = { workspace = true } + +lazy_static = { workspace = true } + +[dev-dependencies] +rmp-serde = { workspace = true } +tokio-test = "0.4" +serde_json = "1.0" diff --git a/crates/ahm/src/error.rs b/crates/ahm/src/error.rs new file mode 100644 index 000000000..894638970 --- /dev/null +++ b/crates/ahm/src/error.rs @@ -0,0 +1,45 @@ +// 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 thiserror::Error; + +#[derive(Debug, Error)] +pub enum Error { + #[error("I/O error: {0}")] + Io(#[from] std::io::Error), + + #[error("Storage error: {0}")] + Storage(#[from] rustfs_ecstore::error::Error), + + #[error("Configuration error: {0}")] + Config(String), + + #[error("Scanner error: {0}")] + Scanner(String), + + #[error("Metrics error: {0}")] + Metrics(String), + + #[error(transparent)] + Other(#[from] anyhow::Error), +} + +pub type Result = std::result::Result; + +// Implement conversion from ahm::Error to std::io::Error for use in main.rs +impl From for std::io::Error { + fn from(err: Error) -> Self { + std::io::Error::other(err) + } +} diff --git a/crates/ahm/src/lib.rs b/crates/ahm/src/lib.rs new file mode 100644 index 000000000..c3caed929 --- /dev/null +++ b/crates/ahm/src/lib.rs @@ -0,0 +1,54 @@ +// 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::OnceLock; +use tokio_util::sync::CancellationToken; + +pub mod error; +pub mod scanner; + +pub use error::{Error, Result}; +pub use scanner::{ + load_data_usage_from_backend, store_data_usage_in_backend, BucketTargetUsageInfo, BucketUsageInfo, DataUsageInfo, Scanner, + ScannerMetrics, +}; + +// Global cancellation token for AHM services (scanner and other background tasks) +static GLOBAL_AHM_SERVICES_CANCEL_TOKEN: OnceLock = OnceLock::new(); + +/// Initialize the global AHM services cancellation token +pub fn init_ahm_services_cancel_token(cancel_token: CancellationToken) -> Result<()> { + GLOBAL_AHM_SERVICES_CANCEL_TOKEN + .set(cancel_token) + .map_err(|_| Error::Config("AHM services cancel token already initialized".to_string())) +} + +/// Get the global AHM services cancellation token +pub fn get_ahm_services_cancel_token() -> Option<&'static CancellationToken> { + GLOBAL_AHM_SERVICES_CANCEL_TOKEN.get() +} + +/// Create and initialize the global AHM services cancellation token +pub fn create_ahm_services_cancel_token() -> CancellationToken { + let cancel_token = CancellationToken::new(); + init_ahm_services_cancel_token(cancel_token.clone()).expect("AHM services cancel token already initialized"); + cancel_token +} + +/// Shutdown all AHM services gracefully +pub fn shutdown_ahm_services() { + if let Some(cancel_token) = GLOBAL_AHM_SERVICES_CANCEL_TOKEN.get() { + cancel_token.cancel(); + } +} diff --git a/crates/ahm/src/scanner/data_scanner.rs b/crates/ahm/src/scanner/data_scanner.rs new file mode 100644 index 000000000..e4d77a995 --- /dev/null +++ b/crates/ahm/src/scanner/data_scanner.rs @@ -0,0 +1,1247 @@ +// Copyright 2024 RustFS Team +// +// Licensed under the Apache License, Version 2.0 (the "License"); +// you may not use this file except in compliance with the License. +// You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, software +// distributed under the License is distributed on an "AS IS" BASIS, +// WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +// See the License for the specific language governing permissions and +// limitations under the License. + +use std::{ + collections::HashMap, + sync::Arc, + time::{Duration, SystemTime}, +}; + +use ecstore::{ + disk::{DiskAPI, DiskStore, WalkDirOptions}, + set_disk::SetDisks, +}; +use rustfs_ecstore as ecstore; +use rustfs_filemeta::MetacacheReader; +use tokio::sync::{Mutex, RwLock}; +use tokio_util::sync::CancellationToken; +use tracing::{debug, error, info, warn}; + +use super::{ + data_usage::DataUsageInfo, + metrics::{BucketMetrics, DiskMetrics, MetricsCollector, ScannerMetrics}, +}; +use crate::{ + error::{Error, Result}, + get_ahm_services_cancel_token, +}; + +use rustfs_ecstore::disk::RUSTFS_META_BUCKET; + +/// Custom scan mode enum for AHM scanner +#[derive(Debug, Clone, Copy, PartialEq, Eq, Default)] +pub enum ScanMode { + /// Normal scan - basic object discovery and metadata collection + #[default] + Normal, + /// Deep scan - includes EC verification and integrity checks + Deep, +} + +/// Scanner configuration +#[derive(Debug, Clone)] +pub struct ScannerConfig { + /// Scan interval between cycles + pub scan_interval: Duration, + /// Deep scan interval (how often to perform deep scan) + pub deep_scan_interval: Duration, + /// Maximum concurrent scans + pub max_concurrent_scans: usize, + /// Whether to enable healing + pub enable_healing: bool, + /// Whether to enable metrics collection + pub enable_metrics: bool, + /// Current scan mode (normal, deep) + pub scan_mode: ScanMode, + /// Whether to enable data usage statistics collection + pub enable_data_usage_stats: bool, +} + +impl Default for ScannerConfig { + fn default() -> Self { + Self { + scan_interval: Duration::from_secs(60), // 1 minute + deep_scan_interval: Duration::from_secs(3600), // 1 hour + max_concurrent_scans: 20, + enable_healing: true, + enable_metrics: true, + scan_mode: ScanMode::Normal, + enable_data_usage_stats: true, + } + } +} + +/// Scanner state +#[derive(Debug, Default)] +pub struct ScannerState { + /// Whether scanner is running + pub is_running: bool, + /// Current scan cycle + pub current_cycle: u64, + /// Last scan start time + pub last_scan_start: Option, + /// Last scan end time + pub last_scan_end: Option, + /// Current scan duration + pub current_scan_duration: Option, + /// Last deep scan time + pub last_deep_scan_time: Option, + /// Buckets being scanned + pub scanning_buckets: Vec, + /// Disks being scanned + pub scanning_disks: Vec, +} + +/// AHM Scanner - Automatic Health Management Scanner +/// +/// This scanner monitors the health of objects in the RustFS storage system. +/// It integrates with ECStore to perform real data scanning across all EC sets +/// and collects metrics. +/// +/// The scanner operates on the entire ECStore, scanning all EC (Erasure Coding) sets, +/// where each set contains multiple disks that store the same objects with different shards. +pub struct Scanner { + /// Scanner configuration + config: Arc>, + /// Scanner state + state: Arc>, + /// Metrics collector + metrics: Arc, + /// Bucket metrics cache + bucket_metrics: Arc>>, + /// Disk metrics cache + disk_metrics: Arc>>, + /// Data usage statistics cache + data_usage_stats: Arc>>, + /// Last data usage statistics collection time + last_data_usage_collection: Arc>>, +} + +impl Scanner { + /// Create a new scanner + pub fn new(config: Option) -> Self { + let config = config.unwrap_or_default(); + + info!("Creating AHM scanner for all EC sets"); + + Self { + config: Arc::new(RwLock::new(config)), + state: Arc::new(RwLock::new(ScannerState::default())), + metrics: Arc::new(MetricsCollector::new()), + bucket_metrics: Arc::new(Mutex::new(HashMap::new())), + disk_metrics: Arc::new(Mutex::new(HashMap::new())), + data_usage_stats: Arc::new(Mutex::new(HashMap::new())), + last_data_usage_collection: Arc::new(RwLock::new(None)), + } + } + + /// Start the scanner + pub async fn start(&self) -> Result<()> { + let mut state = self.state.write().await; + + if state.is_running { + warn!("Scanner is already running"); + return Ok(()); + } + + state.is_running = true; + state.last_scan_start = Some(SystemTime::now()); + + info!("Starting AHM scanner"); + + // Start background scan loop + let scanner = self.clone_for_background(); + tokio::spawn(async move { + if let Err(e) = scanner.scan_loop().await { + error!("Scanner loop failed: {}", e); + } + }); + + Ok(()) + } + + /// Stop the scanner gracefully + pub async fn stop(&self) -> Result<()> { + let mut state = self.state.write().await; + + if !state.is_running { + warn!("Scanner is not running"); + return Ok(()); + } + + info!("Stopping AHM scanner gracefully..."); + + // Trigger cancellation using global cancel token + if let Some(cancel_token) = get_ahm_services_cancel_token() { + cancel_token.cancel(); + } + + state.is_running = false; + state.last_scan_end = Some(SystemTime::now()); + + if let Some(start_time) = state.last_scan_start { + state.current_scan_duration = Some(SystemTime::now().duration_since(start_time).unwrap_or(Duration::ZERO)); + } + + info!("AHM scanner stopped"); + Ok(()) + } + + /// Get integrated data usage statistics for DataUsageInfoHandler + pub async fn get_data_usage_info(&self) -> Result { + let mut integrated_info = DataUsageInfo::new(); + + // Collect data from all buckets + { + let data_usage_guard = self.data_usage_stats.lock().await; + for (bucket_name, bucket_data) in data_usage_guard.iter() { + let _bucket_name = bucket_name; + + // Merge bucket data into integrated info + integrated_info.merge(bucket_data); + } + } + + // Update capacity information from storage info + if let Some(ecstore) = rustfs_ecstore::new_object_layer_fn() { + let mut total_capacity = 0u64; + let mut total_used_capacity = 0u64; + let mut total_free_capacity = 0u64; + + // Collect capacity info from all SetDisks + for pool in &ecstore.pools { + for set_disks in &pool.disk_set { + let (disks, _) = set_disks.get_online_disks_with_healing(false).await; + for disk in disks { + if let Ok(disk_info) = disk + .disk_info(&ecstore::disk::DiskInfoOptions { + disk_id: disk.path().to_string_lossy().to_string(), + metrics: true, + noop: false, + }) + .await + { + total_capacity += disk_info.total; + total_used_capacity += disk_info.used; + total_free_capacity += disk_info.free; + } + } + } + } + + if total_capacity > 0 { + integrated_info.update_capacity(total_capacity, total_used_capacity, total_free_capacity); + } + } + + Ok(integrated_info) + } + + /// Get current scanner metrics + pub async fn get_metrics(&self) -> ScannerMetrics { + let mut metrics = self.metrics.get_metrics(); + + // Add bucket metrics + let bucket_metrics: HashMap = { + let bucket_metrics_guard = self.bucket_metrics.lock().await; + bucket_metrics_guard + .iter() + .map(|(key, value)| (key.clone(), value.clone())) + .collect() + }; + metrics.bucket_metrics = bucket_metrics; + + // Add disk metrics + let disk_metrics: HashMap = { + let disk_metrics_guard = self.disk_metrics.lock().await; + disk_metrics_guard + .iter() + .map(|(key, value)| (key.clone(), value.clone())) + .collect() + }; + metrics.disk_metrics = disk_metrics; + + // Add current scan duration + let state = self.state.read().await; + metrics.current_scan_duration = state.current_scan_duration; + + metrics + } + + /// Perform a single scan cycle + pub async fn scan_cycle(&self) -> Result<()> { + let start_time = SystemTime::now(); + + info!("Starting scan cycle {} for all EC sets", self.metrics.get_metrics().current_cycle + 1); + + // Update state + { + let mut state = self.state.write().await; + state.current_cycle += 1; + state.last_scan_start = Some(start_time); + state.scanning_buckets.clear(); + state.scanning_disks.clear(); + } + + self.metrics.set_current_cycle(self.state.read().await.current_cycle); + self.metrics.increment_total_cycles(); + + // Get ECStore and all SetDisks + let Some(ecstore) = rustfs_ecstore::new_object_layer_fn() else { + warn!("No ECStore available for scanning"); + return Ok(()); + }; + + // Get all SetDisks from all pools + let mut all_set_disks = Vec::new(); + for pool in &ecstore.pools { + for set_disks in &pool.disk_set { + all_set_disks.push(set_disks.clone()); + } + } + + if all_set_disks.is_empty() { + warn!("No EC sets available for scanning"); + return Ok(()); + } + + info!("Scanning {} EC sets across {} pools", all_set_disks.len(), ecstore.pools.len()); + + // Phase 1: Scan all SetDisks concurrently + let config = self.config.read().await; + let semaphore = Arc::new(tokio::sync::Semaphore::new(config.max_concurrent_scans)); + drop(config); + let mut scan_futures = Vec::new(); + + for set_disks in all_set_disks { + let semaphore = semaphore.clone(); + let scanner = self.clone_for_background(); + + let future = async move { + let _permit = semaphore.acquire().await.unwrap(); + scanner.scan_set_disks(set_disks).await + }; + + scan_futures.push(future); + } + + // Wait for all scans to complete + let mut results = Vec::new(); + for future in scan_futures { + results.push(future.await); + } + + // Check results and collect object metadata + let mut successful_scans = 0; + let mut failed_scans = 0; + let mut all_disk_objects = Vec::new(); + + for result in results { + match result { + Ok(disk_objects) => { + successful_scans += 1; + all_disk_objects.extend(disk_objects); + } + Err(e) => { + failed_scans += 1; + error!("SetDisks scan failed: {}", e); + } + } + } + + // Phase 2: Analyze object distribution and perform EC verification + if successful_scans > 0 { + // Get all disks from all SetDisks for analysis + let mut all_disks = Vec::new(); + for pool in &ecstore.pools { + for set_disks in &pool.disk_set { + let (disks, _) = set_disks.get_online_disks_with_healing(false).await; + all_disks.extend(disks); + } + } + + if let Err(e) = self.analyze_object_distribution(&all_disk_objects, &all_disks).await { + error!("Object distribution analysis failed: {}", e); + } + } + + // Update scan duration + let scan_duration = SystemTime::now().duration_since(start_time).unwrap_or(Duration::ZERO); + + { + let mut state = self.state.write().await; + state.last_scan_end = Some(SystemTime::now()); + state.current_scan_duration = Some(scan_duration); + } + + info!( + "Completed scan cycle in {:?} ({} successful, {} failed)", + scan_duration, successful_scans, failed_scans + ); + Ok(()) + } + + /// Scan a single SetDisks (EC set) + async fn scan_set_disks( + &self, + set_disks: Arc, + ) -> Result>>> { + let set_index = set_disks.set_index; + let pool_index = set_disks.pool_index; + + info!("Scanning EC set {} in pool {}", set_index, pool_index); + + // Get online disks from this EC set + let (disks, _) = set_disks.get_online_disks_with_healing(false).await; + + if disks.is_empty() { + warn!("No online disks available for EC set {} in pool {}", set_index, pool_index); + return Ok(Vec::new()); + } + + info!("Scanning {} online disks in EC set {} (pool {})", disks.len(), set_index, pool_index); + + // Scan all disks in this SetDisks concurrently + let config = self.config.read().await; + let semaphore = Arc::new(tokio::sync::Semaphore::new(config.max_concurrent_scans)); + drop(config); + let mut scan_futures = Vec::new(); + + for disk in disks { + let semaphore = semaphore.clone(); + let scanner = self.clone_for_background(); + + let future = async move { + let _permit = semaphore.acquire().await.unwrap(); + scanner.scan_disk(&disk).await + }; + + scan_futures.push(future); + } + + // Wait for all scans to complete + let mut results = Vec::new(); + for future in scan_futures { + results.push(future.await); + } + + // Check results and collect object metadata + let mut successful_scans = 0; + let mut failed_scans = 0; + let mut all_disk_objects = Vec::new(); + + for result in results { + match result { + Ok(disk_objects) => { + successful_scans += 1; + all_disk_objects.push(disk_objects); + } + Err(e) => { + failed_scans += 1; + error!("Disk scan failed in EC set {} (pool {}): {}", set_index, pool_index, e); + // Add empty map for failed disk + all_disk_objects.push(HashMap::new()); + } + } + } + + info!( + "Completed scanning EC set {} (pool {}): {} successful, {} failed", + set_index, pool_index, successful_scans, failed_scans + ); + + Ok(all_disk_objects) + } + + /// Scan a single disk + async fn scan_disk(&self, disk: &DiskStore) -> Result>> { + let disk_path = disk.path().to_string_lossy().to_string(); + + info!("Scanning disk: {}", disk_path); + + // Update disk metrics + { + let mut disk_metrics_guard = self.disk_metrics.lock().await; + let metrics = disk_metrics_guard.entry(disk_path.clone()).or_insert_with(|| DiskMetrics { + disk_path: disk_path.clone(), + ..Default::default() + }); + + metrics.is_scanning = true; + metrics.last_scan_time = Some(SystemTime::now()); + + // Get disk info using DiskStore's disk_info interface + if let Ok(disk_info) = disk + .disk_info(&ecstore::disk::DiskInfoOptions { + disk_id: disk_path.clone(), + metrics: true, + noop: false, + }) + .await + { + metrics.total_space = disk_info.total; + metrics.used_space = disk_info.used; + metrics.free_space = disk_info.free; + metrics.is_online = disk.is_online().await; + + // Additional disk info for debugging + debug!( + "Disk {}: total={}, used={}, free={}, online={}", + disk_path, disk_info.total, disk_info.used, disk_info.free, metrics.is_online + ); + } + } + + // Update state + { + let mut state = self.state.write().await; + state.scanning_disks.push(disk_path.clone()); + } + + // List volumes (buckets) on this disk + let volumes = match disk.list_volumes().await { + Ok(volumes) => volumes, + Err(e) => { + error!("Failed to list volumes on disk {}: {}", disk_path, e); + return Err(Error::Storage(e.into())); + } + }; + + // Scan each volume and collect object metadata + let mut disk_objects = HashMap::new(); + for volume in volumes { + // check cancel token + if let Some(cancel_token) = get_ahm_services_cancel_token() { + if cancel_token.is_cancelled() { + info!("Cancellation requested, stopping disk scan"); + break; + } + } + + match self.scan_volume(disk, &volume.name).await { + Ok(object_metadata) => { + disk_objects.insert(volume.name, object_metadata); + } + Err(e) => { + error!("Failed to scan volume {} on disk {}: {}", volume.name, disk_path, e); + continue; + } + } + } + + // Update disk metrics after scan + { + let mut disk_metrics_guard = self.disk_metrics.lock().await; + if let Some(existing_metrics) = disk_metrics_guard.get(&disk_path) { + let mut updated_metrics = existing_metrics.clone(); + updated_metrics.is_scanning = false; + disk_metrics_guard.insert(disk_path.clone(), updated_metrics); + } + } + + // Update state + { + let mut state = self.state.write().await; + state.scanning_disks.retain(|d| d != &disk_path); + } + + Ok(disk_objects) + } + + /// Scan a single volume (bucket) and collect object information + /// + /// This method collects all objects from a disk for a specific bucket. + /// It returns a map of object names to their metadata for later analysis. + async fn scan_volume(&self, disk: &DiskStore, bucket: &str) -> Result> { + info!("Scanning bucket: {} on disk: {}", bucket, disk.to_string()); + + // Initialize bucket metrics if not exists + { + let mut bucket_metrics_guard = self.bucket_metrics.lock().await; + bucket_metrics_guard + .entry(bucket.to_string()) + .or_insert_with(|| BucketMetrics { + bucket: bucket.to_string(), + ..Default::default() + }); + } + + // Update state + { + let mut state = self.state.write().await; + state.scanning_buckets.push(bucket.to_string()); + } + + self.metrics.increment_bucket_scans_started(1); + + let scan_start = SystemTime::now(); + + // Walk through all objects in the bucket + let walk_opts = WalkDirOptions { + bucket: bucket.to_string(), + base_dir: String::new(), + recursive: true, + report_notfound: false, + filter_prefix: None, + forward_to: None, + limit: 0, + disk_id: String::new(), + }; + + // Use a buffer to collect scan results for processing + let mut scan_buffer = Vec::new(); + + if let Err(e) = disk.walk_dir(walk_opts, &mut scan_buffer).await { + error!("Failed to walk directory for bucket {}: {}", bucket, e); + return Err(Error::Storage(e.into())); + } + + // Process the scan results using MetacacheReader + let mut reader = MetacacheReader::new(std::io::Cursor::new(scan_buffer)); + let mut objects_scanned = 0u64; + let mut objects_with_issues = 0u64; + let mut object_metadata = HashMap::new(); + + // Process each object entry + while let Ok(Some(mut entry)) = reader.peek().await { + objects_scanned += 1; + // Check if this is an actual object (not just a directory) + if entry.is_object() { + debug!("Scanned object: {}", entry.name); + + // Parse object metadata + if let Ok(file_meta) = entry.xl_meta() { + if file_meta.versions.is_empty() { + objects_with_issues += 1; + warn!("Object {} has no versions", entry.name); + } else { + // Store object metadata for later analysis + object_metadata.insert(entry.name.clone(), file_meta.clone()); + } + } else { + objects_with_issues += 1; + warn!("Failed to parse metadata for object {}", entry.name); + } + } + } + + // Update metrics + self.metrics.increment_objects_scanned(objects_scanned); + self.metrics.increment_objects_with_issues(objects_with_issues); + self.metrics.increment_bucket_scans_finished(1); + + // Update bucket metrics + { + let mut bucket_metrics_guard = self.bucket_metrics.lock().await; + if let Some(existing_metrics) = bucket_metrics_guard.get(bucket) { + let mut updated_metrics = existing_metrics.clone(); + updated_metrics.total_objects = objects_scanned; + updated_metrics.objects_with_issues = objects_with_issues; + updated_metrics.scan_duration = Some(SystemTime::now().duration_since(scan_start).unwrap_or(Duration::ZERO)); + bucket_metrics_guard.insert(bucket.to_string(), updated_metrics); + } + } + + // Update state + { + let mut state = self.state.write().await; + state.scanning_buckets.retain(|b| b != bucket); + } + + debug!( + "Completed scanning bucket: {} on disk {} ({} objects, {} issues)", + bucket, + disk.to_string(), + objects_scanned, + objects_with_issues + ); + + Ok(object_metadata) + } + + /// Analyze object distribution across all disks and perform EC verification + /// + /// This method takes the collected object metadata from all disks and: + /// 1. Creates a union of all objects across all disks + /// 2. Identifies missing objects on each disk (for healing) + /// 3. Performs EC decode verification for deep scan mode + async fn analyze_object_distribution( + &self, + all_disk_objects: &[HashMap>], + disks: &[DiskStore], + ) -> Result<()> { + info!("Analyzing object distribution across {} disks", disks.len()); + + // Step 1: Create union of all objects across all disks + let mut all_objects = HashMap::new(); // bucket -> Set + let mut object_locations = HashMap::new(); // (bucket, object) -> Vec + + for (disk_idx, disk_objects) in all_disk_objects.iter().enumerate() { + for (bucket, objects) in disk_objects { + if bucket == RUSTFS_META_BUCKET { + // Skip internal system bucket during analysis to speed up tests + continue; + } + // Add bucket to all_objects + let bucket_objects = all_objects + .entry(bucket.clone()) + .or_insert_with(std::collections::HashSet::new); + + for (object_name, _file_meta) in objects.iter() { + bucket_objects.insert(object_name.clone()); + + // Record which disk has this object + let key = (bucket.clone(), object_name.clone()); + let locations = object_locations.entry(key).or_insert_with(Vec::new); + locations.push(disk_idx); + } + } + } + + info!( + "Found {} buckets with {} total objects", + all_objects.len(), + all_objects.values().map(|s| s.len()).sum::() + ); + + // Step 2: Identify missing objects and perform EC verification + let mut objects_needing_heal = 0u64; + let mut objects_with_ec_issues = 0u64; + + for (bucket, objects) in &all_objects { + // Skip internal RustFS system bucket to avoid lengthy checks on temporary/trash objects + if bucket == RUSTFS_META_BUCKET { + continue; + } + for object_name in objects { + let key = (bucket.clone(), object_name.clone()); + let empty_vec = Vec::new(); + let locations = object_locations.get(&key).unwrap_or(&empty_vec); + + // Check if object is missing from some disks + if locations.len() < disks.len() { + objects_needing_heal += 1; + let missing_disks: Vec = (0..disks.len()).filter(|&i| !locations.contains(&i)).collect(); + warn!("Object {}/{} missing from disks: {:?}", bucket, object_name, missing_disks); + println!("Object {bucket}/{object_name} missing from disks: {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 + ); + + // Step 4: Collect data usage statistics if enabled + let config = self.config.read().await; + if config.enable_data_usage_stats { + if let Err(e) = self.collect_data_usage_statistics(all_disk_objects).await { + error!("Failed to collect data usage statistics: {}", e); + } + } + drop(config); + + Ok(()) + } + + /// 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 {bucket}/{object}: need {read_quorum} reads, got {successful_reads} (errors: {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); + + // Get global cancel token + let cancel_token = if let Some(global_token) = get_ahm_services_cancel_token() { + global_token.clone() + } else { + CancellationToken::new() + }; + + loop { + tokio::select! { + _ = interval.tick() => { + // Check if scanner should still be running + if !self.state.read().await.is_running { + break; + } + + // 检查取消信号 + if cancel_token.is_cancelled() { + info!("Cancellation requested, exiting scanner loop"); + break; + } + + // Determine if it's time for a deep scan + let current_time = SystemTime::now(); + let last_deep_scan_time = self.state.read().await.last_deep_scan_time.unwrap_or(SystemTime::UNIX_EPOCH); + + if current_time.duration_since(last_deep_scan_time).unwrap_or(Duration::ZERO) >= deep_scan_interval { + info!("Deep scan interval reached, switching to deep scan mode"); + self.config.write().await.scan_mode = ScanMode::Deep; + self.state.write().await.last_deep_scan_time = Some(current_time); + } + + // Perform scan cycle + if let Err(e) = self.scan_cycle().await { + error!("Scan cycle failed: {}", e); + } + } + _ = cancel_token.cancelled() => { + info!("Received cancellation, stopping scanner loop"); + break; + } + } + } + + info!("Scanner loop stopped"); + Ok(()) + } + + /// Collect data usage statistics from scanned objects + async fn collect_data_usage_statistics( + &self, + all_disk_objects: &[HashMap>], + ) -> Result<()> { + info!("Collecting data usage statistics from {} disk scans", all_disk_objects.len()); + + let mut data_usage = DataUsageInfo::default(); + + // Collect objects from all disks (avoid duplicates by using first occurrence) + let mut processed_objects = std::collections::HashSet::new(); + + for disk_objects in all_disk_objects { + for (bucket_name, objects) in disk_objects { + if bucket_name == RUSTFS_META_BUCKET { + continue; // skip internal bucket from data usage stats + } + for object_name in objects.keys() { + let object_key = format!("{bucket_name}/{object_name}"); + + // Skip if already processed (avoid duplicates across disks) + if !processed_objects.insert(object_key.clone()) { + continue; + } + + // Add object to data usage statistics (pass entire FileMeta for accurate version counting) + data_usage.add_object_from_file_meta(&object_key, objects.get(object_name).unwrap()); + } + } + } + + // Ensure buckets_count is correctly set + data_usage.buckets_count = data_usage.buckets_usage.len() as u64; + + // Log statistics before storing + info!( + "Collected data usage statistics: {} buckets, {} total objects, {} total size", + data_usage.buckets_count, data_usage.objects_total_count, data_usage.objects_total_size + ); + + // Store in cache and update last collection time + let current_time = SystemTime::now(); + { + let mut data_usage_guard = self.data_usage_stats.lock().await; + data_usage_guard.insert("current".to_string(), data_usage.clone()); + } + { + let mut last_collection = self.last_data_usage_collection.write().await; + *last_collection = Some(current_time); + } + + // Store to backend if configured (spawned to avoid blocking scan loop) + let config = self.config.read().await; + if config.enable_data_usage_stats { + if let Some(store) = rustfs_ecstore::new_object_layer_fn() { + // Offload persistence to background task + let data_clone = data_usage.clone(); + tokio::spawn(async move { + if let Err(e) = super::data_usage::store_data_usage_in_backend(data_clone, store).await { + error!("Failed to store data usage statistics to backend: {}", e); + } else { + info!("Successfully stored data usage statistics to backend"); + } + }); + } else { + warn!("Storage not available, skipping backend persistence"); + } + } + + Ok(()) + } + + /// Clone scanner for background tasks + fn clone_for_background(&self) -> Self { + Self { + config: self.config.clone(), + state: Arc::clone(&self.state), + metrics: Arc::clone(&self.metrics), + bucket_metrics: Arc::clone(&self.bucket_metrics), + disk_metrics: Arc::clone(&self.disk_metrics), + data_usage_stats: Arc::clone(&self.data_usage_stats), + last_data_usage_collection: Arc::clone(&self.last_data_usage_collection), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use rustfs_ecstore::disk::endpoint::Endpoint; + use rustfs_ecstore::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints}; + use rustfs_ecstore::store::ECStore; + use rustfs_ecstore::{ + store_api::{MakeBucketOptions, ObjectIO, PutObjReader}, + StorageAPI, + }; + use std::fs; + use std::net::SocketAddr; + + async fn prepare_test_env(test_dir: Option<&str>, port: Option) -> (Vec, Arc) { + // create temp dir as 4 disks + let test_base_dir = test_dir.unwrap_or("/tmp/rustfs_ahm_test"); + let temp_dir = std::path::PathBuf::from(test_base_dir); + if temp_dir.exists() { + fs::remove_dir_all(&temp_dir).unwrap(); + } + fs::create_dir_all(&temp_dir).unwrap(); + + // create 4 disk dirs + let disk_paths = vec![ + temp_dir.join("disk1"), + temp_dir.join("disk2"), + temp_dir.join("disk3"), + temp_dir.join("disk4"), + ]; + + for disk_path in &disk_paths { + fs::create_dir_all(disk_path).unwrap(); + } + + // create EndpointServerPools + let mut endpoints = Vec::new(); + for (i, disk_path) in disk_paths.iter().enumerate() { + let mut endpoint = Endpoint::try_from(disk_path.to_str().unwrap()).unwrap(); + // set correct index + endpoint.set_pool_index(0); + endpoint.set_set_index(0); + endpoint.set_disk_index(i); + endpoints.push(endpoint); + } + + let pool_endpoints = PoolEndpoints { + legacy: false, + set_count: 1, + drives_per_set: 4, + endpoints: Endpoints::from(endpoints), + cmd_line: "test".to_string(), + platform: format!("OS: {} | Arch: {}", std::env::consts::OS, std::env::consts::ARCH), + }; + + let endpoint_pools = EndpointServerPools(vec![pool_endpoints]); + + // format disks + rustfs_ecstore::store::init_local_disks(endpoint_pools.clone()).await.unwrap(); + + // create ECStore with dynamic port + let port = port.unwrap_or(9000); + let server_addr: SocketAddr = format!("127.0.0.1:{port}").parse().unwrap(); + let ecstore = ECStore::new(server_addr, endpoint_pools).await.unwrap(); + + // init bucket metadata system + let buckets_list = ecstore + .list_bucket(&rustfs_ecstore::store_api::BucketOptions { + no_metadata: true, + ..Default::default() + }) + .await + .unwrap(); + let buckets = buckets_list.into_iter().map(|v| v.name).collect(); + rustfs_ecstore::bucket::metadata_sys::init_bucket_metadata_sys(ecstore.clone(), buckets).await; + + (disk_paths, ecstore) + } + + #[tokio::test(flavor = "multi_thread")] + #[ignore] + async fn test_scanner_basic_functionality() { + const TEST_DIR_BASIC: &str = "/tmp/rustfs_ahm_test_basic"; + let (disk_paths, ecstore) = prepare_test_env(Some(TEST_DIR_BASIC), Some(9001)).await; + + // create some test data + let bucket_name = "test-bucket"; + let object_name = "test-object"; + let test_data = b"Hello, RustFS!"; + + // create bucket and verify + let bucket_opts = MakeBucketOptions::default(); + ecstore + .make_bucket(bucket_name, &bucket_opts) + .await + .expect("make_bucket failed"); + + // check bucket really exists + let buckets = ecstore + .list_bucket(&rustfs_ecstore::store_api::BucketOptions::default()) + .await + .unwrap(); + assert!(buckets.iter().any(|b| b.name == bucket_name), "bucket not found after creation"); + + // write object + let mut put_reader = PutObjReader::from_vec(test_data.to_vec()); + let object_opts = rustfs_ecstore::store_api::ObjectOptions::default(); + ecstore + .put_object(bucket_name, object_name, &mut put_reader, &object_opts) + .await + .expect("put_object failed"); + + // create Scanner and test basic functionality + let scanner = Scanner::new(None); + + // Test 1: Normal scan - verify object is found + println!("=== Test 1: Normal scan ==="); + let scan_result = scanner.scan_cycle().await; + assert!(scan_result.is_ok(), "Normal scan should succeed"); + let metrics = scanner.get_metrics().await; + assert!(metrics.objects_scanned > 0, "Objects scanned should be positive"); + println!("Normal scan completed successfully"); + + // Test 2: Simulate disk corruption - delete object data from disk1 + println!("=== Test 2: Simulate disk corruption ==="); + let disk1_bucket_path = disk_paths[0].join(bucket_name); + let disk1_object_path = disk1_bucket_path.join(object_name); + + // Try to delete the object file from disk1 (simulate corruption) + // Note: This might fail if ECStore is actively using the file + match fs::remove_dir_all(&disk1_object_path) { + Ok(_) => { + println!("Successfully deleted object from disk1: {disk1_object_path:?}"); + + // Verify deletion by checking if the directory still exists + if disk1_object_path.exists() { + println!("WARNING: Directory still exists after deletion: {disk1_object_path:?}"); + } else { + println!("Confirmed: Directory was successfully deleted"); + } + } + Err(e) => { + println!("Could not delete object from disk1 (file may be in use): {disk1_object_path:?} - {e}"); + // This is expected behavior - ECStore might be holding file handles + } + } + + // Scan again - should still complete (even with missing data) + let scan_result_after_corruption = scanner.scan_cycle().await; + println!("Scan after corruption result: {scan_result_after_corruption:?}"); + + // Scanner should handle missing data gracefully + assert!(scan_result_after_corruption.is_ok(), "Scanner should handle missing data gracefully"); + + // Test 3: Verify EC decode capability + println!("=== Test 3: Verify EC decode ==="); + // Note: EC decode verification is done internally during scan_cycle + // We can verify that the scanner handles missing data gracefully + println!("EC decode verification is handled internally during scan cycles"); + + // Test 4: Test metrics collection + println!("=== Test 4: Metrics collection ==="); + let final_metrics = scanner.get_metrics().await; + println!("Final metrics: {final_metrics:?}"); + + // Verify metrics are reasonable + assert!(final_metrics.total_cycles > 0, "Should have completed scan cycles"); + assert!(final_metrics.last_activity.is_some(), "Should have scan activity"); + + // clean up temp dir + let temp_dir = std::path::PathBuf::from(TEST_DIR_BASIC); + if let Err(e) = fs::remove_dir_all(&temp_dir) { + eprintln!("Warning: Failed to clean up temp directory {temp_dir:?}: {e}"); + } + } + + // test data usage statistics collection and validation + #[tokio::test(flavor = "multi_thread")] + #[ignore] + async fn test_scanner_usage_stats() { + const TEST_DIR_USAGE_STATS: &str = "/tmp/rustfs_ahm_test_usage_stats"; + let (_, ecstore) = prepare_test_env(Some(TEST_DIR_USAGE_STATS), Some(9002)).await; + + // prepare test bucket and object + let bucket = "test-bucket"; + ecstore.make_bucket(bucket, &Default::default()).await.unwrap(); + let mut pr = PutObjReader::from_vec(b"hello".to_vec()); + ecstore + .put_object(bucket, "obj1", &mut pr, &Default::default()) + .await + .unwrap(); + + let scanner = Scanner::new(None); + + // enable statistics + { + let mut cfg = scanner.config.write().await; + cfg.enable_data_usage_stats = true; + } + + // first scan and get statistics + scanner.scan_cycle().await.unwrap(); + let du_initial = scanner.get_data_usage_info().await.unwrap(); + assert!(du_initial.objects_total_count > 0); + + // write 3 more objects and get statistics again + for size in [1024, 2048, 4096] { + let name = format!("obj_{size}"); + let mut pr = PutObjReader::from_vec(vec![b'x'; size]); + ecstore.put_object(bucket, &name, &mut pr, &Default::default()).await.unwrap(); + } + + scanner.scan_cycle().await.unwrap(); + let du_after = scanner.get_data_usage_info().await.unwrap(); + assert!(du_after.objects_total_count >= du_initial.objects_total_count + 3); + + // verify correctness of persisted data + tokio::time::sleep(std::time::Duration::from_millis(100)).await; + let persisted = crate::scanner::data_usage::load_data_usage_from_backend(ecstore.clone()) + .await + .expect("load persisted usage"); + assert_eq!(persisted.objects_total_count, du_after.objects_total_count); + assert_eq!(persisted.buckets_count, du_after.buckets_count); + let p_bucket = persisted.buckets_usage.get(bucket).unwrap(); + let m_bucket = du_after.buckets_usage.get(bucket).unwrap(); + assert_eq!(p_bucket.objects_count, m_bucket.objects_count); + assert_eq!(p_bucket.size, m_bucket.size); + + // consistency - again scan should not change count + scanner.scan_cycle().await.unwrap(); + let du_cons = scanner.get_data_usage_info().await.unwrap(); + assert_eq!(du_after.objects_total_count, du_cons.objects_total_count); + + // clean up temp dir + let _ = std::fs::remove_dir_all(std::path::Path::new(TEST_DIR_USAGE_STATS)); + } +} diff --git a/crates/ahm/src/scanner/data_usage.rs b/crates/ahm/src/scanner/data_usage.rs new file mode 100644 index 000000000..2ab97bff4 --- /dev/null +++ b/crates/ahm/src/scanner/data_usage.rs @@ -0,0 +1,671 @@ +// 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::SystemTime}; + +use rustfs_ecstore::{bucket::metadata_sys::get_replication_config, config::com::read_config, store::ECStore}; +use rustfs_utils::path::SLASH_SEPARATOR; +use serde::{Deserialize, Serialize}; +use tracing::{error, info, warn}; + +use crate::error::{Error, Result}; + +// Data usage storage constants +pub const DATA_USAGE_ROOT: &str = SLASH_SEPARATOR; +const DATA_USAGE_OBJ_NAME: &str = ".usage.json"; +const DATA_USAGE_BLOOM_NAME: &str = ".bloomcycle.bin"; +pub const DATA_USAGE_CACHE_NAME: &str = ".usage-cache.bin"; + +// Data usage storage paths +lazy_static::lazy_static! { + pub static ref DATA_USAGE_BUCKET: String = format!("{}{}{}", + rustfs_ecstore::disk::RUSTFS_META_BUCKET, + SLASH_SEPARATOR, + rustfs_ecstore::disk::BUCKET_META_PREFIX + ); + pub static ref DATA_USAGE_OBJ_NAME_PATH: String = format!("{}{}{}", + rustfs_ecstore::disk::BUCKET_META_PREFIX, + SLASH_SEPARATOR, + DATA_USAGE_OBJ_NAME + ); + pub static ref DATA_USAGE_BLOOM_NAME_PATH: String = format!("{}{}{}", + rustfs_ecstore::disk::BUCKET_META_PREFIX, + SLASH_SEPARATOR, + DATA_USAGE_BLOOM_NAME + ); +} + +/// Bucket target usage info provides replication statistics +#[derive(Debug, Default, Clone, Serialize, Deserialize)] +pub struct BucketTargetUsageInfo { + pub replication_pending_size: u64, + pub replication_failed_size: u64, + pub replicated_size: u64, + pub replica_size: u64, + pub replication_pending_count: u64, + pub replication_failed_count: u64, + pub replicated_count: u64, +} + +/// Bucket usage info provides bucket-level statistics +#[derive(Debug, Default, Clone, Serialize, Deserialize)] +pub struct BucketUsageInfo { + pub size: u64, + // Following five fields suffixed with V1 are here for backward compatibility + // Total Size for objects that have not yet been replicated + pub replication_pending_size_v1: u64, + // Total size for objects that have witness one or more failures and will be retried + pub replication_failed_size_v1: u64, + // Total size for objects that have been replicated to destination + pub replicated_size_v1: u64, + // Total number of objects pending replication + pub replication_pending_count_v1: u64, + // Total number of objects that failed replication + pub replication_failed_count_v1: u64, + + pub objects_count: u64, + pub object_size_histogram: HashMap, + pub object_versions_histogram: HashMap, + pub versions_count: u64, + pub delete_markers_count: u64, + pub replica_size: u64, + pub replica_count: u64, + pub replication_info: HashMap, +} + +/// DataUsageInfo represents data usage stats of the underlying storage +#[derive(Debug, Default, Clone, Serialize, Deserialize)] +pub struct DataUsageInfo { + /// Total capacity + pub total_capacity: u64, + /// Total used capacity + pub total_used_capacity: u64, + /// Total free capacity + pub total_free_capacity: u64, + + /// LastUpdate is the timestamp of when the data usage info was last updated + pub last_update: Option, + + /// Objects total count across all buckets + pub objects_total_count: u64, + /// Versions total count across all buckets + pub versions_total_count: u64, + /// Delete markers total count across all buckets + pub delete_markers_total_count: u64, + /// Objects total size across all buckets + pub objects_total_size: u64, + /// Replication info across all buckets + pub replication_info: HashMap, + + /// Total number of buckets in this cluster + pub buckets_count: u64, + /// Buckets usage info provides following information across all buckets + pub buckets_usage: HashMap, + /// Deprecated kept here for backward compatibility reasons + pub bucket_sizes: HashMap, +} + +/// Size summary for a single object or group of objects +#[derive(Debug, Default, Clone)] +pub struct SizeSummary { + /// Total size + pub total_size: usize, + /// Number of versions + pub versions: usize, + /// Number of delete markers + pub delete_markers: usize, + /// Replicated size + pub replicated_size: usize, + /// Replicated count + pub replicated_count: usize, + /// Pending size + pub pending_size: usize, + /// Failed size + pub failed_size: usize, + /// Replica size + pub replica_size: usize, + /// Replica count + pub replica_count: usize, + /// Pending count + pub pending_count: usize, + /// Failed count + pub failed_count: usize, + /// Replication target stats + pub repl_target_stats: HashMap, +} + +/// Replication target size summary +#[derive(Debug, Default, Clone)] +pub struct ReplTargetSizeSummary { + /// Replicated size + pub replicated_size: usize, + /// Replicated count + pub replicated_count: usize, + /// Pending size + pub pending_size: usize, + /// Failed size + pub failed_size: usize, + /// Pending count + pub pending_count: usize, + /// Failed count + pub failed_count: usize, +} + +impl DataUsageInfo { + /// Create a new DataUsageInfo + pub fn new() -> Self { + Self::default() + } + + /// Add object metadata to data usage statistics + pub fn add_object(&mut self, object_path: &str, meta_object: &rustfs_filemeta::MetaObject) { + // This method is kept for backward compatibility + // For accurate version counting, use add_object_from_file_meta instead + let bucket_name = match self.extract_bucket_from_path(object_path) { + Ok(name) => name, + Err(_) => return, + }; + + // Update bucket statistics + if let Some(bucket_usage) = self.buckets_usage.get_mut(&bucket_name) { + bucket_usage.size += meta_object.size as u64; + bucket_usage.objects_count += 1; + bucket_usage.versions_count += 1; // Simplified: assume 1 version per object + + // Update size histogram + let total_size = meta_object.size as u64; + let size_ranges = [ + ("0-1KB", 0, 1024), + ("1KB-1MB", 1024, 1024 * 1024), + ("1MB-10MB", 1024 * 1024, 10 * 1024 * 1024), + ("10MB-100MB", 10 * 1024 * 1024, 100 * 1024 * 1024), + ("100MB-1GB", 100 * 1024 * 1024, 1024 * 1024 * 1024), + ("1GB+", 1024 * 1024 * 1024, u64::MAX), + ]; + + for (range_name, min_size, max_size) in size_ranges { + if total_size >= min_size && total_size < max_size { + *bucket_usage.object_size_histogram.entry(range_name.to_string()).or_insert(0) += 1; + break; + } + } + + // Update version histogram (simplified - count as single version) + *bucket_usage + .object_versions_histogram + .entry("SINGLE_VERSION".to_string()) + .or_insert(0) += 1; + } else { + // Create new bucket usage + let mut bucket_usage = BucketUsageInfo { + size: meta_object.size as u64, + objects_count: 1, + versions_count: 1, + ..Default::default() + }; + bucket_usage.object_size_histogram.insert("0-1KB".to_string(), 1); + bucket_usage.object_versions_histogram.insert("SINGLE_VERSION".to_string(), 1); + self.buckets_usage.insert(bucket_name, bucket_usage); + } + + // Update global statistics + self.objects_total_size += meta_object.size as u64; + self.objects_total_count += 1; + self.versions_total_count += 1; + } + + /// Add object from FileMeta for accurate version counting + pub fn add_object_from_file_meta(&mut self, object_path: &str, file_meta: &rustfs_filemeta::FileMeta) { + let bucket_name = match self.extract_bucket_from_path(object_path) { + Ok(name) => name, + Err(_) => return, + }; + + // Calculate accurate statistics from all versions + let mut total_size = 0u64; + let mut versions_count = 0u64; + let mut delete_markers_count = 0u64; + let mut latest_object_size = 0u64; + + // Process all versions to get accurate counts + for version in &file_meta.versions { + match rustfs_filemeta::FileMetaVersion::try_from(version.clone()) { + Ok(ver) => { + if let Some(obj) = ver.object { + total_size += obj.size as u64; + versions_count += 1; + latest_object_size = obj.size as u64; // Keep track of latest object size + } else if ver.delete_marker.is_some() { + delete_markers_count += 1; + } + } + Err(_) => { + // Skip invalid versions + continue; + } + } + } + + // Update bucket statistics + if let Some(bucket_usage) = self.buckets_usage.get_mut(&bucket_name) { + bucket_usage.size += total_size; + bucket_usage.objects_count += 1; + bucket_usage.versions_count += versions_count; + bucket_usage.delete_markers_count += delete_markers_count; + + // Update size histogram based on latest object size + let size_ranges = [ + ("0-1KB", 0, 1024), + ("1KB-1MB", 1024, 1024 * 1024), + ("1MB-10MB", 1024 * 1024, 10 * 1024 * 1024), + ("10MB-100MB", 10 * 1024 * 1024, 100 * 1024 * 1024), + ("100MB-1GB", 100 * 1024 * 1024, 1024 * 1024 * 1024), + ("1GB+", 1024 * 1024 * 1024, u64::MAX), + ]; + + for (range_name, min_size, max_size) in size_ranges { + if latest_object_size >= min_size && latest_object_size < max_size { + *bucket_usage.object_size_histogram.entry(range_name.to_string()).or_insert(0) += 1; + break; + } + } + + // Update version histogram based on actual version count + let version_ranges = [ + ("1", 1, 1), + ("2-5", 2, 5), + ("6-10", 6, 10), + ("11-50", 11, 50), + ("51-100", 51, 100), + ("100+", 101, usize::MAX), + ]; + + for (range_name, min_versions, max_versions) in version_ranges { + if versions_count as usize >= min_versions && versions_count as usize <= max_versions { + *bucket_usage + .object_versions_histogram + .entry(range_name.to_string()) + .or_insert(0) += 1; + break; + } + } + } else { + // Create new bucket usage + let mut bucket_usage = BucketUsageInfo { + size: total_size, + objects_count: 1, + versions_count, + delete_markers_count, + ..Default::default() + }; + + // Set size histogram + let size_ranges = [ + ("0-1KB", 0, 1024), + ("1KB-1MB", 1024, 1024 * 1024), + ("1MB-10MB", 1024 * 1024, 10 * 1024 * 1024), + ("10MB-100MB", 10 * 1024 * 1024, 100 * 1024 * 1024), + ("100MB-1GB", 100 * 1024 * 1024, 1024 * 1024 * 1024), + ("1GB+", 1024 * 1024 * 1024, u64::MAX), + ]; + + for (range_name, min_size, max_size) in size_ranges { + if latest_object_size >= min_size && latest_object_size < max_size { + bucket_usage.object_size_histogram.insert(range_name.to_string(), 1); + break; + } + } + + // Set version histogram + let version_ranges = [ + ("1", 1, 1), + ("2-5", 2, 5), + ("6-10", 6, 10), + ("11-50", 11, 50), + ("51-100", 51, 100), + ("100+", 101, usize::MAX), + ]; + + for (range_name, min_versions, max_versions) in version_ranges { + if versions_count as usize >= min_versions && versions_count as usize <= max_versions { + bucket_usage.object_versions_histogram.insert(range_name.to_string(), 1); + break; + } + } + + self.buckets_usage.insert(bucket_name, bucket_usage); + // Update buckets count when adding new bucket + self.buckets_count = self.buckets_usage.len() as u64; + } + + // Update global statistics + self.objects_total_size += total_size; + self.objects_total_count += 1; + self.versions_total_count += versions_count; + self.delete_markers_total_count += delete_markers_count; + } + + /// Extract bucket name from object path + fn extract_bucket_from_path(&self, object_path: &str) -> Result { + let parts: Vec<&str> = object_path.split('/').collect(); + if parts.is_empty() { + return Err(Error::Scanner("Invalid object path: empty".to_string())); + } + Ok(parts[0].to_string()) + } + + /// Update capacity information + pub fn update_capacity(&mut self, total: u64, used: u64, free: u64) { + self.total_capacity = total; + self.total_used_capacity = used; + self.total_free_capacity = free; + self.last_update = Some(SystemTime::now()); + } + + /// Add bucket usage info + pub fn add_bucket_usage(&mut self, bucket: String, usage: BucketUsageInfo) { + self.buckets_usage.insert(bucket.clone(), usage); + self.buckets_count = self.buckets_usage.len() as u64; + self.last_update = Some(SystemTime::now()); + } + + /// Get bucket usage info + pub fn get_bucket_usage(&self, bucket: &str) -> Option<&BucketUsageInfo> { + self.buckets_usage.get(bucket) + } + + /// Calculate total statistics from all buckets + pub fn calculate_totals(&mut self) { + self.objects_total_count = 0; + self.versions_total_count = 0; + self.delete_markers_total_count = 0; + self.objects_total_size = 0; + + for usage in self.buckets_usage.values() { + self.objects_total_count += usage.objects_count; + self.versions_total_count += usage.versions_count; + self.delete_markers_total_count += usage.delete_markers_count; + self.objects_total_size += usage.size; + } + } + + /// Merge another DataUsageInfo into this one + pub fn merge(&mut self, other: &DataUsageInfo) { + // Merge bucket usage + for (bucket, usage) in &other.buckets_usage { + if let Some(existing) = self.buckets_usage.get_mut(bucket) { + existing.merge(usage); + } else { + self.buckets_usage.insert(bucket.clone(), usage.clone()); + } + } + + // Recalculate totals + self.calculate_totals(); + + // Ensure buckets_count stays consistent with buckets_usage + self.buckets_count = self.buckets_usage.len() as u64; + + // Update last update time + if let Some(other_update) = other.last_update { + if self.last_update.is_none() || other_update > self.last_update.unwrap() { + self.last_update = Some(other_update); + } + } + } +} + +impl BucketUsageInfo { + /// Create a new BucketUsageInfo + pub fn new() -> Self { + Self::default() + } + + /// Add size summary to this bucket usage + pub fn add_size_summary(&mut self, summary: &SizeSummary) { + self.size += summary.total_size as u64; + self.versions_count += summary.versions as u64; + self.delete_markers_count += summary.delete_markers as u64; + self.replica_size += summary.replica_size as u64; + self.replica_count += summary.replica_count as u64; + } + + /// Merge another BucketUsageInfo into this one + pub fn merge(&mut self, other: &BucketUsageInfo) { + self.size += other.size; + self.objects_count += other.objects_count; + self.versions_count += other.versions_count; + self.delete_markers_count += other.delete_markers_count; + self.replica_size += other.replica_size; + self.replica_count += other.replica_count; + + // Merge histograms + for (key, value) in &other.object_size_histogram { + *self.object_size_histogram.entry(key.clone()).or_insert(0) += value; + } + + for (key, value) in &other.object_versions_histogram { + *self.object_versions_histogram.entry(key.clone()).or_insert(0) += value; + } + + // Merge replication info + for (target, info) in &other.replication_info { + let entry = self.replication_info.entry(target.clone()).or_default(); + entry.replicated_size += info.replicated_size; + entry.replica_size += info.replica_size; + entry.replication_pending_size += info.replication_pending_size; + entry.replication_failed_size += info.replication_failed_size; + entry.replication_pending_count += info.replication_pending_count; + entry.replication_failed_count += info.replication_failed_count; + entry.replicated_count += info.replicated_count; + } + + // Merge backward compatibility fields + self.replication_pending_size_v1 += other.replication_pending_size_v1; + self.replication_failed_size_v1 += other.replication_failed_size_v1; + self.replicated_size_v1 += other.replicated_size_v1; + self.replication_pending_count_v1 += other.replication_pending_count_v1; + self.replication_failed_count_v1 += other.replication_failed_count_v1; + } +} + +impl SizeSummary { + /// Create a new SizeSummary + pub fn new() -> Self { + Self::default() + } + + /// Add another SizeSummary to this one + pub fn add(&mut self, other: &SizeSummary) { + self.total_size += other.total_size; + self.versions += other.versions; + self.delete_markers += other.delete_markers; + self.replicated_size += other.replicated_size; + self.replicated_count += other.replicated_count; + self.pending_size += other.pending_size; + self.failed_size += other.failed_size; + self.replica_size += other.replica_size; + self.replica_count += other.replica_count; + self.pending_count += other.pending_count; + self.failed_count += other.failed_count; + + // Merge replication target stats + for (target, stats) in &other.repl_target_stats { + let entry = self.repl_target_stats.entry(target.clone()).or_default(); + entry.replicated_size += stats.replicated_size; + entry.replicated_count += stats.replicated_count; + entry.pending_size += stats.pending_size; + entry.failed_size += stats.failed_size; + entry.pending_count += stats.pending_count; + entry.failed_count += stats.failed_count; + } + } +} + +/// Store data usage info to backend storage +pub async fn store_data_usage_in_backend(data_usage_info: DataUsageInfo, store: Arc) -> Result<()> { + let data = + serde_json::to_vec(&data_usage_info).map_err(|e| Error::Config(format!("Failed to serialize data usage info: {e}")))?; + + // Save to backend using the same mechanism as original code + rustfs_ecstore::config::com::save_config(store, &DATA_USAGE_OBJ_NAME_PATH, data) + .await + .map_err(Error::Storage)?; + + Ok(()) +} + +/// Load data usage info from backend storage +pub async fn load_data_usage_from_backend(store: Arc) -> Result { + let buf = match read_config(store, &DATA_USAGE_OBJ_NAME_PATH).await { + Ok(data) => data, + Err(e) => { + error!("Failed to read data usage info from backend: {}", e); + if e == rustfs_ecstore::error::Error::ConfigNotFound { + return Ok(DataUsageInfo::default()); + } + return Err(Error::Storage(e)); + } + }; + + let mut data_usage_info: DataUsageInfo = + serde_json::from_slice(&buf).map_err(|e| Error::Config(format!("Failed to deserialize data usage info: {e}")))?; + + warn!("Loaded data usage info from backend {:?}", &data_usage_info); + + // Handle backward compatibility like original code + if data_usage_info.buckets_usage.is_empty() { + data_usage_info.buckets_usage = data_usage_info + .bucket_sizes + .iter() + .map(|(bucket, &size)| { + ( + bucket.clone(), + BucketUsageInfo { + size, + ..Default::default() + }, + ) + }) + .collect(); + } + + if data_usage_info.bucket_sizes.is_empty() { + data_usage_info.bucket_sizes = data_usage_info + .buckets_usage + .iter() + .map(|(bucket, bui)| (bucket.clone(), bui.size)) + .collect(); + } + + for (bucket, bui) in &data_usage_info.buckets_usage { + if bui.replicated_size_v1 > 0 + || bui.replication_failed_count_v1 > 0 + || bui.replication_failed_size_v1 > 0 + || bui.replication_pending_count_v1 > 0 + { + if let Ok((cfg, _)) = get_replication_config(bucket).await { + if !cfg.role.is_empty() { + data_usage_info.replication_info.insert( + cfg.role.clone(), + BucketTargetUsageInfo { + replication_failed_size: bui.replication_failed_size_v1, + replication_failed_count: bui.replication_failed_count_v1, + replicated_size: bui.replicated_size_v1, + replication_pending_count: bui.replication_pending_count_v1, + replication_pending_size: bui.replication_pending_size_v1, + ..Default::default() + }, + ); + } + } + } + } + + Ok(data_usage_info) +} + +/// Example function showing how to use AHM data usage functionality +/// This demonstrates the integration pattern for DataUsageInfoHandler +pub async fn example_data_usage_integration() -> Result<()> { + // Get the global storage instance + let Some(store) = rustfs_ecstore::new_object_layer_fn() else { + return Err(Error::Config("Storage not initialized".to_string())); + }; + + // Load data usage from backend (this replaces the original load_data_usage_from_backend) + let data_usage = load_data_usage_from_backend(store).await?; + + info!( + "Loaded data usage info: {} buckets, {} total objects", + data_usage.buckets_count, data_usage.objects_total_count + ); + + // Example: Store updated data usage back to backend + // This would typically be called by the scanner after collecting new statistics + // store_data_usage_in_backend(data_usage, store).await?; + + Ok(()) +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_data_usage_info_creation() { + let mut info = DataUsageInfo::new(); + info.update_capacity(1000, 500, 500); + + assert_eq!(info.total_capacity, 1000); + assert_eq!(info.total_used_capacity, 500); + assert_eq!(info.total_free_capacity, 500); + assert!(info.last_update.is_some()); + } + + #[test] + fn test_bucket_usage_info_merge() { + let mut usage1 = BucketUsageInfo::new(); + usage1.size = 100; + usage1.objects_count = 10; + usage1.versions_count = 5; + + let mut usage2 = BucketUsageInfo::new(); + usage2.size = 200; + usage2.objects_count = 20; + usage2.versions_count = 10; + + usage1.merge(&usage2); + + assert_eq!(usage1.size, 300); + assert_eq!(usage1.objects_count, 30); + assert_eq!(usage1.versions_count, 15); + } + + #[test] + fn test_size_summary_add() { + let mut summary1 = SizeSummary::new(); + summary1.total_size = 100; + summary1.versions = 5; + + let mut summary2 = SizeSummary::new(); + summary2.total_size = 200; + summary2.versions = 10; + + summary1.add(&summary2); + + assert_eq!(summary1.total_size, 300); + assert_eq!(summary1.versions, 15); + } +} diff --git a/crates/ahm/src/scanner/histogram.rs b/crates/ahm/src/scanner/histogram.rs new file mode 100644 index 000000000..778569ced --- /dev/null +++ b/crates/ahm/src/scanner/histogram.rs @@ -0,0 +1,277 @@ +// 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; + +/// Size interval for object size histogram +#[derive(Debug, Clone)] +pub struct SizeInterval { + pub start: u64, + pub end: u64, + pub name: &'static str, +} + +/// Version interval for object versions histogram +#[derive(Debug, Clone)] +pub struct VersionInterval { + pub start: u64, + pub end: u64, + pub name: &'static str, +} + +/// Object size histogram intervals +pub const OBJECTS_HISTOGRAM_INTERVALS: &[SizeInterval] = &[ + SizeInterval { + start: 0, + end: 1024 - 1, + name: "LESS_THAN_1_KiB", + }, + SizeInterval { + start: 1024, + end: 1024 * 1024 - 1, + name: "1_KiB_TO_1_MiB", + }, + SizeInterval { + start: 1024 * 1024, + end: 10 * 1024 * 1024 - 1, + name: "1_MiB_TO_10_MiB", + }, + SizeInterval { + start: 10 * 1024 * 1024, + end: 64 * 1024 * 1024 - 1, + name: "10_MiB_TO_64_MiB", + }, + SizeInterval { + start: 64 * 1024 * 1024, + end: 128 * 1024 * 1024 - 1, + name: "64_MiB_TO_128_MiB", + }, + SizeInterval { + start: 128 * 1024 * 1024, + end: 512 * 1024 * 1024 - 1, + name: "128_MiB_TO_512_MiB", + }, + SizeInterval { + start: 512 * 1024 * 1024, + end: u64::MAX, + name: "MORE_THAN_512_MiB", + }, +]; + +/// Object version count histogram intervals +pub const OBJECTS_VERSION_COUNT_INTERVALS: &[VersionInterval] = &[ + VersionInterval { + start: 1, + end: 1, + name: "1_VERSION", + }, + VersionInterval { + start: 2, + end: 10, + name: "2_TO_10_VERSIONS", + }, + VersionInterval { + start: 11, + end: 100, + name: "11_TO_100_VERSIONS", + }, + VersionInterval { + start: 101, + end: 1000, + name: "101_TO_1000_VERSIONS", + }, + VersionInterval { + start: 1001, + end: u64::MAX, + name: "MORE_THAN_1000_VERSIONS", + }, +]; + +/// Size histogram for object size distribution +#[derive(Debug, Clone, Default)] +pub struct SizeHistogram { + counts: Vec, +} + +/// Versions histogram for object version count distribution +#[derive(Debug, Clone, Default)] +pub struct VersionsHistogram { + counts: Vec, +} + +impl SizeHistogram { + /// Create a new size histogram + pub fn new() -> Self { + Self { + counts: vec![0; OBJECTS_HISTOGRAM_INTERVALS.len()], + } + } + + /// Add a size to the histogram + pub fn add(&mut self, size: u64) { + for (idx, interval) in OBJECTS_HISTOGRAM_INTERVALS.iter().enumerate() { + if size >= interval.start && size <= interval.end { + self.counts[idx] += 1; + break; + } + } + } + + /// Get the histogram as a map + pub fn to_map(&self) -> HashMap { + let mut result = HashMap::new(); + for (idx, count) in self.counts.iter().enumerate() { + let interval = &OBJECTS_HISTOGRAM_INTERVALS[idx]; + result.insert(interval.name.to_string(), *count); + } + result + } + + /// Merge another histogram into this one + pub fn merge(&mut self, other: &SizeHistogram) { + for (idx, count) in other.counts.iter().enumerate() { + self.counts[idx] += count; + } + } + + /// Get total count + pub fn total_count(&self) -> u64 { + self.counts.iter().sum() + } + + /// Reset the histogram + pub fn reset(&mut self) { + for count in &mut self.counts { + *count = 0; + } + } +} + +impl VersionsHistogram { + /// Create a new versions histogram + pub fn new() -> Self { + Self { + counts: vec![0; OBJECTS_VERSION_COUNT_INTERVALS.len()], + } + } + + /// Add a version count to the histogram + pub fn add(&mut self, versions: u64) { + for (idx, interval) in OBJECTS_VERSION_COUNT_INTERVALS.iter().enumerate() { + if versions >= interval.start && versions <= interval.end { + self.counts[idx] += 1; + break; + } + } + } + + /// Get the histogram as a map + pub fn to_map(&self) -> HashMap { + let mut result = HashMap::new(); + for (idx, count) in self.counts.iter().enumerate() { + let interval = &OBJECTS_VERSION_COUNT_INTERVALS[idx]; + result.insert(interval.name.to_string(), *count); + } + result + } + + /// Merge another histogram into this one + pub fn merge(&mut self, other: &VersionsHistogram) { + for (idx, count) in other.counts.iter().enumerate() { + self.counts[idx] += count; + } + } + + /// Get total count + pub fn total_count(&self) -> u64 { + self.counts.iter().sum() + } + + /// Reset the histogram + pub fn reset(&mut self) { + for count in &mut self.counts { + *count = 0; + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_size_histogram() { + let mut histogram = SizeHistogram::new(); + + // Add some sizes + histogram.add(512); // LESS_THAN_1_KiB + histogram.add(1024); // 1_KiB_TO_1_MiB + histogram.add(1024 * 1024); // 1_MiB_TO_10_MiB + histogram.add(5 * 1024 * 1024); // 1_MiB_TO_10_MiB + + let map = histogram.to_map(); + + assert_eq!(map.get("LESS_THAN_1_KiB"), Some(&1)); + assert_eq!(map.get("1_KiB_TO_1_MiB"), Some(&1)); + assert_eq!(map.get("1_MiB_TO_10_MiB"), Some(&2)); + assert_eq!(map.get("10_MiB_TO_64_MiB"), Some(&0)); + } + + #[test] + fn test_versions_histogram() { + let mut histogram = VersionsHistogram::new(); + + // Add some version counts + histogram.add(1); // 1_VERSION + histogram.add(5); // 2_TO_10_VERSIONS + histogram.add(50); // 11_TO_100_VERSIONS + histogram.add(500); // 101_TO_1000_VERSIONS + + let map = histogram.to_map(); + + assert_eq!(map.get("1_VERSION"), Some(&1)); + assert_eq!(map.get("2_TO_10_VERSIONS"), Some(&1)); + assert_eq!(map.get("11_TO_100_VERSIONS"), Some(&1)); + assert_eq!(map.get("101_TO_1000_VERSIONS"), Some(&1)); + } + + #[test] + fn test_histogram_merge() { + let mut histogram1 = SizeHistogram::new(); + histogram1.add(1024); + histogram1.add(1024 * 1024); + + let mut histogram2 = SizeHistogram::new(); + histogram2.add(1024); + histogram2.add(5 * 1024 * 1024); + + histogram1.merge(&histogram2); + + let map = histogram1.to_map(); + assert_eq!(map.get("1_KiB_TO_1_MiB"), Some(&2)); // 1 from histogram1 + 1 from histogram2 + assert_eq!(map.get("1_MiB_TO_10_MiB"), Some(&2)); // 1 from histogram1 + 1 from histogram2 + } + + #[test] + fn test_histogram_reset() { + let mut histogram = SizeHistogram::new(); + histogram.add(1024); + histogram.add(1024 * 1024); + + assert_eq!(histogram.total_count(), 2); + + histogram.reset(); + assert_eq!(histogram.total_count(), 0); + } +} diff --git a/crates/ahm/src/scanner/metrics.rs b/crates/ahm/src/scanner/metrics.rs new file mode 100644 index 000000000..10d010058 --- /dev/null +++ b/crates/ahm/src/scanner/metrics.rs @@ -0,0 +1,284 @@ +// 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 +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct ScannerMetrics { + /// Total objects scanned since server start + pub objects_scanned: u64, + /// Total object versions scanned since server start + pub versions_scanned: u64, + /// Total directories scanned since server start + pub directories_scanned: u64, + /// Total bucket scans started since server start + pub bucket_scans_started: u64, + /// Total bucket scans finished since server start + pub bucket_scans_finished: u64, + /// Total objects with health issues found + pub objects_with_issues: u64, + /// Total heal tasks queued + pub heal_tasks_queued: u64, + /// Total heal tasks completed + pub heal_tasks_completed: u64, + /// Total heal tasks failed + pub heal_tasks_failed: u64, + /// Last scan activity time + pub last_activity: Option, + /// Current scan cycle + pub current_cycle: u64, + /// Total scan cycles completed + pub total_cycles: u64, + /// Current scan duration + pub current_scan_duration: Option, + /// Average scan duration + pub avg_scan_duration: Duration, + /// Objects scanned per second + pub objects_per_second: f64, + /// Buckets scanned per second + pub buckets_per_second: f64, + /// Storage metrics by bucket + pub bucket_metrics: HashMap, + /// Disk metrics + pub disk_metrics: HashMap, +} + +/// Bucket-specific metrics +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct BucketMetrics { + /// Bucket name + pub bucket: String, + /// Total objects in bucket + pub total_objects: u64, + /// Total size of objects in bucket (bytes) + pub total_size: u64, + /// Objects with health issues + pub objects_with_issues: u64, + /// Last scan time + pub last_scan_time: Option, + /// Scan duration + pub scan_duration: Option, + /// Heal tasks queued for this bucket + pub heal_tasks_queued: u64, + /// Heal tasks completed for this bucket + pub heal_tasks_completed: u64, + /// Heal tasks failed for this bucket + pub heal_tasks_failed: u64, +} + +/// Disk-specific metrics +#[derive(Debug, Clone, Default, Serialize, Deserialize)] +pub struct DiskMetrics { + /// Disk path + pub disk_path: String, + /// Total disk space (bytes) + pub total_space: u64, + /// Used disk space (bytes) + pub used_space: u64, + /// Free disk space (bytes) + pub free_space: u64, + /// Objects scanned on this disk + pub objects_scanned: u64, + /// Objects with issues on this disk + pub objects_with_issues: u64, + /// Last scan time + pub last_scan_time: Option, + /// Whether disk is online + pub is_online: bool, + /// Whether disk is being scanned + pub is_scanning: bool, +} + +/// Thread-safe metrics collector +pub struct MetricsCollector { + /// Atomic counters for real-time metrics + objects_scanned: AtomicU64, + versions_scanned: AtomicU64, + directories_scanned: AtomicU64, + bucket_scans_started: AtomicU64, + bucket_scans_finished: AtomicU64, + objects_with_issues: AtomicU64, + heal_tasks_queued: AtomicU64, + heal_tasks_completed: AtomicU64, + heal_tasks_failed: AtomicU64, + current_cycle: AtomicU64, + total_cycles: AtomicU64, +} + +impl MetricsCollector { + /// Create a new metrics collector + pub fn new() -> Self { + Self { + objects_scanned: AtomicU64::new(0), + versions_scanned: AtomicU64::new(0), + directories_scanned: AtomicU64::new(0), + bucket_scans_started: AtomicU64::new(0), + bucket_scans_finished: AtomicU64::new(0), + objects_with_issues: AtomicU64::new(0), + heal_tasks_queued: AtomicU64::new(0), + heal_tasks_completed: AtomicU64::new(0), + heal_tasks_failed: AtomicU64::new(0), + current_cycle: AtomicU64::new(0), + total_cycles: AtomicU64::new(0), + } + } + + /// Increment objects scanned count + pub fn increment_objects_scanned(&self, count: u64) { + self.objects_scanned.fetch_add(count, Ordering::Relaxed); + } + + /// Increment versions scanned count + pub fn increment_versions_scanned(&self, count: u64) { + self.versions_scanned.fetch_add(count, Ordering::Relaxed); + } + + /// Increment directories scanned count + pub fn increment_directories_scanned(&self, count: u64) { + self.directories_scanned.fetch_add(count, Ordering::Relaxed); + } + + /// Increment bucket scans started count + pub fn increment_bucket_scans_started(&self, count: u64) { + self.bucket_scans_started.fetch_add(count, Ordering::Relaxed); + } + + /// Increment bucket scans finished count + pub fn increment_bucket_scans_finished(&self, count: u64) { + self.bucket_scans_finished.fetch_add(count, Ordering::Relaxed); + } + + /// Increment objects with issues count + pub fn increment_objects_with_issues(&self, count: u64) { + self.objects_with_issues.fetch_add(count, Ordering::Relaxed); + } + + /// Increment heal tasks queued count + pub fn increment_heal_tasks_queued(&self, count: u64) { + self.heal_tasks_queued.fetch_add(count, Ordering::Relaxed); + } + + /// Increment heal tasks completed count + pub fn increment_heal_tasks_completed(&self, count: u64) { + self.heal_tasks_completed.fetch_add(count, Ordering::Relaxed); + } + + /// Increment heal tasks failed count + pub fn increment_heal_tasks_failed(&self, count: u64) { + self.heal_tasks_failed.fetch_add(count, Ordering::Relaxed); + } + + /// Set current cycle + pub fn set_current_cycle(&self, cycle: u64) { + self.current_cycle.store(cycle, Ordering::Relaxed); + } + + /// Increment total cycles + pub fn increment_total_cycles(&self) { + self.total_cycles.fetch_add(1, Ordering::Relaxed); + } + + /// Get current metrics snapshot + pub fn get_metrics(&self) -> ScannerMetrics { + ScannerMetrics { + objects_scanned: self.objects_scanned.load(Ordering::Relaxed), + versions_scanned: self.versions_scanned.load(Ordering::Relaxed), + directories_scanned: self.directories_scanned.load(Ordering::Relaxed), + bucket_scans_started: self.bucket_scans_started.load(Ordering::Relaxed), + bucket_scans_finished: self.bucket_scans_finished.load(Ordering::Relaxed), + objects_with_issues: self.objects_with_issues.load(Ordering::Relaxed), + heal_tasks_queued: self.heal_tasks_queued.load(Ordering::Relaxed), + heal_tasks_completed: self.heal_tasks_completed.load(Ordering::Relaxed), + heal_tasks_failed: self.heal_tasks_failed.load(Ordering::Relaxed), + last_activity: Some(SystemTime::now()), + current_cycle: self.current_cycle.load(Ordering::Relaxed), + total_cycles: self.total_cycles.load(Ordering::Relaxed), + current_scan_duration: None, // Will be set by scanner + avg_scan_duration: Duration::ZERO, // Will be calculated + objects_per_second: 0.0, // Will be calculated + buckets_per_second: 0.0, // Will be calculated + bucket_metrics: HashMap::new(), // Will be populated by scanner + disk_metrics: HashMap::new(), // Will be populated by scanner + } + } + + /// Reset all metrics + pub fn reset(&self) { + self.objects_scanned.store(0, Ordering::Relaxed); + self.versions_scanned.store(0, Ordering::Relaxed); + self.directories_scanned.store(0, Ordering::Relaxed); + self.bucket_scans_started.store(0, Ordering::Relaxed); + self.bucket_scans_finished.store(0, Ordering::Relaxed); + self.objects_with_issues.store(0, Ordering::Relaxed); + self.heal_tasks_queued.store(0, Ordering::Relaxed); + self.heal_tasks_completed.store(0, Ordering::Relaxed); + self.heal_tasks_failed.store(0, Ordering::Relaxed); + self.current_cycle.store(0, Ordering::Relaxed); + self.total_cycles.store(0, Ordering::Relaxed); + + info!("Scanner metrics reset"); + } +} + +impl Default for MetricsCollector { + fn default() -> Self { + Self::new() + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn test_metrics_collector_creation() { + let collector = MetricsCollector::new(); + let metrics = collector.get_metrics(); + assert_eq!(metrics.objects_scanned, 0); + assert_eq!(metrics.versions_scanned, 0); + } + + #[test] + fn test_metrics_increment() { + let collector = MetricsCollector::new(); + + collector.increment_objects_scanned(10); + collector.increment_versions_scanned(5); + collector.increment_objects_with_issues(2); + + let metrics = collector.get_metrics(); + assert_eq!(metrics.objects_scanned, 10); + assert_eq!(metrics.versions_scanned, 5); + assert_eq!(metrics.objects_with_issues, 2); + } + + #[test] + fn test_metrics_reset() { + let collector = MetricsCollector::new(); + + collector.increment_objects_scanned(10); + collector.reset(); + + let metrics = collector.get_metrics(); + assert_eq!(metrics.objects_scanned, 0); + } +} diff --git a/crates/ahm/src/scanner/mod.rs b/crates/ahm/src/scanner/mod.rs new file mode 100644 index 000000000..e5a116142 --- /dev/null +++ b/crates/ahm/src/scanner/mod.rs @@ -0,0 +1,25 @@ +// 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. + +pub mod data_scanner; +pub mod data_usage; +pub mod histogram; +pub mod metrics; + +// Re-export main types for convenience +pub use data_scanner::Scanner; +pub use data_usage::{ + load_data_usage_from_backend, store_data_usage_in_backend, BucketTargetUsageInfo, BucketUsageInfo, DataUsageInfo, +}; +pub use metrics::ScannerMetrics; diff --git a/crates/ecstore/src/global.rs b/crates/ecstore/src/global.rs index 37760eeea..93c38fe0d 100644 --- a/crates/ecstore/src/global.rs +++ b/crates/ecstore/src/global.rs @@ -30,6 +30,7 @@ use std::{ time::SystemTime, }; use tokio::sync::{OnceCell, RwLock}; +use tokio_util::sync::CancellationToken; use uuid::Uuid; pub const DISK_ASSUME_UNKNOWN_SIZE: u64 = 1 << 30; @@ -66,6 +67,9 @@ pub static ref GLOBAL_NodeNamesHex: HashMap = HashMap::new(); pub static ref GLOBAL_REGION: OnceLock = OnceLock::new(); } +// Global cancellation token for background services (data scanner and auto heal) +static GLOBAL_BACKGROUND_SERVICES_CANCEL_TOKEN: OnceLock = OnceLock::new(); + static GLOBAL_ACTIVE_CRED: OnceLock = OnceLock::new(); pub fn init_global_action_cred(ak: Option, sk: Option) { @@ -192,3 +196,27 @@ pub fn set_global_region(region: String) { pub fn get_global_region() -> Option { GLOBAL_REGION.get().cloned() } + +/// Initialize the global background services cancellation token +pub fn init_background_services_cancel_token(cancel_token: CancellationToken) -> Result<(), CancellationToken> { + GLOBAL_BACKGROUND_SERVICES_CANCEL_TOKEN.set(cancel_token) +} + +/// Get the global background services cancellation token +pub fn get_background_services_cancel_token() -> Option<&'static CancellationToken> { + GLOBAL_BACKGROUND_SERVICES_CANCEL_TOKEN.get() +} + +/// Create and initialize the global background services cancellation token +pub fn create_background_services_cancel_token() -> CancellationToken { + let cancel_token = CancellationToken::new(); + init_background_services_cancel_token(cancel_token.clone()).expect("Background services cancel token already initialized"); + cancel_token +} + +/// Shutdown all background services gracefully +pub fn shutdown_background_services() { + if let Some(cancel_token) = GLOBAL_BACKGROUND_SERVICES_CANCEL_TOKEN.get() { + cancel_token.cancel(); + } +} diff --git a/crates/ecstore/src/heal/background_heal_ops.rs b/crates/ecstore/src/heal/background_heal_ops.rs index e447a5852..0bc3e2591 100644 --- a/crates/ecstore/src/heal/background_heal_ops.rs +++ b/crates/ecstore/src/heal/background_heal_ops.rs @@ -24,6 +24,7 @@ use tokio::{ }, time::interval, }; +use tokio_util::sync::CancellationToken; use tracing::{error, info}; use uuid::Uuid; @@ -32,7 +33,7 @@ use super::{ heal_ops::{HealSequence, new_bg_heal_sequence}, }; use crate::error::{Error, Result}; -use crate::global::GLOBAL_MRFState; +use crate::global::{GLOBAL_MRFState, get_background_services_cancel_token}; use crate::heal::error::ERR_RETRY_HEALING; use crate::heal::heal_commands::{HEAL_ITEM_BUCKET, HealScanMode}; use crate::heal::heal_ops::{BG_HEALING_UUID, HealSource}; @@ -54,6 +55,13 @@ use crate::{ pub static DEFAULT_MONITOR_NEW_DISK_INTERVAL: Duration = Duration::from_secs(10); pub async fn init_auto_heal() { + info!("Initializing auto heal background task"); + + let Some(cancel_token) = get_background_services_cancel_token() else { + error!("Background services cancel token not initialized"); + return; + }; + init_background_healing().await; let v = env::var("_RUSTFS_AUTO_DRIVE_HEALING").unwrap_or("on".to_string()); if v == "on" { @@ -61,12 +69,16 @@ pub async fn init_auto_heal() { GLOBAL_BackgroundHealState .push_heal_local_disks(&get_local_disks_to_heal().await) .await; - spawn(async { - monitor_local_disks_and_heal().await; + + let cancel_clone = cancel_token.clone(); + spawn(async move { + monitor_local_disks_and_heal(cancel_clone).await; }); } - spawn(async { - GLOBAL_MRFState.heal_routine().await; + + let cancel_clone = cancel_token.clone(); + spawn(async move { + GLOBAL_MRFState.heal_routine_with_cancel(cancel_clone).await; }); } @@ -108,50 +120,66 @@ pub async fn get_local_disks_to_heal() -> Vec { disks_to_heal } -async fn monitor_local_disks_and_heal() { +async fn monitor_local_disks_and_heal(cancel_token: CancellationToken) { + info!("Auto heal monitor started"); let mut interval = interval(DEFAULT_MONITOR_NEW_DISK_INTERVAL); loop { - interval.tick().await; - let heal_disks = GLOBAL_BackgroundHealState.get_heal_local_disk_endpoints().await; - if heal_disks.is_empty() { - info!("heal local disks is empty"); - interval.reset(); - continue; - } + tokio::select! { + _ = cancel_token.cancelled() => { + info!("Auto heal monitor received shutdown signal, exiting gracefully"); + break; + } + _ = interval.tick() => { + let heal_disks = GLOBAL_BackgroundHealState.get_heal_local_disk_endpoints().await; + if heal_disks.is_empty() { + info!("heal local disks is empty"); + interval.reset(); + continue; + } - info!("heal local disks: {:?}", heal_disks); + info!("heal local disks: {:?}", heal_disks); - let store = new_object_layer_fn().expect("errServerNotInitialized"); - if let (_result, Some(err)) = store.heal_format(false).await.expect("heal format failed") { - error!("heal local disk format error: {}", err); - if err == Error::NoHealRequired { - } else { - info!("heal format err: {}", err.to_string()); + let store = new_object_layer_fn().expect("errServerNotInitialized"); + if let (_result, Some(err)) = store.heal_format(false).await.expect("heal format failed") { + error!("heal local disk format error: {}", err); + if err == Error::NoHealRequired { + } else { + info!("heal format err: {}", err.to_string()); + interval.reset(); + continue; + } + } + + let mut futures = Vec::new(); + for disk in heal_disks.into_ref().iter() { + let disk_clone = disk.clone(); + let cancel_clone = cancel_token.clone(); + futures.push(async move { + let disk_for_cancel = disk_clone.clone(); + tokio::select! { + _ = cancel_clone.cancelled() => { + info!("Disk healing task cancelled for disk: {}", disk_for_cancel); + } + _ = async { + GLOBAL_BackgroundHealState + .set_disk_healing_status(disk_clone.clone(), true) + .await; + if heal_fresh_disk(&disk_clone).await.is_err() { + info!("heal_fresh_disk is err"); + GLOBAL_BackgroundHealState + .set_disk_healing_status(disk_clone.clone(), false) + .await; + } + GLOBAL_BackgroundHealState.pop_heal_local_disks(&[disk_clone]).await; + } => {} + } + }); + } + let _ = join_all(futures).await; interval.reset(); - continue; } } - - let mut futures = Vec::new(); - for disk in heal_disks.into_ref().iter() { - let disk_clone = disk.clone(); - futures.push(async move { - GLOBAL_BackgroundHealState - .set_disk_healing_status(disk_clone.clone(), true) - .await; - if heal_fresh_disk(&disk_clone).await.is_err() { - info!("heal_fresh_disk is err"); - GLOBAL_BackgroundHealState - .set_disk_healing_status(disk_clone.clone(), false) - .await; - return; - } - GLOBAL_BackgroundHealState.pop_heal_local_disks(&[disk_clone]).await; - }); - } - let _ = join_all(futures).await; - interval.reset(); } } diff --git a/crates/ecstore/src/heal/data_scanner.rs b/crates/ecstore/src/heal/data_scanner.rs index 1b4721af5..0a6a88fe9 100644 --- a/crates/ecstore/src/heal/data_scanner.rs +++ b/crates/ecstore/src/heal/data_scanner.rs @@ -20,14 +20,13 @@ use std::{ path::{Path, PathBuf}, pin::Pin, sync::{ - Arc, OnceLock, + Arc, atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering}, }, time::{Duration, SystemTime}, }; use time::{self, OffsetDateTime}; -use tokio_util::sync::CancellationToken; use super::{ data_scanner_metric::{ScannerMetric, ScannerMetrics, globalScannerMetrics}, @@ -51,7 +50,7 @@ use crate::{ metadata_sys, }, event_notification::{EventArgs, send_event}, - global::GLOBAL_LocalNodeName, + global::{GLOBAL_LocalNodeName, get_background_services_cancel_token}, store_api::{ObjectOptions, ObjectToDelete, StorageAPI}, }; use crate::{ @@ -128,8 +127,6 @@ lazy_static! { pub static ref globalHealConfig: Arc> = Arc::new(RwLock::new(Config::default())); } -static GLOBAL_SCANNER_CANCEL_TOKEN: OnceLock = OnceLock::new(); - struct DynamicSleeper { factor: f64, max_sleep: Duration, @@ -198,21 +195,18 @@ fn new_dynamic_sleeper(factor: f64, max_wait: Duration, is_scanner: bool) -> Dyn /// - Minimum sleep duration to avoid excessive CPU usage /// - Proper error handling and logging /// -/// # Returns -/// A CancellationToken that can be used to gracefully shutdown the scanner -/// /// # Architecture /// 1. Initialize with random seed for sleep intervals /// 2. Run scanner cycles in a loop /// 3. Use randomized sleep between cycles to avoid thundering herd /// 4. Ensure minimum sleep duration to prevent CPU thrashing -pub async fn init_data_scanner() -> CancellationToken { +pub async fn init_data_scanner() { info!("Initializing data scanner background task"); - let cancel_token = CancellationToken::new(); - GLOBAL_SCANNER_CANCEL_TOKEN - .set(cancel_token.clone()) - .expect("Scanner already initialized"); + let Some(cancel_token) = get_background_services_cancel_token() else { + error!("Background services cancel token not initialized"); + return; + }; let cancel_clone = cancel_token.clone(); tokio::spawn(async move { @@ -256,8 +250,6 @@ pub async fn init_data_scanner() -> CancellationToken { info!("Data scanner background task stopped gracefully"); }); - - cancel_token } /// Run a single data scanner cycle @@ -282,7 +274,7 @@ async fn run_data_scanner_cycle() { }; // Check for cancellation before starting expensive operations - if let Some(token) = GLOBAL_SCANNER_CANCEL_TOKEN.get() { + if let Some(token) = get_background_services_cancel_token() { if token.is_cancelled() { debug!("Scanner cancelled before starting cycle"); return; @@ -397,9 +389,8 @@ async fn execute_namespace_scan( cycle: u64, scan_mode: HealScanMode, ) -> Result<()> { - let cancel_token = GLOBAL_SCANNER_CANCEL_TOKEN - .get() - .ok_or_else(|| Error::other("Scanner not initialized"))?; + let cancel_token = + get_background_services_cancel_token().ok_or_else(|| Error::other("Background services not initialized"))?; tokio::select! { result = store.ns_scanner(tx, cycle as usize, scan_mode) => { diff --git a/crates/ecstore/src/heal/mrf.rs b/crates/ecstore/src/heal/mrf.rs index 106dde7ef..3f42fa202 100644 --- a/crates/ecstore/src/heal/mrf.rs +++ b/crates/ecstore/src/heal/mrf.rs @@ -25,7 +25,8 @@ use std::time::Duration; use tokio::sync::RwLock; use tokio::sync::mpsc::{Receiver, Sender}; use tokio::time::sleep; -use tracing::error; +use tokio_util::sync::CancellationToken; +use tracing::{error, info}; use uuid::Uuid; pub const MRF_OPS_QUEUE_SIZE: u64 = 100000; @@ -87,56 +88,96 @@ impl MRFState { let _ = self.tx.send(op).await; } - pub async fn heal_routine(&self) { + /// Enhanced heal routine with cancellation support + /// + /// This method implements the same healing logic as the original heal_routine, + /// but adds proper cancellation support via CancellationToken. + /// The core logic remains identical to maintain compatibility. + pub async fn heal_routine_with_cancel(&self, cancel_token: CancellationToken) { + info!("MRF heal routine started with cancellation support"); + loop { - // rx used only there, - if let Some(op) = self.rx.write().await.recv().await { - if op.bucket == RUSTFS_META_BUCKET { - for pattern in &*PATTERNS { - if pattern.is_match(&op.object) { - return; + tokio::select! { + _ = cancel_token.cancelled() => { + info!("MRF heal routine received shutdown signal, exiting gracefully"); + break; + } + op_result = async { + let mut rx_guard = self.rx.write().await; + rx_guard.recv().await + } => { + if let Some(op) = op_result { + // Special path filtering (original logic) + if op.bucket == RUSTFS_META_BUCKET { + for pattern in &*PATTERNS { + if pattern.is_match(&op.object) { + continue; // Skip this operation, continue with next + } + } } - } - } - let now = Utc::now(); - if now.sub(op.queued).num_seconds() < 1 { - sleep(Duration::from_secs(1)).await; - } + // Network reconnection delay (original logic) + let now = Utc::now(); + if now.sub(op.queued).num_seconds() < 1 { + tokio::select! { + _ = cancel_token.cancelled() => { + info!("MRF heal routine cancelled during reconnection delay"); + break; + } + _ = sleep(Duration::from_secs(1)) => {} + } + } - let scan_mode = if op.bitrot_scan { HEAL_DEEP_SCAN } else { HEAL_NORMAL_SCAN }; - if op.object.is_empty() { - if let Err(err) = heal_bucket(&op.bucket).await { - error!("heal bucket failed, bucket: {}, err: {:?}", op.bucket, err); - } - } else if op.versions.is_empty() { - if let Err(err) = - heal_object(&op.bucket, &op.object, &op.version_id.clone().unwrap_or_default(), scan_mode).await - { - error!("heal object failed, bucket: {}, object: {}, err: {:?}", op.bucket, op.object, err); - } - } else { - let vers = op.versions.len() / 16; - if vers > 0 { - for i in 0..vers { - let start = i * 16; - let end = start + 16; + // Core healing logic (original logic preserved) + let scan_mode = if op.bitrot_scan { HEAL_DEEP_SCAN } else { HEAL_NORMAL_SCAN }; + + if op.object.is_empty() { + // Heal bucket (original logic) + if let Err(err) = heal_bucket(&op.bucket).await { + error!("heal bucket failed, bucket: {}, err: {:?}", op.bucket, err); + } + } else if op.versions.is_empty() { + // Heal single object (original logic) if let Err(err) = heal_object( &op.bucket, &op.object, - &Uuid::from_slice(&op.versions[start..end]).expect("").to_string(), - scan_mode, - ) - .await - { + &op.version_id.clone().unwrap_or_default(), + scan_mode + ).await { error!("heal object failed, bucket: {}, object: {}, err: {:?}", op.bucket, op.object, err); } + } else { + // Heal multiple versions (original logic) + let vers = op.versions.len() / 16; + if vers > 0 { + for i in 0..vers { + // Check for cancellation before each version + if cancel_token.is_cancelled() { + info!("MRF heal routine cancelled during version processing"); + return; + } + + let start = i * 16; + let end = start + 16; + if let Err(err) = heal_object( + &op.bucket, + &op.object, + &Uuid::from_slice(&op.versions[start..end]).expect("").to_string(), + scan_mode, + ).await { + error!("heal object failed, bucket: {}, object: {}, err: {:?}", op.bucket, op.object, err); + } + } + } } + } else { + info!("MRF heal routine channel closed, exiting"); + break; } } - } else { - return; } } + + info!("MRF heal routine stopped gracefully"); } } diff --git a/rustfs/Cargo.toml b/rustfs/Cargo.toml index 2dc43351e..31e9fab41 100644 --- a/rustfs/Cargo.toml +++ b/rustfs/Cargo.toml @@ -34,6 +34,7 @@ path = "src/main.rs" workspace = true [dependencies] +rustfs-ahm = { workspace = true } rustfs-zip = { workspace = true } rustfs-madmin = { workspace = true } rustfs-s3select-api = { workspace = true } diff --git a/rustfs/src/admin/handlers.rs b/rustfs/src/admin/handlers.rs index ecf988484..11457b814 100644 --- a/rustfs/src/admin/handlers.rs +++ b/rustfs/src/admin/handlers.rs @@ -31,6 +31,7 @@ use rustfs_ecstore::cmd::bucket_targets::{self, GLOBAL_Bucket_Target_Sys}; use rustfs_ecstore::error::StorageError; use rustfs_ecstore::global::GLOBAL_ALlHealState; use rustfs_ecstore::global::get_global_action_cred; +// use rustfs_ecstore::heal::data_usage::load_data_usage_from_backend; use rustfs_ecstore::heal::data_usage::load_data_usage_from_backend; use rustfs_ecstore::heal::heal_commands::HealOpts; use rustfs_ecstore::heal::heal_ops::new_heal_sequence; diff --git a/rustfs/src/main.rs b/rustfs/src/main.rs index 19b39ba54..07ac18790 100644 --- a/rustfs/src/main.rs +++ b/rustfs/src/main.rs @@ -42,18 +42,24 @@ use hyper_util::{ service::TowerToHyperService, }; use license::init_license; +use rustfs_ahm::scanner::data_scanner::ScannerConfig; +use rustfs_ahm::{Scanner, create_ahm_services_cancel_token, shutdown_ahm_services}; use rustfs_common::globals::set_global_addr; use rustfs_config::{DEFAULT_ACCESS_KEY, DEFAULT_SECRET_KEY, RUSTFS_TLS_CERT, RUSTFS_TLS_KEY}; use rustfs_ecstore::bucket::metadata_sys::init_bucket_metadata_sys; use rustfs_ecstore::cmd::bucket_replication::init_bucket_replication_pool; use rustfs_ecstore::config as ecconfig; use rustfs_ecstore::config::GLOBAL_ConfigSys; -use rustfs_ecstore::heal::background_heal_ops::init_auto_heal; use rustfs_ecstore::rpc::make_server; use rustfs_ecstore::store_api::BucketOptions; use rustfs_ecstore::{ - StorageAPI, endpoints::EndpointServerPools, global::set_global_rustfs_port, heal::data_scanner::init_data_scanner, - notification_sys::new_global_notification_sys, set_global_endpoints, store::ECStore, store::init_local_disks, + StorageAPI, + endpoints::EndpointServerPools, + global::{set_global_rustfs_port, shutdown_background_services}, + notification_sys::new_global_notification_sys, + set_global_endpoints, + store::ECStore, + store::init_local_disks, update_erasure_type, }; use rustfs_iam::init_iam_sys; @@ -442,13 +448,16 @@ async fn run(opt: config::Opt) -> Result<()> { Error::other(err) })?; - // init scanner - let scanner_cancel_token = init_data_scanner().await; - // init auto heal - init_auto_heal().await; + // init scanner and auto heal with unified cancellation token + // let _background_services_cancel_token = create_background_services_cancel_token(); + // init_data_scanner().await; + // init_auto_heal().await; + let _ = create_ahm_services_cancel_token(); + let scanner = Scanner::new(Some(ScannerConfig::default())); + scanner.start().await?; + // init console configuration init_console_cfg(local_ip, server_port); - print_server_info(); init_bucket_replication_pool().await; @@ -507,11 +516,11 @@ async fn run(opt: config::Opt) -> Result<()> { match wait_for_shutdown().await { #[cfg(unix)] ShutdownSignal::CtrlC | ShutdownSignal::Sigint | ShutdownSignal::Sigterm => { - handle_shutdown(&state_manager, &shutdown_tx, &scanner_cancel_token).await; + handle_shutdown(&state_manager, &shutdown_tx).await; } #[cfg(not(unix))] ShutdownSignal::CtrlC => { - handle_shutdown(&state_manager, &shutdown_tx, &scanner_cancel_token).await; + handle_shutdown(&state_manager, &shutdown_tx).await; } } @@ -617,18 +626,18 @@ fn process_connection( } /// Handles the shutdown process of the server -async fn handle_shutdown( - state_manager: &ServiceStateManager, - shutdown_tx: &tokio::sync::broadcast::Sender<()>, - scanner_cancel_token: &tokio_util::sync::CancellationToken, -) { +async fn handle_shutdown(state_manager: &ServiceStateManager, shutdown_tx: &tokio::sync::broadcast::Sender<()>) { info!("Shutdown signal received in main thread"); // update the status to stopping first state_manager.update(ServiceState::Stopping); - // Stop data scanner gracefully - info!("Stopping data scanner..."); - scanner_cancel_token.cancel(); + // Stop background services (data scanner and auto heal) gracefully + info!("Stopping background services (data scanner and auto heal)..."); + shutdown_background_services(); + + // Stop AHM services gracefully + info!("Stopping AHM services..."); + shutdown_ahm_services(); // Stop the notification system shutdown_event_notifier().await;