From ea210d52dc8b59702822a204fa2f20d6b6fa19d0 Mon Sep 17 00:00:00 2001 From: junxiang Mu <1948535941@qq.com> Date: Mon, 21 Jul 2025 18:16:42 +0800 Subject: [PATCH] refactor(heal): unify heal request interface, add disk field, update ahm/ecstore/common for erasure set healing Signed-off-by: junxiang Mu <1948535941@qq.com> --- Cargo.lock | 6 + crates/ahm/src/heal/channel.rs | 41 +- crates/ahm/src/heal/erasure_healer.rs | 10 +- crates/ahm/src/heal/manager.rs | 6 - crates/ahm/src/heal/storage.rs | 6 +- crates/ahm/src/heal/task.rs | 36 +- crates/ahm/src/scanner/data_scanner.rs | 17 +- crates/ahm/tests/heal_integration_test.rs | 14 +- crates/common/Cargo.toml | 6 + crates/common/src/data_usage.rs | 1281 +++++++++++++++++ crates/common/src/heal_channel.rs | 232 ++- crates/common/src/lib.rs | 1 + .../bucket/lifecycle/bucket_lifecycle_ops.rs | 169 ++- .../bucket/lifecycle/tier_last_day_stats.rs | 2 +- crates/ecstore/src/bucket/metadata_sys.rs | 2 +- crates/ecstore/src/data_usage.rs | 297 ++++ crates/ecstore/src/disk/mod.rs | 40 +- crates/ecstore/src/global.rs | 4 - crates/ecstore/src/heal/mod.rs | 16 +- crates/ecstore/src/lib.rs | 1 + crates/ecstore/src/metrics_realtime.rs | 4 +- crates/ecstore/src/pools.rs | 4 +- crates/ecstore/src/rpc/peer_rest_client.rs | 35 +- crates/ecstore/src/rpc/peer_s3_client.rs | 33 +- crates/ecstore/src/rpc/tonic_service.rs | 166 +-- crates/ecstore/src/sets.rs | 38 +- crates/ecstore/src/store.rs | 274 +--- crates/ecstore/src/store_api.rs | 8 +- crates/ecstore/src/store_init.rs | 9 +- 29 files changed, 2119 insertions(+), 639 deletions(-) create mode 100644 crates/common/src/data_usage.rs create mode 100644 crates/ecstore/src/data_usage.rs diff --git a/Cargo.lock b/Cargo.lock index b94029abf..683f5a05a 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -7966,8 +7966,14 @@ dependencies = [ name = "rustfs-common" version = "0.0.5" dependencies = [ + "async-trait", "chrono", + "path-clean", + "rmp-serde", + "rustfs-filemeta", "rustfs-madmin", + "s3s", + "serde", "tokio", "tonic", "uuid", diff --git a/crates/ahm/src/heal/channel.rs b/crates/ahm/src/heal/channel.rs index f9b59652b..ecfeae783 100644 --- a/crates/ahm/src/heal/channel.rs +++ b/crates/ahm/src/heal/channel.rs @@ -19,7 +19,7 @@ use crate::heal::{ }; use rustfs_common::heal_channel::{ - HealChannelCommand, HealChannelPriority, HealChannelReceiver, HealChannelRequest, HealChannelResponse, + HealChannelCommand, HealChannelPriority, HealChannelReceiver, HealChannelRequest, HealChannelResponse, HealScanMode, }; use std::sync::Arc; use tokio::sync::mpsc; @@ -173,15 +173,27 @@ impl HealChannelProcessor { /// Convert channel request to heal request fn convert_to_heal_request(&self, request: HealChannelRequest) -> Result { - let heal_type = match &request.object_prefix { - Some(prefix) if !prefix.is_empty() => HealType::Object { + let heal_type = if let Some(disk_id) = &request.disk { + HealType::ErasureSet { + buckets: vec![], + set_disk_id: disk_id.clone(), + } + } else if let Some(prefix) = &request.object_prefix { + if !prefix.is_empty() { + HealType::Object { + bucket: request.bucket.clone(), + object: prefix.clone(), + version_id: None, + } + } else { + HealType::Bucket { + bucket: request.bucket.clone(), + } + } + } else { + HealType::Bucket { bucket: request.bucket.clone(), - object: prefix.clone(), - version_id: None, - }, - _ => HealType::Bucket { - bucket: request.bucket.clone(), - }, + } }; let priority = match request.priority { @@ -191,18 +203,9 @@ impl HealChannelProcessor { HealChannelPriority::Critical => HealPriority::Urgent, }; - // Convert scan mode - let scan_mode = match request.scan_mode { - Some(rustfs_common::heal_channel::HealChannelScanMode::Normal) => { - rustfs_ecstore::heal::heal_commands::HEAL_NORMAL_SCAN - } - Some(rustfs_common::heal_channel::HealChannelScanMode::Deep) => rustfs_ecstore::heal::heal_commands::HEAL_DEEP_SCAN, - None => rustfs_ecstore::heal::heal_commands::HEAL_NORMAL_SCAN, - }; - // Build HealOptions with all available fields let mut options = HealOptions { - scan_mode, + scan_mode: request.scan_mode.unwrap_or(HealScanMode::Normal), remove_corrupted: request.remove_corrupted.unwrap_or(false), recreate_missing: request.recreate_missing.unwrap_or(true), update_parity: request.update_parity.unwrap_or(true), diff --git a/crates/ahm/src/heal/erasure_healer.rs b/crates/ahm/src/heal/erasure_healer.rs index 210648c41..f60d4afba 100644 --- a/crates/ahm/src/heal/erasure_healer.rs +++ b/crates/ahm/src/heal/erasure_healer.rs @@ -19,10 +19,8 @@ use crate::heal::{ storage::HealStorageAPI, }; use futures::future::join_all; -use rustfs_ecstore::{ - disk::DiskStore, - heal::heal_commands::{HealOpts, HEAL_NORMAL_SCAN}, -}; +use rustfs_common::heal_channel::{HealOpts, HealScanMode}; +use rustfs_ecstore::disk::DiskStore; use std::sync::Arc; use tokio::sync::RwLock; use tracing::{error, info, warn}; @@ -252,7 +250,7 @@ impl ErasureSetHealer { // heal object let heal_opts = HealOpts { - scan_mode: HEAL_NORMAL_SCAN, + scan_mode: HealScanMode::Normal, remove: true, recreate: true, ..Default::default() @@ -361,7 +359,7 @@ impl ErasureSetHealer { // 4. heal objects concurrently let heal_opts = HealOpts { - scan_mode: HEAL_NORMAL_SCAN, + scan_mode: HealScanMode::Normal, remove: true, // remove corrupted data recreate: true, // recreate missing data ..Default::default() diff --git a/crates/ahm/src/heal/manager.rs b/crates/ahm/src/heal/manager.rs index 180eaf665..358ad3668 100644 --- a/crates/ahm/src/heal/manager.rs +++ b/crates/ahm/src/heal/manager.rs @@ -288,12 +288,6 @@ impl HealManager { continue; } } - // disk currently healing and not finished - if let Some(h) = disk.healing().await { - if !h.finished { - endpoints.push(disk.endpoint()); - } - } } } diff --git a/crates/ahm/src/heal/storage.rs b/crates/ahm/src/heal/storage.rs index aa204307d..dcdf9781a 100644 --- a/crates/ahm/src/heal/storage.rs +++ b/crates/ahm/src/heal/storage.rs @@ -14,9 +14,9 @@ use crate::error::{Error, Result}; use async_trait::async_trait; +use rustfs_common::heal_channel::{HealOpts, HealScanMode}; use rustfs_ecstore::{ disk::{endpoint::Endpoint, DiskStore}, - heal::heal_commands::{HealOpts, HEAL_DEEP_SCAN, HEAL_NORMAL_SCAN}, store::ECStore, store_api::{BucketInfo, ObjectIO, StorageAPI}, }; @@ -238,7 +238,7 @@ impl HealStorageAPI for ECStoreHealStorage { dry_run: false, remove: false, recreate: true, - scan_mode: HEAL_DEEP_SCAN, + scan_mode: HealScanMode::Deep, update_parity: true, no_lock: false, pool: None, @@ -322,7 +322,7 @@ impl HealStorageAPI for ECStoreHealStorage { dry_run: false, remove: false, recreate: false, - scan_mode: HEAL_NORMAL_SCAN, + scan_mode: HealScanMode::Normal, update_parity: false, no_lock: false, pool: None, diff --git a/crates/ahm/src/heal/task.rs b/crates/ahm/src/heal/task.rs index 8e9d856d7..cf929b44e 100644 --- a/crates/ahm/src/heal/task.rs +++ b/crates/ahm/src/heal/task.rs @@ -13,9 +13,9 @@ // limitations under the License. use crate::error::{Error, Result}; -use crate::heal::{erasure_healer::ErasureSetHealer, progress::HealProgress, storage::HealStorageAPI}; -use rustfs_ecstore::heal::heal_commands::HealScanMode; -use rustfs_ecstore::heal::heal_commands::HEAL_NORMAL_SCAN; +use crate::heal::ErasureSetHealer; +use crate::heal::{progress::HealProgress, storage::HealStorageAPI}; +use rustfs_common::heal_channel::{HealOpts, HealScanMode}; use serde::{Deserialize, Serialize}; use std::sync::Arc; use std::time::{Duration, SystemTime}; @@ -93,7 +93,7 @@ pub struct HealOptions { impl Default for HealOptions { fn default() -> Self { Self { - scan_mode: HEAL_NORMAL_SCAN, + scan_mode: HealScanMode::Normal, remove_corrupted: false, recreate_missing: true, update_parity: true, @@ -324,7 +324,7 @@ impl HealTask { // Step 2: directly call ecstore to perform heal info!("Step 2: Performing heal using ecstore"); - let heal_opts = rustfs_ecstore::heal::heal_commands::HealOpts { + let heal_opts = HealOpts { recursive: self.options.recursive, dry_run: self.options.dry_run, remove: self.options.remove_corrupted, @@ -410,12 +410,12 @@ impl HealTask { info!("Attempting to recreate missing object: {}/{}", bucket, object); // Use ecstore's heal_object with recreate option - let heal_opts = rustfs_ecstore::heal::heal_commands::HealOpts { + let heal_opts = HealOpts { recursive: false, dry_run: self.options.dry_run, remove: false, recreate: true, - scan_mode: rustfs_ecstore::heal::heal_commands::HEAL_DEEP_SCAN, + scan_mode: HealScanMode::Deep, update_parity: true, no_lock: false, pool: None, @@ -476,7 +476,7 @@ impl HealTask { // Step 2: Perform bucket heal using ecstore info!("Step 2: Performing bucket heal using ecstore"); - let heal_opts = rustfs_ecstore::heal::heal_commands::HealOpts { + let heal_opts = HealOpts { recursive: self.options.recursive, dry_run: self.options.dry_run, remove: self.options.remove_corrupted, @@ -538,12 +538,12 @@ impl HealTask { // Step 2: Perform metadata heal using ecstore info!("Step 2: Performing metadata heal using ecstore"); - let heal_opts = rustfs_ecstore::heal::heal_commands::HealOpts { + let heal_opts = HealOpts { recursive: false, dry_run: self.options.dry_run, remove: false, recreate: false, - scan_mode: rustfs_ecstore::heal::heal_commands::HEAL_DEEP_SCAN, + scan_mode: HealScanMode::Deep, update_parity: false, no_lock: false, pool: self.options.pool_index, @@ -612,12 +612,12 @@ impl HealTask { // Step 1: Perform MRF heal using ecstore info!("Step 1: Performing MRF heal using ecstore"); - let heal_opts = rustfs_ecstore::heal::heal_commands::HealOpts { + let heal_opts = HealOpts { recursive: true, dry_run: self.options.dry_run, remove: self.options.remove_corrupted, recreate: self.options.recreate_missing, - scan_mode: rustfs_ecstore::heal::heal_commands::HEAL_DEEP_SCAN, + scan_mode: HealScanMode::Deep, update_parity: true, no_lock: false, pool: None, @@ -685,12 +685,12 @@ impl HealTask { // Step 2: Perform EC decode heal using ecstore info!("Step 2: Performing EC decode heal using ecstore"); - let heal_opts = rustfs_ecstore::heal::heal_commands::HealOpts { + let heal_opts = HealOpts { recursive: false, dry_run: self.options.dry_run, remove: false, recreate: true, - scan_mode: rustfs_ecstore::heal::heal_commands::HEAL_DEEP_SCAN, + scan_mode: HealScanMode::Deep, update_parity: true, no_lock: false, pool: None, @@ -748,6 +748,14 @@ impl HealTask { progress.update_progress(0, 4, 0, 0); } + let buckets = if buckets.is_empty() { + info!("No buckets specified, listing all buckets"); + let bucket_infos = self.storage.list_buckets().await?; + bucket_infos.into_iter().map(|info| info.name).collect() + } else { + buckets + }; + // Step 1: Perform disk format heal using ecstore info!("Step 1: Performing disk format heal using ecstore"); match self.storage.heal_format(self.options.dry_run).await { diff --git a/crates/ahm/src/scanner/data_scanner.rs b/crates/ahm/src/scanner/data_scanner.rs index a98e1f9f8..037879268 100644 --- a/crates/ahm/src/scanner/data_scanner.rs +++ b/crates/ahm/src/scanner/data_scanner.rs @@ -22,22 +22,22 @@ use ecstore::{ disk::{DiskAPI, DiskStore, WalkDirOptions}, set_disk::SetDisks, }; -use rustfs_ecstore::{self as ecstore, StorageAPI}; +use rustfs_ecstore::{self as ecstore, data_usage::store_data_usage_in_backend, StorageAPI}; use rustfs_filemeta::MetacacheReader; use tokio::sync::{Mutex, RwLock}; use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, warn}; -use super::{ - data_usage::DataUsageInfo, - metrics::{BucketMetrics, DiskMetrics, MetricsCollector, ScannerMetrics}, -}; +use super::metrics::{BucketMetrics, DiskMetrics, MetricsCollector, ScannerMetrics}; use crate::heal::HealManager; use crate::{ error::{Error, Result}, get_ahm_services_cancel_token, HealRequest, }; -use rustfs_common::metrics::{globalMetrics, Metric, Metrics}; +use rustfs_common::{ + data_usage::DataUsageInfo, + metrics::{globalMetrics, Metric, Metrics}, +}; use rustfs_ecstore::disk::RUSTFS_META_BUCKET; @@ -1182,7 +1182,7 @@ impl Scanner { // 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 { + if let Err(e) = store_data_usage_in_backend(data_clone, store).await { error!("Failed to store data usage statistics to backend: {}", e); } else { info!("Successfully stored data usage statistics to backend"); @@ -1214,6 +1214,7 @@ impl Scanner { #[cfg(test)] mod tests { use super::*; + use rustfs_ecstore::data_usage::load_data_usage_from_backend; use rustfs_ecstore::disk::endpoint::Endpoint; use rustfs_ecstore::endpoints::{EndpointServerPools, Endpoints, PoolEndpoints}; use rustfs_ecstore::store::ECStore; @@ -1441,7 +1442,7 @@ mod tests { // 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()) + let persisted = load_data_usage_from_backend(ecstore.clone()) .await .expect("load persisted usage"); assert_eq!(persisted.objects_total_count, du_after.objects_total_count); diff --git a/crates/ahm/tests/heal_integration_test.rs b/crates/ahm/tests/heal_integration_test.rs index f5dbb0c76..edda32dc4 100644 --- a/crates/ahm/tests/heal_integration_test.rs +++ b/crates/ahm/tests/heal_integration_test.rs @@ -3,10 +3,10 @@ use rustfs_ahm::heal::{ storage::{ECStoreHealStorage, HealStorageAPI}, task::{HealOptions, HealPriority, HealRequest, HealTaskStatus, HealType}, }; +use rustfs_common::heal_channel::{HealOpts, HealScanMode}; use rustfs_ecstore::{ disk::endpoint::Endpoint, endpoints::{EndpointServerPools, Endpoints, PoolEndpoints}, - heal::heal_commands::HEAL_NORMAL_SCAN, store::ECStore, store_api::{ObjectIO, ObjectOptions, PutObjReader, StorageAPI}, }; @@ -175,7 +175,7 @@ async fn test_heal_object_basic() { recursive: false, remove_corrupted: false, recreate_missing: true, - scan_mode: HEAL_NORMAL_SCAN, + scan_mode: HealScanMode::Normal, update_parity: true, timeout: Some(Duration::from_secs(300)), pool_index: None, @@ -240,7 +240,7 @@ async fn test_heal_bucket_basic() { recursive: true, remove_corrupted: false, recreate_missing: false, - scan_mode: HEAL_NORMAL_SCAN, + scan_mode: HealScanMode::Normal, update_parity: false, timeout: Some(Duration::from_secs(300)), pool_index: None, @@ -367,12 +367,12 @@ async fn test_heal_storage_api_direct() { let bucket_name = "test-bucket-direct"; create_test_bucket(&ecstore, bucket_name).await; - let heal_opts = rustfs_ecstore::heal::heal_commands::HealOpts { + let heal_opts = HealOpts { recursive: true, dry_run: true, remove: false, recreate: false, - scan_mode: rustfs_ecstore::heal::heal_commands::HEAL_NORMAL_SCAN, + scan_mode: HealScanMode::Normal, update_parity: false, no_lock: false, pool: None, @@ -388,12 +388,12 @@ async fn test_heal_storage_api_direct() { let test_data = b"Test data for direct heal API"; upload_test_object(&ecstore, bucket_name, object_name, test_data).await; - let object_heal_opts = rustfs_ecstore::heal::heal_commands::HealOpts { + let object_heal_opts = HealOpts { recursive: false, dry_run: true, remove: false, recreate: false, - scan_mode: rustfs_ecstore::heal::heal_commands::HEAL_NORMAL_SCAN, + scan_mode: HealScanMode::Normal, update_parity: false, no_lock: false, pool: None, diff --git a/crates/common/Cargo.toml b/crates/common/Cargo.toml index daba9456e..88c8a3f4b 100644 --- a/crates/common/Cargo.toml +++ b/crates/common/Cargo.toml @@ -33,3 +33,9 @@ tonic = { workspace = true } uuid = { workspace = true } chrono = { workspace = true } rustfs-madmin = { workspace = true } +rustfs-filemeta = { workspace = true } +serde = { workspace = true } +path-clean = { workspace = true } +rmp-serde = { workspace = true } +async-trait = { workspace = true } +s3s = { workspace = true } diff --git a/crates/common/src/data_usage.rs b/crates/common/src/data_usage.rs new file mode 100644 index 000000000..b9b93b690 --- /dev/null +++ b/crates/common/src/data_usage.rs @@ -0,0 +1,1281 @@ +// 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 path_clean::PathClean; +use serde::{Deserialize, Serialize}; +use std::hash::{DefaultHasher, Hash, Hasher}; +use std::path::Path; +use std::{ + collections::{HashMap, HashSet}, + time::SystemTime, +}; + +#[derive(Clone, Copy, Default, Debug, Serialize, Deserialize, PartialEq)] +pub struct TierStats { + pub total_size: u64, + pub num_versions: i32, + pub num_objects: i32, +} + +impl TierStats { + pub fn add(&self, u: &TierStats) -> TierStats { + TierStats { + total_size: self.total_size + u.total_size, + num_versions: self.num_versions + u.num_versions, + num_objects: self.num_objects + u.num_objects, + } + } +} + +#[derive(Clone, Debug, Default, Serialize, Deserialize, PartialEq)] +pub struct AllTierStats { + pub tiers: HashMap, +} + +impl AllTierStats { + pub fn new() -> Self { + Self { tiers: HashMap::new() } + } + + pub fn add_sizes(&mut self, tiers: HashMap) { + for (tier, st) in tiers { + self.tiers + .insert(tier.clone(), self.tiers.get(&tier).unwrap_or(&TierStats::default()).add(&st)); + } + } + + pub fn merge(&mut self, other: AllTierStats) { + for (tier, st) in other.tiers { + self.tiers + .insert(tier.clone(), self.tiers.get(&tier).unwrap_or(&TierStats::default()).add(&st)); + } + } + + pub fn populate_stats(&self, stats: &mut HashMap) { + for (tier, st) in &self.tiers { + stats.insert( + tier.clone(), + TierStats { + total_size: st.total_size, + num_versions: st.num_versions, + num_objects: st.num_objects, + }, + ); + } + } +} + +/// 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, +} + +// ===== 缓存相关数据结构 ===== + +/// Data usage hash for path-based caching +#[derive(Clone, Debug, Default, Eq, PartialEq)] +pub struct DataUsageHash(pub String); + +impl DataUsageHash { + pub fn string(&self) -> String { + self.0.clone() + } + + pub fn key(&self) -> String { + self.0.clone() + } + + pub fn mod_(&self, cycle: u32, cycles: u32) -> bool { + if cycles <= 1 { + return cycles == 1; + } + + let hash = self.calculate_hash(); + hash as u32 % cycles == cycle % cycles + } + + pub fn mod_alt(&self, cycle: u32, cycles: u32) -> bool { + if cycles <= 1 { + return cycles == 1; + } + + let hash = self.calculate_hash(); + (hash >> 32) as u32 % cycles == cycle % cycles + } + + fn calculate_hash(&self) -> u64 { + let mut hasher = DefaultHasher::new(); + self.0.hash(&mut hasher); + hasher.finish() + } +} + +/// Data usage hash map type +pub type DataUsageHashMap = HashSet; + +/// Size histogram for object size distribution +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct SizeHistogram(Vec); + +impl Default for SizeHistogram { + fn default() -> Self { + Self(vec![0; 11]) // DATA_USAGE_BUCKET_LEN = 11 + } +} + +impl SizeHistogram { + pub fn add(&mut self, size: u64) { + let intervals = [ + (0, 1024), // LESS_THAN_1024_B + (1024, 64 * 1024 - 1), // BETWEEN_1024_B_AND_64_KB + (64 * 1024, 256 * 1024 - 1), // BETWEEN_64_KB_AND_256_KB + (256 * 1024, 512 * 1024 - 1), // BETWEEN_256_KB_AND_512_KB + (512 * 1024, 1024 * 1024 - 1), // BETWEEN_512_KB_AND_1_MB + (1024, 1024 * 1024 - 1), // BETWEEN_1024B_AND_1_MB + (1024 * 1024, 10 * 1024 * 1024 - 1), // BETWEEN_1_MB_AND_10_MB + (10 * 1024 * 1024, 64 * 1024 * 1024 - 1), // BETWEEN_10_MB_AND_64_MB + (64 * 1024 * 1024, 128 * 1024 * 1024 - 1), // BETWEEN_64_MB_AND_128_MB + (128 * 1024 * 1024, 512 * 1024 * 1024 - 1), // BETWEEN_128_MB_AND_512_MB + (512 * 1024 * 1024, u64::MAX), // GREATER_THAN_512_MB + ]; + + for (idx, (start, end)) in intervals.iter().enumerate() { + if size >= *start && size <= *end { + self.0[idx] += 1; + break; + } + } + } + + pub fn to_map(&self) -> HashMap { + let names = [ + "LESS_THAN_1024_B", + "BETWEEN_1024_B_AND_64_KB", + "BETWEEN_64_KB_AND_256_KB", + "BETWEEN_256_KB_AND_512_KB", + "BETWEEN_512_KB_AND_1_MB", + "BETWEEN_1024B_AND_1_MB", + "BETWEEN_1_MB_AND_10_MB", + "BETWEEN_10_MB_AND_64_MB", + "BETWEEN_64_MB_AND_128_MB", + "BETWEEN_128_MB_AND_512_MB", + "GREATER_THAN_512_MB", + ]; + + let mut res = HashMap::new(); + let mut spl_count = 0; + for (count, name) in self.0.iter().zip(names.iter()) { + if name == &"BETWEEN_1024B_AND_1_MB" { + res.insert(name.to_string(), spl_count); + } else if name.starts_with("BETWEEN_") && name.contains("_KB_") && name.contains("_MB") { + spl_count += count; + res.insert(name.to_string(), *count); + } else { + res.insert(name.to_string(), *count); + } + } + res + } +} + +/// Versions histogram for version count distribution +#[derive(Clone, Debug, Serialize, Deserialize)] +pub struct VersionsHistogram(Vec); + +impl Default for VersionsHistogram { + fn default() -> Self { + Self(vec![0; 7]) // DATA_USAGE_VERSION_LEN = 7 + } +} + +impl VersionsHistogram { + pub fn add(&mut self, count: u64) { + let intervals = [ + (0, 0), // UNVERSIONED + (1, 1), // SINGLE_VERSION + (2, 9), // BETWEEN_2_AND_10 + (10, 99), // BETWEEN_10_AND_100 + (100, 999), // BETWEEN_100_AND_1000 + (1000, 9999), // BETWEEN_1000_AND_10000 + (10000, u64::MAX), // GREATER_THAN_10000 + ]; + + for (idx, (start, end)) in intervals.iter().enumerate() { + if count >= *start && count <= *end { + self.0[idx] += 1; + break; + } + } + } + + pub fn to_map(&self) -> HashMap { + let names = [ + "UNVERSIONED", + "SINGLE_VERSION", + "BETWEEN_2_AND_10", + "BETWEEN_10_AND_100", + "BETWEEN_100_AND_1000", + "BETWEEN_1000_AND_10000", + "GREATER_THAN_10000", + ]; + + let mut res = HashMap::new(); + for (count, name) in self.0.iter().zip(names.iter()) { + res.insert(name.to_string(), *count); + } + res + } +} + +/// Replication statistics for a single target +#[derive(Debug, Default, Clone, Serialize, Deserialize)] +pub struct ReplicationStats { + pub pending_size: u64, + pub replicated_size: u64, + pub failed_size: u64, + pub failed_count: u64, + pub pending_count: u64, + pub missed_threshold_size: u64, + pub after_threshold_size: u64, + pub missed_threshold_count: u64, + pub after_threshold_count: u64, + pub replicated_count: u64, +} + +impl ReplicationStats { + pub fn empty(&self) -> bool { + self.replicated_size == 0 && self.failed_size == 0 && self.failed_count == 0 + } +} + +/// Replication statistics for all targets +#[derive(Debug, Default, Clone, Serialize, Deserialize)] +pub struct ReplicationAllStats { + pub targets: HashMap, + pub replica_size: u64, + pub replica_count: u64, +} + +impl ReplicationAllStats { + pub fn empty(&self) -> bool { + if self.replica_size != 0 && self.replica_count != 0 { + return false; + } + for (_, v) in self.targets.iter() { + if !v.empty() { + return false; + } + } + true + } +} + +/// Data usage cache entry +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub struct DataUsageEntry { + pub children: DataUsageHashMap, + // These fields do not include any children. + pub size: usize, + pub objects: usize, + pub versions: usize, + pub delete_markers: usize, + pub obj_sizes: SizeHistogram, + pub obj_versions: VersionsHistogram, + pub replication_stats: Option, + pub compacted: bool, +} + +impl DataUsageEntry { + pub fn add_child(&mut self, hash: &DataUsageHash) { + if self.children.contains(&hash.key()) { + return; + } + self.children.insert(hash.key()); + } + + pub fn add_sizes(&mut self, summary: &SizeSummary) { + self.size += summary.total_size; + self.versions += summary.versions; + self.delete_markers += summary.delete_markers; + self.obj_sizes.add(summary.total_size as u64); + self.obj_versions.add(summary.versions as u64); + + let replication_stats = if self.replication_stats.is_none() { + self.replication_stats = Some(ReplicationAllStats::default()); + self.replication_stats.as_mut().unwrap() + } else { + self.replication_stats.as_mut().unwrap() + }; + replication_stats.replica_size += summary.replica_size as u64; + replication_stats.replica_count += summary.replica_count as u64; + + for (arn, st) in &summary.repl_target_stats { + let tgt_stat = replication_stats + .targets + .entry(arn.to_string()) + .or_insert(ReplicationStats::default()); + tgt_stat.pending_size += st.pending_size as u64; + tgt_stat.failed_size += st.failed_size as u64; + tgt_stat.replicated_size += st.replicated_size as u64; + tgt_stat.replicated_count += st.replicated_count as u64; + tgt_stat.failed_count += st.failed_count as u64; + tgt_stat.pending_count += st.pending_count as u64; + } + } + + pub fn merge(&mut self, other: &DataUsageEntry) { + self.objects += other.objects; + self.versions += other.versions; + self.delete_markers += other.delete_markers; + self.size += other.size; + + if let Some(o_rep) = &other.replication_stats { + if self.replication_stats.is_none() { + self.replication_stats = Some(ReplicationAllStats::default()); + } + let s_rep = self.replication_stats.as_mut().unwrap(); + s_rep.targets.clear(); + s_rep.replica_size += o_rep.replica_size; + s_rep.replica_count += o_rep.replica_count; + for (arn, stat) in o_rep.targets.iter() { + let st = s_rep.targets.entry(arn.clone()).or_default(); + *st = ReplicationStats { + pending_size: stat.pending_size + st.pending_size, + failed_size: stat.failed_size + st.failed_size, + replicated_size: stat.replicated_size + st.replicated_size, + pending_count: stat.pending_count + st.pending_count, + failed_count: stat.failed_count + st.failed_count, + replicated_count: stat.replicated_count + st.replicated_count, + ..Default::default() + }; + } + } + + for (i, v) in other.obj_sizes.0.iter().enumerate() { + self.obj_sizes.0[i] += v; + } + + for (i, v) in other.obj_versions.0.iter().enumerate() { + self.obj_versions.0[i] += v; + } + } +} + +/// Data usage cache info +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub struct DataUsageCacheInfo { + pub name: String, + pub next_cycle: u32, + pub last_update: Option, + pub skip_healing: bool, +} + +/// Data usage cache +#[derive(Clone, Debug, Default, Serialize, Deserialize)] +pub struct DataUsageCache { + pub info: DataUsageCacheInfo, + pub cache: HashMap, +} + +impl DataUsageCache { + pub fn replace(&mut self, path: &str, parent: &str, e: DataUsageEntry) { + let hash = hash_path(path); + self.cache.insert(hash.key(), e); + if !parent.is_empty() { + let phash = hash_path(parent); + let p = { + let p = self.cache.entry(phash.key()).or_default(); + p.add_child(&hash); + p.clone() + }; + self.cache.insert(phash.key(), p); + } + } + + pub fn replace_hashed(&mut self, hash: &DataUsageHash, parent: &Option, e: &DataUsageEntry) { + self.cache.insert(hash.key(), e.clone()); + if let Some(parent) = parent { + self.cache.entry(parent.key()).or_default().add_child(hash); + } + } + + pub fn find(&self, path: &str) -> Option { + self.cache.get(&hash_path(path).key()).cloned() + } + + pub fn find_children_copy(&mut self, h: DataUsageHash) -> DataUsageHashMap { + self.cache.entry(h.string()).or_default().children.clone() + } + + pub fn flatten(&self, root: &DataUsageEntry) -> DataUsageEntry { + let mut root = root.clone(); + for id in root.children.clone().iter() { + if let Some(e) = self.cache.get(id) { + let mut e = e.clone(); + if !e.children.is_empty() { + e = self.flatten(&e); + } + root.merge(&e); + } + } + root.children.clear(); + root + } + + pub fn copy_with_children(&mut self, src: &DataUsageCache, hash: &DataUsageHash, parent: &Option) { + if let Some(e) = src.cache.get(&hash.string()) { + self.cache.insert(hash.key(), e.clone()); + for ch in e.children.iter() { + if *ch == hash.key() { + return; + } + self.copy_with_children(src, &DataUsageHash(ch.to_string()), &Some(hash.clone())); + } + if let Some(parent) = parent { + let p = self.cache.entry(parent.key()).or_default(); + p.add_child(hash); + } + } + } + + pub fn delete_recursive(&mut self, hash: &DataUsageHash) { + let mut need_remove = Vec::new(); + if let Some(v) = self.cache.get(&hash.string()) { + for child in v.children.iter() { + need_remove.push(child.clone()); + } + } + self.cache.remove(&hash.string()); + need_remove.iter().for_each(|child| { + self.delete_recursive(&DataUsageHash(child.to_string())); + }); + } + + pub fn size_recursive(&self, path: &str) -> Option { + match self.find(path) { + Some(root) => { + if root.children.is_empty() { + return Some(root); + } + let mut flat = self.flatten(&root); + if flat.replication_stats.is_some() && flat.replication_stats.as_ref().unwrap().empty() { + flat.replication_stats = None; + } + Some(flat) + } + None => None, + } + } + + pub fn search_parent(&self, hash: &DataUsageHash) -> Option { + let want = hash.key(); + if let Some(last_index) = want.rfind('/') { + if let Some(v) = self.find(&want[0..last_index]) { + if v.children.contains(&want) { + let found = hash_path(&want[0..last_index]); + return Some(found); + } + } + } + + for (k, v) in self.cache.iter() { + if v.children.contains(&want) { + let found = DataUsageHash(k.clone()); + return Some(found); + } + } + None + } + + pub fn is_compacted(&self, hash: &DataUsageHash) -> bool { + match self.cache.get(&hash.key()) { + Some(due) => due.compacted, + None => false, + } + } + + pub fn force_compact(&mut self, limit: usize) { + if self.cache.len() < limit { + return; + } + let top = hash_path(&self.info.name).key(); + let top_e = match self.find(&top) { + Some(e) => e, + None => return, + }; + // Note: DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS constant would need to be passed as parameter + // or defined in common crate if needed + if top_e.children.len() > 250_000 { + // DATA_SCANNER_FORCE_COMPACT_AT_FOLDERS + self.reduce_children_of(&hash_path(&self.info.name), limit, true); + } + if self.cache.len() <= limit { + return; + } + + let mut found = HashSet::new(); + found.insert(top); + mark(self, &top_e, &mut found); + self.cache.retain(|k, _| { + if !found.contains(k) { + return false; + } + true + }); + } + + pub fn reduce_children_of(&mut self, path: &DataUsageHash, limit: usize, compact_self: bool) { + let e = match self.cache.get(&path.key()) { + Some(e) => e, + None => return, + }; + + if e.compacted { + return; + } + + if e.children.len() > limit && compact_self { + let mut flat = self.size_recursive(&path.key()).unwrap_or_default(); + flat.compacted = true; + self.delete_recursive(path); + self.replace_hashed(path, &None, &flat); + return; + } + let total = self.total_children_rec(&path.key()); + if total < limit { + return; + } + + let mut leaves = Vec::new(); + let mut remove = total - limit; + add(self, path, &mut leaves); + leaves.sort_by(|a, b| a.objects.cmp(&b.objects)); + + while remove > 0 && !leaves.is_empty() { + let e = leaves.first().unwrap(); + let candidate = e.path.clone(); + if candidate == *path && !compact_self { + break; + } + let removing = self.total_children_rec(&candidate.key()); + let mut flat = match self.size_recursive(&candidate.key()) { + Some(flat) => flat, + None => { + leaves.remove(0); + continue; + } + }; + + flat.compacted = true; + self.delete_recursive(&candidate); + self.replace_hashed(&candidate, &None, &flat); + + remove -= removing; + leaves.remove(0); + } + } + + pub fn total_children_rec(&self, path: &str) -> usize { + let root = self.find(path); + + if root.is_none() { + return 0; + } + let root = root.unwrap(); + if root.children.is_empty() { + return 0; + } + + let mut n = root.children.len(); + for ch in root.children.iter() { + n += self.total_children_rec(ch); + } + n + } + + pub fn merge(&mut self, o: &DataUsageCache) { + let mut existing_root = self.root(); + let other_root = o.root(); + if existing_root.is_none() && other_root.is_none() { + return; + } + if other_root.is_none() { + return; + } + if existing_root.is_none() { + *self = o.clone(); + return; + } + if o.info.last_update.gt(&self.info.last_update) { + self.info.last_update = o.info.last_update; + } + + existing_root.as_mut().unwrap().merge(other_root.as_ref().unwrap()); + self.cache.insert(hash_path(&self.info.name).key(), existing_root.unwrap()); + let e_hash = self.root_hash(); + for key in other_root.as_ref().unwrap().children.iter() { + let entry = &o.cache[key]; + let flat = o.flatten(entry); + let mut existing = self.cache[key].clone(); + existing.merge(&flat); + self.replace_hashed(&DataUsageHash(key.clone()), &Some(e_hash.clone()), &existing); + } + } + + pub fn root_hash(&self) -> DataUsageHash { + hash_path(&self.info.name) + } + + pub fn root(&self) -> Option { + self.find(&self.info.name) + } + + /// Convert cache to DataUsageInfo for a specific path + pub fn dui(&self, path: &str, buckets: &[String]) -> DataUsageInfo { + let e = match self.find(path) { + Some(e) => e, + None => return DataUsageInfo::default(), + }; + let flat = self.flatten(&e); + + let mut buckets_usage = HashMap::new(); + for bucket_name in buckets.iter() { + let e = match self.find(bucket_name) { + Some(e) => e, + None => continue, + }; + let flat = self.flatten(&e); + let mut bui = BucketUsageInfo { + size: flat.size as u64, + versions_count: flat.versions as u64, + objects_count: flat.objects as u64, + delete_markers_count: flat.delete_markers as u64, + object_size_histogram: flat.obj_sizes.to_map(), + object_versions_histogram: flat.obj_versions.to_map(), + ..Default::default() + }; + + if let Some(rs) = &flat.replication_stats { + bui.replica_size = rs.replica_size; + bui.replica_count = rs.replica_count; + + for (arn, stat) in rs.targets.iter() { + bui.replication_info.insert( + arn.clone(), + BucketTargetUsageInfo { + replication_pending_size: stat.pending_size, + replicated_size: stat.replicated_size, + replication_failed_size: stat.failed_size, + replication_pending_count: stat.pending_count, + replication_failed_count: stat.failed_count, + replicated_count: stat.replicated_count, + ..Default::default() + }, + ); + } + } + buckets_usage.insert(bucket_name.clone(), bui); + } + + DataUsageInfo { + last_update: self.info.last_update, + objects_total_count: flat.objects as u64, + versions_total_count: flat.versions as u64, + delete_markers_total_count: flat.delete_markers as u64, + objects_total_size: flat.size as u64, + buckets_count: e.children.len() as u64, + buckets_usage, + ..Default::default() + } + } + + pub fn marshal_msg(&self) -> Result, Box> { + let mut buf = Vec::new(); + self.serialize(&mut rmp_serde::Serializer::new(&mut buf))?; + Ok(buf) + } + + pub fn unmarshal(buf: &[u8]) -> Result> { + let t: Self = rmp_serde::from_slice(buf)?; + Ok(t) + } + + // Note: load and save methods are storage-specific and should be implemented + // in the ecstore crate where storage access is available +} + +/// Trait for storage-specific operations on DataUsageCache +#[async_trait::async_trait] +pub trait DataUsageCacheStorage { + /// Load data usage cache from backend storage + async fn load(store: &dyn std::any::Any, name: &str) -> Result> + where + Self: Sized; + + /// Save data usage cache to backend storage + async fn save(&self, name: &str) -> Result<(), Box>; +} + +// Helper structs and functions for cache operations +#[derive(Default, Clone)] +struct Inner { + objects: usize, + path: DataUsageHash, +} + +fn add(data_usage_cache: &DataUsageCache, path: &DataUsageHash, leaves: &mut Vec) { + let e = match data_usage_cache.cache.get(&path.key()) { + Some(e) => e, + None => return, + }; + if !e.children.is_empty() { + return; + } + + let sz = data_usage_cache.size_recursive(&path.key()).unwrap_or_default(); + leaves.push(Inner { + objects: sz.objects, + path: path.clone(), + }); + for ch in e.children.iter() { + add(data_usage_cache, &DataUsageHash(ch.clone()), leaves); + } +} + +fn mark(duc: &DataUsageCache, entry: &DataUsageEntry, found: &mut HashSet) { + for k in entry.children.iter() { + found.insert(k.to_string()); + if let Some(ch) = duc.cache.get(k) { + mark(duc, ch, found); + } + } +} + +/// Hash a path for data usage caching +pub fn hash_path(data: &str) -> DataUsageHash { + DataUsageHash(Path::new(&data).clean().to_string_lossy().to_string()) +} + +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 + pub fn extract_bucket_from_path(&self, object_path: &str) -> Result> { + let parts: Vec<&str> = object_path.split('/').collect(); + if parts.is_empty() { + return Err("Invalid object path: empty".into()); + } + 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; + } + } +} + +#[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/common/src/heal_channel.rs b/crates/common/src/heal_channel.rs index 56399cb37..6988e06b8 100644 --- a/crates/common/src/heal_channel.rs +++ b/crates/common/src/heal_channel.rs @@ -12,10 +12,109 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::sync::OnceLock; +use s3s::dto::{BucketLifecycleConfiguration, ExpirationStatus, LifecycleRule, ReplicationConfiguration, ReplicationRuleStatus}; +use serde::{Deserialize, Serialize}; +use std::{ + fmt::{self, Display}, + sync::OnceLock, +}; use tokio::sync::mpsc; use uuid::Uuid; +pub const HEAL_DELETE_DANGLING: bool = true; +pub const RUSTFS_RESERVED_BUCKET: &str = "rustfs"; +pub const RUSTFS_RESERVED_BUCKET_PATH: &str = "/rustfs"; + +#[derive(Clone, Copy, Debug, Serialize, Deserialize)] +pub enum HealItemType { + Metadata, + Bucket, + BucketMetadata, + Object, +} + +impl HealItemType { + pub fn to_str(&self) -> &str { + match self { + HealItemType::Metadata => "metadata", + HealItemType::Bucket => "bucket", + HealItemType::BucketMetadata => "bucket-metadata", + HealItemType::Object => "object", + } + } +} + +impl Display for HealItemType { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.to_str()) + } +} + +#[derive(Clone, Copy, Debug, Serialize, Deserialize)] +pub enum DriveState { + Ok, + Offline, + Corrupt, + Missing, + PermissionDenied, + Faulty, + RootMount, + Unknown, + Unformatted, // only returned by disk +} + +impl DriveState { + pub fn to_str(&self) -> &str { + match self { + DriveState::Ok => "ok", + DriveState::Offline => "offline", + DriveState::Corrupt => "corrupt", + DriveState::Missing => "missing", + DriveState::PermissionDenied => "permission-denied", + DriveState::Faulty => "faulty", + DriveState::RootMount => "root-mount", + DriveState::Unknown => "unknown", + DriveState::Unformatted => "unformatted", + } + } +} + +impl Display for DriveState { + fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result { + write!(f, "{}", self.to_str()) + } +} + +#[derive(Clone, Copy, Debug, Serialize, Deserialize, PartialEq, Eq)] +pub enum HealScanMode { + Unknown, + Normal, + Deep, +} + +impl Default for HealScanMode { + fn default() -> Self { + Self::Normal + } +} + +#[derive(Clone, Copy, Debug, Default, Serialize, Deserialize)] +pub struct HealOpts { + pub recursive: bool, + #[serde(rename = "dryRun")] + pub dry_run: bool, + pub remove: bool, + pub recreate: bool, + #[serde(rename = "scanMode")] + pub scan_mode: HealScanMode, + #[serde(rename = "updateParity")] + pub update_parity: bool, + #[serde(rename = "nolock")] + pub no_lock: bool, + pub pool: Option, + pub set: Option, +} + /// Heal channel command type #[derive(Debug, Clone)] pub enum HealChannelCommand { @@ -32,6 +131,8 @@ pub enum HealChannelCommand { pub struct HealChannelRequest { /// Unique request ID pub id: String, + /// Disk ID for heal disk/erasure set task + pub disk: Option, /// Bucket name pub bucket: String, /// Object prefix (optional) @@ -45,7 +146,7 @@ pub struct HealChannelRequest { /// Set index (optional) pub set_index: Option, /// Scan mode (optional) - pub scan_mode: Option, + pub scan_mode: Option, /// Whether to remove corrupted data pub remove_corrupted: Option, /// Whether to recreate missing data @@ -164,6 +265,7 @@ pub fn create_heal_request( recursive: None, dry_run: None, timeout_seconds: None, + disk: None, } } @@ -203,11 +305,123 @@ pub fn create_heal_response( } } -/// Heal scan mode -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub enum HealChannelScanMode { - /// Normal scan - Normal, - /// Deep scan - Deep, +fn lc_get_prefix(rule: &LifecycleRule) -> String { + if let Some(p) = &rule.prefix { + return p.to_string(); + } else if let Some(filter) = &rule.filter { + if let Some(p) = &filter.prefix { + return p.to_string(); + } else if let Some(and) = &filter.and { + if let Some(p) = &and.prefix { + return p.to_string(); + } + } + } + + "".into() +} + +pub fn lc_has_active_rules(config: &BucketLifecycleConfiguration, prefix: &str) -> bool { + if config.rules.is_empty() { + return false; + } + + for rule in config.rules.iter() { + if rule.status == ExpirationStatus::from_static(ExpirationStatus::DISABLED) { + continue; + } + let rule_prefix = lc_get_prefix(rule); + if !prefix.is_empty() && !rule_prefix.is_empty() && !prefix.starts_with(&rule_prefix) && !rule_prefix.starts_with(prefix) + { + continue; + } + + if let Some(e) = &rule.noncurrent_version_expiration { + if let Some(true) = e.noncurrent_days.map(|d| d > 0) { + return true; + } + if let Some(true) = e.newer_noncurrent_versions.map(|d| d > 0) { + return true; + } + } + + if rule.noncurrent_version_transitions.is_some() { + return true; + } + if let Some(true) = rule.expiration.as_ref().map(|e| e.date.is_some()) { + return true; + } + + if let Some(true) = rule.expiration.as_ref().map(|e| e.days.is_some()) { + return true; + } + + if let Some(Some(true)) = rule.expiration.as_ref().map(|e| e.expired_object_delete_marker) { + return true; + } + + if let Some(true) = rule.transitions.as_ref().map(|t| !t.is_empty()) { + return true; + } + + if rule.transitions.is_some() { + return true; + } + } + false +} + +pub fn rep_has_active_rules(config: &ReplicationConfiguration, prefix: &str, recursive: bool) -> bool { + if config.rules.is_empty() { + return false; + } + + for rule in config.rules.iter() { + if rule + .status + .eq(&ReplicationRuleStatus::from_static(ReplicationRuleStatus::DISABLED)) + { + continue; + } + if !prefix.is_empty() { + if let Some(filter) = &rule.filter { + if let Some(r_prefix) = &filter.prefix { + if !r_prefix.is_empty() { + // incoming prefix must be in rule prefix + if !recursive && !prefix.starts_with(r_prefix) { + continue; + } + // If recursive, we can skip this rule if it doesn't match the tested prefix or level below prefix + // does not match + if recursive && !r_prefix.starts_with(prefix) && !prefix.starts_with(r_prefix) { + continue; + } + } + } + } + } + return true; + } + false +} + +pub async fn send_heal_disk(set_disk_id: String, priority: Option) -> Result<(), String> { + let req = HealChannelRequest { + id: Uuid::new_v4().to_string(), + bucket: "".to_string(), + object_prefix: None, + disk: Some(set_disk_id), + force_start: false, + priority: priority.unwrap_or_default(), + pool_index: None, + set_index: None, + scan_mode: None, + remove_corrupted: None, + recreate_missing: None, + update_parity: None, + recursive: None, + dry_run: None, + timeout_seconds: None, + }; + send_heal_request(req).await } diff --git a/crates/common/src/lib.rs b/crates/common/src/lib.rs index 3a41462a5..09dc164a3 100644 --- a/crates/common/src/lib.rs +++ b/crates/common/src/lib.rs @@ -14,6 +14,7 @@ pub mod bucket_stats; // pub mod error; +pub mod data_usage; pub mod globals; pub mod heal_channel; pub mod last_minute; diff --git a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs index 40af33df7..29b2097f6 100644 --- a/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs +++ b/crates/ecstore/src/bucket/lifecycle/bucket_lifecycle_ops.rs @@ -22,7 +22,10 @@ use async_channel::{Receiver as A_Receiver, Sender as A_Sender, bounded}; use futures::Future; use http::HeaderMap; use lazy_static::lazy_static; +use rustfs_common::data_usage::TierStats; +use rustfs_common::heal_channel::rep_has_active_rules; use rustfs_common::metrics::{IlmAction, Metrics}; +use rustfs_utils::path::encode_dir_object; use s3s::Body; use sha2::{Digest, Sha256}; use std::any::Any; @@ -32,6 +35,7 @@ use std::io::Write; use std::pin::Pin; use std::sync::atomic::{AtomicI64, Ordering}; use std::sync::{Arc, Mutex}; +use time::OffsetDateTime; use tokio::select; use tokio::sync::mpsc::{Receiver, Sender}; use tokio::sync::{RwLock, mpsc}; @@ -45,6 +49,7 @@ use super::bucket_lifecycle_audit::{LcAuditEvent, LcEventSrc}; use super::lifecycle::{self, ExpirationOptions, Lifecycle, TransitionOptions}; use super::tier_last_day_stats::{DailyAllTierStats, LastDayTierStats}; use super::tier_sweeper::{Jentry, delete_object_from_remote_tier}; +use crate::bucket::object_lock::objectlock_sys::enforce_retention_for_deletion; use crate::bucket::{metadata_sys::get_lifecycle_config, versioning_sys::BucketVersioningSys}; use crate::client::object_api_utils::new_getobjectreader; use crate::error::Error; @@ -53,15 +58,11 @@ use crate::event::name::EventName; use crate::event_notification::{EventArgs, send_event}; use crate::global::GLOBAL_LocalNodeName; use crate::global::{GLOBAL_LifecycleSys, GLOBAL_TierConfigMgr, get_global_deployment_id}; -use crate::heal::{ - data_scanner::{apply_expiry_on_non_transitioned_objects, apply_expiry_on_transitioned_object}, - data_usage_cache::TierStats, -}; use crate::store::ECStore; use crate::store_api::StorageAPI; use crate::store_api::{GetObjectReader, HTTPRangeSpec, ObjectInfo, ObjectOptions, ObjectToDelete}; use crate::tier::warm_backend::WarmBackendGetOpts; -use s3s::dto::BucketLifecycleConfiguration; +use s3s::dto::{BucketLifecycleConfiguration, DefaultRetention, ReplicationConfiguration}; pub type TimeFn = Arc Pin + Send>> + Send + Sync + 'static>; pub type TraceFn = @@ -842,3 +843,161 @@ pub struct RestoreObjectRequest { } const _MAX_RESTORE_OBJECT_REQUEST_SIZE: i64 = 2 << 20; + +pub async fn eval_action_from_lifecycle( + lc: &BucketLifecycleConfiguration, + lr: Option, + rcfg: Option<(ReplicationConfiguration, OffsetDateTime)>, + oi: &ObjectInfo, +) -> lifecycle::Event { + let event = lc.eval(&oi.to_lifecycle_opts()).await; + //if serverDebugLog { + info!("lifecycle: Secondary scan: {}", event.action); + //} + + let lock_enabled = if let Some(lr) = lr { lr.mode.is_some() } else { false }; + + match event.action { + lifecycle::IlmAction::DeleteAllVersionsAction | lifecycle::IlmAction::DelMarkerDeleteAllVersionsAction => { + if lock_enabled { + return lifecycle::Event::default(); + } + } + lifecycle::IlmAction::DeleteVersionAction | lifecycle::IlmAction::DeleteRestoredVersionAction => { + if oi.version_id.is_none() { + return lifecycle::Event::default(); + } + if lock_enabled && enforce_retention_for_deletion(oi) { + //if serverDebugLog { + if oi.version_id.is_some() { + info!("lifecycle: {} v({}) is locked, not deleting", oi.name, oi.version_id.expect("err")); + } else { + info!("lifecycle: {} is locked, not deleting", oi.name); + } + //} + return lifecycle::Event::default(); + } + if let Some(rcfg) = rcfg { + if rep_has_active_rules(&rcfg.0, &oi.name, true) { + return lifecycle::Event::default(); + } + } + } + _ => (), + } + + event +} + +async fn apply_transition_rule(event: &lifecycle::Event, src: &LcEventSrc, oi: &ObjectInfo) -> bool { + if oi.delete_marker || oi.is_dir { + return false; + } + GLOBAL_TransitionState.queue_transition_task(oi, event, src).await; + true +} + +pub async fn apply_expiry_on_transitioned_object( + api: Arc, + oi: &ObjectInfo, + lc_event: &lifecycle::Event, + src: &LcEventSrc, +) -> bool { + // let time_ilm = ScannerMetrics::time_ilm(lc_event.action.clone()); + if let Err(_err) = expire_transitioned_object(api, oi, lc_event, src).await { + return false; + } + // let _ = time_ilm(1); + + true +} + +pub async fn apply_expiry_on_non_transitioned_objects( + api: Arc, + oi: &ObjectInfo, + lc_event: &lifecycle::Event, + _src: &LcEventSrc, +) -> bool { + let mut opts = ObjectOptions { + expiration: ExpirationOptions { expire: true }, + ..Default::default() + }; + + if lc_event.action.delete_versioned() { + opts.version_id = Some(oi.version_id.expect("err").to_string()); + } + + opts.versioned = BucketVersioningSys::prefix_enabled(&oi.bucket, &oi.name).await; + opts.version_suspended = BucketVersioningSys::prefix_suspended(&oi.bucket, &oi.name).await; + + if lc_event.action.delete_all() { + opts.delete_prefix = true; + opts.delete_prefix_object = true; + } + + // let time_ilm = ScannerMetrics::time_ilm(lc_event.action.clone()); + + let mut dobj = api + .delete_object(&oi.bucket, &encode_dir_object(&oi.name), opts) + .await + .unwrap(); + if dobj.name.is_empty() { + dobj = oi.clone(); + } + + //let tags = LcAuditEvent::new(lc_event.clone(), src.clone()).tags(); + //tags["version-id"] = dobj.version_id; + + let mut event_name = EventName::ObjectRemovedDelete; + if oi.delete_marker { + event_name = EventName::ObjectRemovedDeleteMarkerCreated; + } + match lc_event.action { + lifecycle::IlmAction::DeleteAllVersionsAction => event_name = EventName::ObjectRemovedDeleteAllVersions, + lifecycle::IlmAction::DelMarkerDeleteAllVersionsAction => event_name = EventName::ILMDelMarkerExpirationDelete, + _ => (), + } + send_event(EventArgs { + event_name: event_name.as_ref().to_string(), + bucket_name: dobj.bucket.clone(), + object: dobj, + user_agent: "Internal: [ILM-Expiry]".to_string(), + host: GLOBAL_LocalNodeName.to_string(), + ..Default::default() + }); + + if lc_event.action != lifecycle::IlmAction::NoneAction { + // let mut num_versions = 1_u64; + // if lc_event.action.delete_all() { + // num_versions = oi.num_versions as u64; + // } + // let _ = time_ilm(num_versions); + } + + true +} + +async fn apply_expiry_rule(event: &lifecycle::Event, src: &LcEventSrc, oi: &ObjectInfo) -> bool { + let mut expiry_state = GLOBAL_ExpiryState.write().await; + expiry_state.enqueue_by_days(oi, event, src).await; + true +} + +pub async fn apply_lifecycle_action(event: &lifecycle::Event, src: &LcEventSrc, oi: &ObjectInfo) -> bool { + let mut success = false; + match event.action { + lifecycle::IlmAction::DeleteVersionAction + | lifecycle::IlmAction::DeleteAction + | lifecycle::IlmAction::DeleteRestoredAction + | lifecycle::IlmAction::DeleteRestoredVersionAction + | lifecycle::IlmAction::DeleteAllVersionsAction + | lifecycle::IlmAction::DelMarkerDeleteAllVersionsAction => { + success = apply_expiry_rule(event, src, oi).await; + } + lifecycle::IlmAction::TransitionAction | lifecycle::IlmAction::TransitionVersionAction => { + success = apply_transition_rule(event, src, oi).await; + } + _ => (), + } + success +} diff --git a/crates/ecstore/src/bucket/lifecycle/tier_last_day_stats.rs b/crates/ecstore/src/bucket/lifecycle/tier_last_day_stats.rs index 78e3e8ed0..557d6189d 100644 --- a/crates/ecstore/src/bucket/lifecycle/tier_last_day_stats.rs +++ b/crates/ecstore/src/bucket/lifecycle/tier_last_day_stats.rs @@ -25,7 +25,7 @@ use std::ops::Sub; use time::OffsetDateTime; use tracing::{error, warn}; -use crate::heal::data_usage_cache::TierStats; +use rustfs_common::data_usage::TierStats; pub type DailyAllTierStats = HashMap; diff --git a/crates/ecstore/src/bucket/metadata_sys.rs b/crates/ecstore/src/bucket/metadata_sys.rs index 791134da9..b192cd2f3 100644 --- a/crates/ecstore/src/bucket/metadata_sys.rs +++ b/crates/ecstore/src/bucket/metadata_sys.rs @@ -18,9 +18,9 @@ use crate::bucket::utils::{deserialize, is_meta_bucketname}; use crate::cmd::bucket_targets; use crate::error::{Error, Result, is_err_bucket_not_found}; use crate::global::{GLOBAL_Endpoints, is_dist_erasure, is_erasure, new_object_layer_fn}; -use crate::heal::heal_commands::HealOpts; use crate::store::ECStore; use futures::future::join_all; +use rustfs_common::heal_channel::HealOpts; use rustfs_policy::policy::BucketPolicy; use s3s::dto::{ BucketLifecycleConfiguration, NotificationConfiguration, ObjectLockConfiguration, ReplicationConfiguration, diff --git a/crates/ecstore/src/data_usage.rs b/crates/ecstore/src/data_usage.rs new file mode 100644 index 000000000..425081ef1 --- /dev/null +++ b/crates/ecstore/src/data_usage.rs @@ -0,0 +1,297 @@ +// 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}; + +use crate::{bucket::metadata_sys::get_replication_config, config::com::read_config, store::ECStore}; +use rustfs_common::data_usage::{BucketTargetUsageInfo, DataUsageCache, DataUsageEntry, DataUsageInfo, SizeSummary}; +use rustfs_utils::path::SLASH_SEPARATOR; +use tracing::{error, warn}; + +use crate::error::Error; + +// 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!("{}{}{}", + crate::disk::RUSTFS_META_BUCKET, + SLASH_SEPARATOR, + crate::disk::BUCKET_META_PREFIX + ); + pub static ref DATA_USAGE_OBJ_NAME_PATH: String = format!("{}{}{}", + crate::disk::BUCKET_META_PREFIX, + SLASH_SEPARATOR, + DATA_USAGE_OBJ_NAME + ); + pub static ref DATA_USAGE_BLOOM_NAME_PATH: String = format!("{}{}{}", + crate::disk::BUCKET_META_PREFIX, + SLASH_SEPARATOR, + DATA_USAGE_BLOOM_NAME + ); +} + +/// Store data usage info to backend storage +pub async fn store_data_usage_in_backend(data_usage_info: DataUsageInfo, store: Arc) -> Result<(), Error> { + let data = + serde_json::to_vec(&data_usage_info).map_err(|e| Error::other(format!("Failed to serialize data usage info: {e}")))?; + + // Save to backend using the same mechanism as original code + crate::config::com::save_config(store, &DATA_USAGE_OBJ_NAME_PATH, data) + .await + .map_err(Error::other)?; + + Ok(()) +} + +/// Load data usage info from backend storage +pub async fn load_data_usage_from_backend(store: Arc) -> Result { + let buf: Vec = 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 == crate::error::Error::ConfigNotFound { + return Ok(DataUsageInfo::default()); + } + return Err(Error::other(e)); + } + }; + + let mut data_usage_info: DataUsageInfo = + serde_json::from_slice(&buf).map_err(|e| Error::other(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(), + rustfs_common::data_usage::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) +} + +/// Create a data usage cache entry from size summary +pub fn create_cache_entry_from_summary(summary: &SizeSummary) -> DataUsageEntry { + let mut entry = DataUsageEntry::default(); + entry.add_sizes(summary); + entry +} + +/// Convert data usage cache to DataUsageInfo +pub fn cache_to_data_usage_info(cache: &DataUsageCache, path: &str, buckets: &[crate::store_api::BucketInfo]) -> DataUsageInfo { + let e = match cache.find(path) { + Some(e) => e, + None => return DataUsageInfo::default(), + }; + let flat = cache.flatten(&e); + + let mut buckets_usage = HashMap::new(); + for bucket in buckets.iter() { + let e = match cache.find(&bucket.name) { + Some(e) => e, + None => continue, + }; + let flat = cache.flatten(&e); + let mut bui = rustfs_common::data_usage::BucketUsageInfo { + size: flat.size as u64, + versions_count: flat.versions as u64, + objects_count: flat.objects as u64, + delete_markers_count: flat.delete_markers as u64, + object_size_histogram: flat.obj_sizes.to_map(), + object_versions_histogram: flat.obj_versions.to_map(), + ..Default::default() + }; + + if let Some(rs) = &flat.replication_stats { + bui.replica_size = rs.replica_size; + bui.replica_count = rs.replica_count; + + for (arn, stat) in rs.targets.iter() { + bui.replication_info.insert( + arn.clone(), + BucketTargetUsageInfo { + replication_pending_size: stat.pending_size, + replicated_size: stat.replicated_size, + replication_failed_size: stat.failed_size, + replication_pending_count: stat.pending_count, + replication_failed_count: stat.failed_count, + replicated_count: stat.replicated_count, + ..Default::default() + }, + ); + } + } + buckets_usage.insert(bucket.name.clone(), bui); + } + + DataUsageInfo { + last_update: cache.info.last_update, + objects_total_count: flat.objects as u64, + versions_total_count: flat.versions as u64, + delete_markers_total_count: flat.delete_markers as u64, + objects_total_size: flat.size as u64, + buckets_count: e.children.len() as u64, + buckets_usage, + ..Default::default() + } +} + +// Helper functions for DataUsageCache operations +pub async fn load_data_usage_cache(store: &crate::set_disk::SetDisks, name: &str) -> crate::error::Result { + use crate::disk::{BUCKET_META_PREFIX, RUSTFS_META_BUCKET}; + use crate::store_api::{ObjectIO, ObjectOptions}; + use http::HeaderMap; + use rand::Rng; + use std::path::Path; + use std::time::Duration; + use tokio::time::sleep; + + let mut d = DataUsageCache::default(); + let mut retries = 0; + while retries < 5 { + let path = Path::new(BUCKET_META_PREFIX).join(name); + match store + .get_object_reader( + RUSTFS_META_BUCKET, + path.to_str().unwrap(), + None, + HeaderMap::new(), + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + { + Ok(mut reader) => { + if let Ok(info) = DataUsageCache::unmarshal(&reader.read_all().await?) { + d = info + } + break; + } + Err(err) => match err { + crate::error::Error::FileNotFound | crate::error::Error::VolumeNotFound => { + match store + .get_object_reader( + RUSTFS_META_BUCKET, + name, + None, + HeaderMap::new(), + &ObjectOptions { + no_lock: true, + ..Default::default() + }, + ) + .await + { + Ok(mut reader) => { + if let Ok(info) = DataUsageCache::unmarshal(&reader.read_all().await?) { + d = info + } + break; + } + Err(_) => match err { + crate::error::Error::FileNotFound | crate::error::Error::VolumeNotFound => { + break; + } + _ => {} + }, + } + } + _ => { + break; + } + }, + } + retries += 1; + let dur = { + let mut rng = rand::rng(); + rng.random_range(0..1_000) + }; + sleep(Duration::from_millis(dur)).await; + } + Ok(d) +} + +pub async fn save_data_usage_cache(cache: &DataUsageCache, name: &str) -> crate::error::Result<()> { + use crate::config::com::save_config; + use crate::disk::BUCKET_META_PREFIX; + use crate::new_object_layer_fn; + use std::path::Path; + + let Some(store) = new_object_layer_fn() else { + return Err(crate::error::Error::other("errServerNotInitialized")); + }; + let buf = cache.marshal_msg().map_err(crate::error::Error::other)?; + let buf_clone = buf.clone(); + + let store_clone = store.clone(); + + let name = Path::new(BUCKET_META_PREFIX).join(name).to_string_lossy().to_string(); + + let name_clone = name.clone(); + tokio::spawn(async move { + let _ = save_config(store_clone, &format!("{}{}", &name_clone, ".bkp"), buf_clone).await; + }); + save_config(store, &name, buf).await?; + Ok(()) +} diff --git a/crates/ecstore/src/disk/mod.rs b/crates/ecstore/src/disk/mod.rs index 1f918ee36..f680a42db 100644 --- a/crates/ecstore/src/disk/mod.rs +++ b/crates/ecstore/src/disk/mod.rs @@ -30,11 +30,6 @@ pub const FORMAT_CONFIG_FILE: &str = "format.json"; pub const STORAGE_FORMAT_FILE: &str = "xl.meta"; pub const STORAGE_FORMAT_FILE_BACKUP: &str = "xl.meta.bkp"; -use crate::heal::{ - data_scanner::ShouldSleepFn, - data_usage_cache::{DataUsageCache, DataUsageEntry}, - heal_commands::{HealScanMode, HealingTracker}, -}; use crate::rpc::RemoteDisk; use bytes::Bytes; use endpoint::Endpoint; @@ -46,10 +41,7 @@ use rustfs_madmin::info_commands::DiskMetrics; use serde::{Deserialize, Serialize}; use std::{fmt::Debug, path::PathBuf, sync::Arc}; use time::OffsetDateTime; -use tokio::{ - io::{AsyncRead, AsyncWrite}, - sync::mpsc::Sender, -}; +use tokio::io::{AsyncRead, AsyncWrite}; use uuid::Uuid; pub type DiskStore = Arc; @@ -406,28 +398,6 @@ impl DiskAPI for Disk { Disk::Remote(remote_disk) => remote_disk.disk_info(opts).await, } } - - #[tracing::instrument(skip(self, cache, we_sleep, scan_mode))] - async fn ns_scanner( - &self, - cache: &DataUsageCache, - updates: Sender, - scan_mode: HealScanMode, - we_sleep: ShouldSleepFn, - ) -> Result { - match self { - Disk::Local(local_disk) => local_disk.ns_scanner(cache, updates, scan_mode, we_sleep).await, - Disk::Remote(remote_disk) => remote_disk.ns_scanner(cache, updates, scan_mode, we_sleep).await, - } - } - - #[tracing::instrument(skip(self))] - async fn healing(&self) -> Option { - match self { - Disk::Local(local_disk) => local_disk.healing().await, - Disk::Remote(remote_disk) => remote_disk.healing().await, - } - } } pub async fn new_disk(ep: &Endpoint, opt: &DiskOption) -> Result { @@ -527,14 +497,6 @@ pub trait DiskAPI: Debug + Send + Sync + 'static { async fn write_all(&self, volume: &str, path: &str, data: Bytes) -> Result<()>; async fn read_all(&self, volume: &str, path: &str) -> Result; async fn disk_info(&self, opts: &DiskInfoOptions) -> Result; - async fn ns_scanner( - &self, - cache: &DataUsageCache, - updates: Sender, - scan_mode: HealScanMode, - we_sleep: ShouldSleepFn, - ) -> Result; - async fn healing(&self) -> Option; } #[derive(Debug, Default, Serialize, Deserialize)] diff --git a/crates/ecstore/src/global.rs b/crates/ecstore/src/global.rs index d8411fba1..533512266 100644 --- a/crates/ecstore/src/global.rs +++ b/crates/ecstore/src/global.rs @@ -17,7 +17,6 @@ use crate::{ disk::DiskStore, endpoints::{EndpointServerPools, PoolEndpoints, SetupType}, event_notification::EventNotifier, - heal::{background_heal_ops::HealRoutine, heal_ops::AllHealState}, store::ECStore, tier::tier::TierConfigMgr, }; @@ -50,9 +49,6 @@ pub static ref GLOBAL_LOCAL_DISK_MAP: Arc> = Arc::new(RwLock::new(Vec::new())); pub static ref GLOBAL_Endpoints: OnceLock = OnceLock::new(); pub static ref GLOBAL_RootDiskThreshold: RwLock = RwLock::new(0); -pub static ref GLOBAL_BackgroundHealRoutine: Arc = HealRoutine::new(); -pub static ref GLOBAL_BackgroundHealState: Arc = AllHealState::new(false); -// pub static ref GLOBAL_MRFState: Arc = Arc::new(MRFState::new()); pub static ref GLOBAL_TierConfigMgr: Arc> = TierConfigMgr::new(); pub static ref GLOBAL_LifecycleSys: Arc = LifecycleSys::new(); pub static ref GLOBAL_EventNotifier: Arc> = EventNotifier::new(); diff --git a/crates/ecstore/src/heal/mod.rs b/crates/ecstore/src/heal/mod.rs index 0733d91d5..6b34dc620 100644 --- a/crates/ecstore/src/heal/mod.rs +++ b/crates/ecstore/src/heal/mod.rs @@ -12,12 +12,12 @@ // See the License for the specific language governing permissions and // limitations under the License. -pub mod background_heal_ops; -pub mod data_scanner; +// pub mod background_heal_ops; +// pub mod data_scanner; // pub mod data_scanner_metric; -pub mod data_usage; -pub mod data_usage_cache; -pub mod error; -pub mod heal_commands; -pub mod heal_ops; -pub mod mrf; +// pub mod data_usage; +// pub mod data_usage_cache; +// pub mod error; +// pub mod heal_commands; +// pub mod heal_ops; +// pub mod mrf; diff --git a/crates/ecstore/src/lib.rs b/crates/ecstore/src/lib.rs index daf032595..177618892 100644 --- a/crates/ecstore/src/lib.rs +++ b/crates/ecstore/src/lib.rs @@ -23,6 +23,7 @@ mod chunk_stream; pub mod cmd; pub mod compress; pub mod config; +pub mod data_usage; pub mod disk; pub mod disks_layout; pub mod endpoints; diff --git a/crates/ecstore/src/metrics_realtime.rs b/crates/ecstore/src/metrics_realtime.rs index 2363f1c0c..a5f5f3a3a 100644 --- a/crates/ecstore/src/metrics_realtime.rs +++ b/crates/ecstore/src/metrics_realtime.rs @@ -17,6 +17,7 @@ use std::collections::{HashMap, HashSet}; use chrono::Utc; use rustfs_common::{ globals::{GLOBAL_Local_Node_Name, GLOBAL_Rustfs_Addr}, + heal_channel::DriveState, metrics::globalMetrics, }; use rustfs_madmin::metrics::{DiskIOStats, DiskMetric, RealtimeMetrics}; @@ -26,7 +27,6 @@ use tracing::info; use crate::{ admin_server_info::get_local_server_property, - heal::heal_commands::{DRIVE_STATE_OK, DRIVE_STATE_UNFORMATTED}, new_object_layer_fn, store_api::StorageAPI, // utils::os::get_drive_stats, @@ -147,7 +147,7 @@ async fn collect_local_disks_metrics(disks: &HashSet) -> HashMap Result { - let mut client = node_service_time_out_client(&self.grid_host) - .await - .map_err(|err| Error::other(err.to_string()))?; - let request = Request::new(BackgroundHealStatusRequest {}); - - let response = client.background_heal_status(request).await?.into_inner(); - if !response.success { - if let Some(msg) = response.error_info { - return Err(Error::other(msg)); - } - return Err(Error::other("")); - } - let data = response.bg_heal_state; - - let mut buf = Deserializer::new(Cursor::new(data)); - let bg_heal_state: BgHealState = Deserialize::deserialize(&mut buf)?; - - Ok(bg_heal_state) - } - pub async fn get_metacache_listing(&self) -> Result<()> { let _client = node_service_time_out_client(&self.grid_host) .await diff --git a/crates/ecstore/src/rpc/peer_s3_client.rs b/crates/ecstore/src/rpc/peer_s3_client.rs index a7789aabc..10a00e279 100644 --- a/crates/ecstore/src/rpc/peer_s3_client.rs +++ b/crates/ecstore/src/rpc/peer_s3_client.rs @@ -17,10 +17,6 @@ use crate::disk::error::{Error, Result}; use crate::disk::error_reduce::{BUCKET_OP_IGNORED_ERRS, is_all_buckets_not_found, reduce_write_quorum_errs}; use crate::disk::{DiskAPI, DiskStore}; use crate::global::GLOBAL_LOCAL_DISK_MAP; -use crate::heal::heal_commands::{ - DRIVE_STATE_CORRUPT, DRIVE_STATE_MISSING, DRIVE_STATE_OFFLINE, DRIVE_STATE_OK, HEAL_ITEM_BUCKET, HealOpts, -}; -use crate::heal::heal_ops::RUSTFS_RESERVED_BUCKET; use crate::store::all_local_disk; use crate::store_utils::is_reserved_or_invalid_bucket; use crate::{ @@ -30,6 +26,7 @@ use crate::{ }; use async_trait::async_trait; use futures::future::join_all; +use rustfs_common::heal_channel::{DriveState, HealItemType, HealOpts, RUSTFS_RESERVED_BUCKET}; use rustfs_madmin::heal_commands::{HealDriveInfo, HealResultItem}; use rustfs_protos::node_service_time_out_client; use rustfs_protos::proto_gen::node_service::{ @@ -542,7 +539,7 @@ impl PeerS3Client for RemotePeerS3Client { } Ok(HealResultItem { - heal_item_type: HEAL_ITEM_BUCKET.to_string(), + heal_item_type: HealItemType::Bucket.to_string(), bucket: bucket.to_string(), set_count: 0, ..Default::default() @@ -651,13 +648,13 @@ pub async fn heal_bucket_local(bucket: &str, opts: &HealOpts) -> Result disk, None => { - bs_clone.write().await[index] = DRIVE_STATE_OFFLINE.to_string(); - as_clone.write().await[index] = DRIVE_STATE_OFFLINE.to_string(); + bs_clone.write().await[index] = DriveState::Offline.to_string(); + as_clone.write().await[index] = DriveState::Offline.to_string(); return Some(Error::DiskNotFound); } }; - bs_clone.write().await[index] = DRIVE_STATE_OK.to_string(); - as_clone.write().await[index] = DRIVE_STATE_OK.to_string(); + bs_clone.write().await[index] = DriveState::Ok.to_string(); + as_clone.write().await[index] = DriveState::Ok.to_string(); if bucket == RUSTFS_RESERVED_BUCKET { return None; @@ -667,18 +664,18 @@ pub async fn heal_bucket_local(bucket: &str, opts: &HealOpts) -> Result None, Err(err) => match err { Error::DiskNotFound => { - bs_clone.write().await[index] = DRIVE_STATE_OFFLINE.to_string(); - as_clone.write().await[index] = DRIVE_STATE_OFFLINE.to_string(); + bs_clone.write().await[index] = DriveState::Offline.to_string(); + as_clone.write().await[index] = DriveState::Offline.to_string(); Some(err) } Error::VolumeNotFound => { - bs_clone.write().await[index] = DRIVE_STATE_MISSING.to_string(); - as_clone.write().await[index] = DRIVE_STATE_MISSING.to_string(); + bs_clone.write().await[index] = DriveState::Missing.to_string(); + as_clone.write().await[index] = DriveState::Missing.to_string(); Some(err) } _ => { - bs_clone.write().await[index] = DRIVE_STATE_CORRUPT.to_string(); - as_clone.write().await[index] = DRIVE_STATE_CORRUPT.to_string(); + bs_clone.write().await[index] = DriveState::Corrupt.to_string(); + as_clone.write().await[index] = DriveState::Corrupt.to_string(); Some(err) } }, @@ -687,7 +684,7 @@ pub async fn heal_bucket_local(bucket: &str, opts: &HealOpts) -> Result Result { - as_clone.write().await[idx] = DRIVE_STATE_OK.to_string(); + as_clone.write().await[idx] = DriveState::Ok.to_string(); return None; } Err(err) => { diff --git a/crates/ecstore/src/rpc/tonic_service.rs b/crates/ecstore/src/rpc/tonic_service.rs index ffcc3c0a5..ad6d58d91 100644 --- a/crates/ecstore/src/rpc/tonic_service.rs +++ b/crates/ecstore/src/rpc/tonic_service.rs @@ -22,21 +22,16 @@ use crate::{ DeleteOptions, DiskAPI, DiskInfoOptions, DiskStore, FileInfoVersions, ReadMultipleReq, ReadOptions, UpdateMetadataOpts, error::DiskError, }, - heal::{ - data_usage_cache::DataUsageCache, - heal_commands::{HealOpts, get_local_background_heal_status}, - }, metrics_realtime::{CollectMetricsOpts, MetricType, collect_local_metrics}, new_object_layer_fn, rpc::{LocalPeerS3Client, PeerS3Client}, store::{all_local_disk_path, find_local_disk}, store_api::{BucketOptions, DeleteBucketOptions, MakeBucketOptions, StorageAPI}, }; -use futures::{Stream, StreamExt}; +use futures::Stream; use futures_util::future::join_all; -use rustfs_common::globals::GLOBAL_Local_Node_Name; -use rustfs_lock::{LockClient, LockRequest}; +use rustfs_common::{globals::GLOBAL_Local_Node_Name, heal_channel::HealOpts}; use bytes::Bytes; use rmp_serde::{Deserializer, Serializer}; @@ -1439,120 +1434,22 @@ impl Node for NodeService { } } - type NsScannerStream = ResponseStream; - async fn ns_scanner(&self, request: Request>) -> Result, Status> { - info!("ns_scanner"); - - let mut in_stream = request.into_inner(); - let (tx, rx) = mpsc::channel(10); - - tokio::spawn(async move { - match in_stream.next().await { - Some(Ok(request)) => { - if let Some(disk) = find_local_disk(&request.disk).await { - let cache = match serde_json::from_str::(&request.cache) { - Ok(cache) => cache, - Err(err) => { - tx.send(Ok(NsScannerResponse { - success: false, - update: "".to_string(), - data_usage_cache: "".to_string(), - error: Some(DiskError::other(format!("decode DataUsageCache failed: {err}")).into()), - })) - .await - .expect("working rx"); - return; - } - }; - let (updates_tx, mut updates_rx) = mpsc::channel(100); - let tx_clone = tx.clone(); - let task = tokio::spawn(async move { - loop { - match updates_rx.recv().await { - Some(update) => { - let update = serde_json::to_string(&update).expect("encode failed"); - tx_clone - .send(Ok(NsScannerResponse { - success: true, - update, - data_usage_cache: "".to_string(), - error: None, - })) - .await - .expect("working rx"); - } - None => return, - } - } - }); - let data_usage_cache = disk.ns_scanner(&cache, updates_tx, request.scan_mode as usize, None).await; - let _ = task.await; - match data_usage_cache { - Ok(data_usage_cache) => { - let data_usage_cache = serde_json::to_string(&data_usage_cache).expect("encode failed"); - tx.send(Ok(NsScannerResponse { - success: true, - update: "".to_string(), - data_usage_cache, - error: None, - })) - .await - .expect("working rx"); - } - Err(err) => { - tx.send(Ok(NsScannerResponse { - success: false, - update: "".to_string(), - data_usage_cache: "".to_string(), - error: Some(err.into()), - })) - .await - .expect("working rx"); - } - } - } else { - tx.send(Ok(NsScannerResponse { - success: false, - update: "".to_string(), - data_usage_cache: "".to_string(), - error: Some(DiskError::other("can not find disk".to_string()).into()), - })) - .await - .expect("working rx"); - } - } - _ => todo!(), - } - }); - - let out_stream = ReceiverStream::new(rx); - Ok(tonic::Response::new(Box::pin(out_stream))) - } - async fn lock(&self, request: Request) -> Result, Status> { let request = request.into_inner(); - // Parse the request to extract resource and owner - let args: LockRequest = match serde_json::from_str(&request.args) { - Ok(args) => args, - Err(err) => { - return Ok(tonic::Response::new(GenerallyLockResponse { + match &serde_json::from_str::(&request.args) { + Ok(args) => match GLOBAL_LOCAL_SERVER.write().await.lock(args).await { + Ok(result) => Ok(tonic::Response::new(GenerallyLockResponse { + success: result, + error_info: None, + })), + Err(err) => Ok(tonic::Response::new(GenerallyLockResponse { success: false, - error_info: Some(format!("can not decode args, err: {err}")), - })); - } - }; - - match self.lock_manager.acquire_exclusive(&args).await { - Ok(result) => Ok(tonic::Response::new(GenerallyLockResponse { - success: result.success, - error_info: None, - })), + error_info: Some(format!("can not lock, args: {args}, err: {err}")), + })), + }, Err(err) => Ok(tonic::Response::new(GenerallyLockResponse { success: false, - error_info: Some(format!( - "can not lock, resource: {0}, owner: {1}, err: {2}", - args.resource, args.owner, err - )), + error_info: Some(format!("can not decode args, err: {err}")), })), } } @@ -2196,28 +2093,7 @@ impl Node for NodeService { &self, _request: Request, ) -> Result, Status> { - let (state, ok) = get_local_background_heal_status().await; - if !ok { - return Ok(tonic::Response::new(BackgroundHealStatusResponse { - success: false, - bg_heal_state: Bytes::new(), - error_info: Some("errServerNotInitialized".to_string()), - })); - } - - let mut buf = Vec::new(); - if let Err(err) = state.serialize(&mut Serializer::new(&mut buf)) { - return Ok(tonic::Response::new(BackgroundHealStatusResponse { - success: false, - bg_heal_state: Bytes::new(), - error_info: Some(err.to_string()), - })); - } - Ok(tonic::Response::new(BackgroundHealStatusResponse { - success: true, - bg_heal_state: buf.into(), - error_info: None, - })) + todo!() } async fn get_metacache_listing( @@ -3412,20 +3288,6 @@ mod tests { assert!(!proc_response.proc_info.is_empty()); } - #[tokio::test] - async fn test_background_heal_status() { - let service = create_test_node_service(); - - let request = Request::new(BackgroundHealStatusRequest {}); - - let response = service.background_heal_status(request).await; - assert!(response.is_ok()); - - let heal_response = response.unwrap().into_inner(); - // May fail if heal status is not available - assert!(heal_response.success || heal_response.error_info.is_some()); - } - #[tokio::test] async fn test_reload_pool_meta() { let service = create_test_node_service(); diff --git a/crates/ecstore/src/sets.rs b/crates/ecstore/src/sets.rs index 79a35684c..bea7d9232 100644 --- a/crates/ecstore/src/sets.rs +++ b/crates/ecstore/src/sets.rs @@ -28,9 +28,6 @@ use crate::{ endpoints::{Endpoints, PoolEndpoints}, error::StorageError, global::{GLOBAL_LOCAL_DISK_SET_DRIVES, is_dist_erasure}, - heal::heal_commands::{ - DRIVE_STATE_CORRUPT, DRIVE_STATE_MISSING, DRIVE_STATE_OFFLINE, DRIVE_STATE_OK, HEAL_ITEM_METADATA, HealOpts, - }, set_disk::SetDisks, store_api::{ BucketInfo, BucketOptions, CompletePart, DeleteBucketOptions, DeletedObject, GetObjectReader, HTTPRangeSpec, @@ -41,7 +38,11 @@ use crate::{ }; use futures::future::join_all; use http::HeaderMap; -use rustfs_common::globals::GLOBAL_Local_Node_Name; +use rustfs_common::heal_channel::HealOpts; +use rustfs_common::{ + globals::GLOBAL_Local_Node_Name, + heal_channel::{DriveState, HealItemType}, +}; use rustfs_filemeta::FileInfo; use rustfs_madmin::heal_commands::{HealDriveInfo, HealResultItem}; @@ -49,7 +50,6 @@ use rustfs_utils::{crc_hash, path::path_join_buf, sip_hash}; use tokio::sync::RwLock; use uuid::Uuid; -use crate::heal::heal_ops::HealSequence; use tokio::sync::broadcast::{Receiver, Sender}; use tokio::time::Duration; use tracing::warn; @@ -787,7 +787,7 @@ impl StorageAPI for Sets { Err(err) => return Ok((HealResultItem::default(), Some(err))), }; let mut res = HealResultItem { - heal_item_type: HEAL_ITEM_METADATA.to_string(), + heal_item_type: HealItemType::Metadata.to_string(), detail: "disk-format".to_string(), disk_count: self.set_count * self.set_drive_count, set_count: self.set_count, @@ -811,7 +811,6 @@ impl StorageAPI for Sets { // return Ok((res, Some(Error::new(DiskError::CorruptedFormat)))); // } - let format_op_id = Uuid::new_v4().to_string(); let (new_format_sets, _) = new_heal_format_sets(&ref_format, self.set_count, self.set_drive_count, &formats, &errs); if !dry_run { let mut tmp_new_formats = vec![None; self.set_count * self.set_drive_count]; @@ -819,14 +818,14 @@ impl StorageAPI for Sets { for (j, fm) in set.iter().enumerate() { if let Some(fm) = fm { res.after.drives[i * self.set_drive_count + j].uuid = fm.erasure.this.to_string(); - res.after.drives[i * self.set_drive_count + j].state = DRIVE_STATE_OK.to_string(); + res.after.drives[i * self.set_drive_count + j].state = DriveState::Ok.to_string(); tmp_new_formats[i * self.set_drive_count + j] = Some(fm.clone()); } } } // Save new formats `format.json` on unformatted disks. for (fm, disk) in tmp_new_formats.iter_mut().zip(disks.iter()) { - if fm.is_some() && disk.is_some() && save_format_file(disk, fm, &format_op_id).await.is_err() { + if fm.is_some() && disk.is_some() && save_format_file(disk, fm).await.is_err() { let _ = disk.as_ref().unwrap().close().await; *fm = None; } @@ -869,17 +868,6 @@ impl StorageAPI for Sets { .await } #[tracing::instrument(skip(self))] - async fn heal_objects( - &self, - _bucket: &str, - _prefix: &str, - _opts: &HealOpts, - _hs: Arc, - _is_meta: bool, - ) -> Result<()> { - unimplemented!() - } - #[tracing::instrument(skip(self))] async fn get_pool_and_set(&self, _id: &str) -> Result<(Option, Option, Option)> { unimplemented!() } @@ -957,17 +945,17 @@ fn formats_to_drives_info(endpoints: &Endpoints, formats: &[Option], e for (index, format) in formats.iter().enumerate() { let drive = endpoints.get_string(index); let state = if format.is_some() { - DRIVE_STATE_OK + DriveState::Ok.to_string() } else if let Some(Some(err)) = errs.get(index) { if *err == DiskError::UnformattedDisk { - DRIVE_STATE_MISSING + DriveState::Missing.to_string() } else if *err == DiskError::DiskNotFound { - DRIVE_STATE_OFFLINE + DriveState::Offline.to_string() } else { - DRIVE_STATE_CORRUPT + DriveState::Corrupt.to_string() } } else { - DRIVE_STATE_CORRUPT + DriveState::Corrupt.to_string() }; let uuid = if let Some(format) = format { diff --git a/crates/ecstore/src/store.rs b/crates/ecstore/src/store.rs index 193d3e061..765415fb8 100644 --- a/crates/ecstore/src/store.rs +++ b/crates/ecstore/src/store.rs @@ -30,11 +30,6 @@ use crate::global::{ GLOBAL_LOCAL_DISK_MAP, GLOBAL_LOCAL_DISK_SET_DRIVES, GLOBAL_TierConfigMgr, get_global_endpoints, is_dist_erasure, is_erasure_sd, set_global_deployment_id, set_object_layer, }; -use crate::heal::data_usage::{DATA_USAGE_ROOT, DataUsageInfo}; -use crate::heal::data_usage_cache::{DataUsageCache, DataUsageCacheInfo}; -use crate::heal::heal_commands::{HEAL_ITEM_METADATA, HealOpts, HealScanMode}; -use crate::heal::heal_ops::{HealEntryFn, HealSequence}; -use crate::new_object_layer_fn; use crate::notification_sys::get_global_notification_sys; use crate::pools::PoolMeta; use crate::rebalance::RebalanceMeta; @@ -54,13 +49,12 @@ use crate::{ store_init, }; use futures::future::join_all; -use glob::Pattern; use http::HeaderMap; use lazy_static::lazy_static; use rand::Rng as _; use rustfs_common::globals::{GLOBAL_Local_Node_Name, GLOBAL_Rustfs_Host, GLOBAL_Rustfs_Port}; +use rustfs_common::heal_channel::{HealItemType, HealOpts}; use rustfs_filemeta::FileInfo; -use rustfs_filemeta::MetaCacheEntry; use rustfs_madmin::heal_commands::HealResultItem; use rustfs_utils::crypto::base64_decode; use rustfs_utils::path::{SLASH_SEPARATOR, decode_dir_object, encode_dir_object, path_join_buf}; @@ -73,9 +67,8 @@ use std::time::SystemTime; use std::{collections::HashMap, sync::Arc, time::Duration}; use time::OffsetDateTime; use tokio::select; -use tokio::sync::mpsc::Sender; -use tokio::sync::{RwLock, broadcast, mpsc}; -use tokio::time::{interval, sleep}; +use tokio::sync::{RwLock, broadcast}; +use tokio::time::sleep; use tracing::{debug, info}; use tracing::{error, warn}; use uuid::Uuid; @@ -811,123 +804,6 @@ impl ECStore { errs } - pub async fn ns_scanner( - &self, - updates: Sender, - want_cycle: usize, - heal_scan_mode: HealScanMode, - ) -> Result<()> { - info!("ns_scanner updates - {}", want_cycle); - let all_buckets = self.list_bucket(&BucketOptions::default()).await?; - if all_buckets.is_empty() { - info!("No buckets found"); - let _ = updates.send(DataUsageInfo::default()).await; - return Ok(()); - } - - let mut total_results = 0; - let mut result_index = 0; - self.pools.iter().for_each(|pool| { - total_results += pool.disk_set.len(); - }); - let results = Arc::new(RwLock::new(vec![DataUsageCache::default(); total_results])); - let (cancel, _) = broadcast::channel(100); - let first_err = Arc::new(RwLock::new(None)); - let mut futures = Vec::new(); - for pool in self.pools.iter() { - for set in pool.disk_set.iter() { - let index = result_index; - let results_clone = results.clone(); - let first_err_clone = first_err.clone(); - let cancel_clone = cancel.clone(); - let all_buckets_clone = all_buckets.clone(); - futures.push(async move { - let (tx, mut rx) = mpsc::channel(1); - let task = tokio::spawn(async move { - loop { - match rx.recv().await { - Some(info) => { - results_clone.write().await[index] = info; - } - None => { - return; - } - } - } - }); - if let Err(err) = set - .clone() - .ns_scanner(&all_buckets_clone, want_cycle as u32, tx, heal_scan_mode) - .await - { - let mut f_w = first_err_clone.write().await; - if f_w.is_none() { - *f_w = Some(err); - } - let _ = cancel_clone.send(true); - return; - } - let _ = task.await; - }); - result_index += 1; - } - } - let (update_closer_tx, mut update_close_rx) = mpsc::channel(10); - let mut ctx_clone = cancel.subscribe(); - let all_buckets_clone = all_buckets.clone(); - // 新增：从环境变量读取 interval，默认 30 秒 - let ns_scanner_interval_secs = std::env::var("RUSTFS_NS_SCANNER_INTERVAL") - .ok() - .and_then(|v| v.parse::().ok()) - .unwrap_or(30); - - // 检查是否跳过后台任务 - let skip_background_task = std::env::var("RUSTFS_SKIP_BACKGROUND_TASK") - .ok() - .and_then(|v| v.parse::().ok()) - .unwrap_or(false); - - if skip_background_task { - info!("跳过后台任务执行：RUSTFS_SKIP_BACKGROUND_TASK=true"); - return Ok(()); - } - - let task = tokio::spawn(async move { - let mut last_update: Option = None; - let mut interval = interval(Duration::from_secs(ns_scanner_interval_secs)); - let all_merged = Arc::new(RwLock::new(DataUsageCache::default())); - loop { - select! { - _ = ctx_clone.recv() => { - return; - } - _ = update_close_rx.recv() => { - update_scan(all_merged.clone(), results.clone(), &mut last_update, all_buckets_clone.clone(), updates.clone()).await; - return; - } - _ = interval.tick() => { - update_scan(all_merged.clone(), results.clone(), &mut last_update, all_buckets_clone.clone(), updates.clone()).await; - } - } - } - }); - let _ = join_all(futures).await; - let mut ctx_closer = cancel.subscribe(); - select! { - _ = update_closer_tx.send(true) => { - - } - _ = ctx_closer.recv() => { - - } - } - let _ = task.await; - if let Some(err) = first_err.read().await.as_ref() { - return Err(err.clone()); - } - Ok(()) - } - async fn get_latest_object_info_with_idx( &self, bucket: &str, @@ -1068,34 +944,6 @@ impl ECStore { } } -#[tracing::instrument(level = "info", skip(all_buckets, updates))] -async fn update_scan( - all_merged: Arc>, - results: Arc>>, - last_update: &mut Option, - all_buckets: Vec, - updates: Sender, -) { - let mut w = all_merged.write().await; - *w = DataUsageCache { - info: DataUsageCacheInfo { - name: DATA_USAGE_ROOT.to_string(), - ..Default::default() - }, - ..Default::default() - }; - for info in results.read().await.iter() { - if info.info.last_update.is_none() { - return; - } - w.merge(info); - } - if (last_update.is_none() || w.info.last_update > *last_update) && w.root().is_some() { - let _ = updates.send(w.dui(&w.info.name, &all_buckets)).await; - *last_update = w.info.last_update; - } -} - pub async fn find_local_disk(disk_path: &String) -> Option { let disk_map = GLOBAL_LOCAL_DISK_MAP.read().await; @@ -2237,7 +2085,7 @@ impl StorageAPI for ECStore { async fn heal_format(&self, dry_run: bool) -> Result<(HealResultItem, Option)> { info!("heal_format"); let mut r = HealResultItem { - heal_item_type: HEAL_ITEM_METADATA.to_string(), + heal_item_type: HealItemType::Metadata.to_string(), detail: "disk-format".to_string(), ..Default::default() }; @@ -2351,120 +2199,6 @@ impl StorageAPI for ECStore { Ok((HealResultItem::default(), Some(Error::FileNotFound))) } - #[tracing::instrument(skip(self))] - async fn heal_objects( - &self, - bucket: &str, - prefix: &str, - opts: &HealOpts, - hs: Arc, - is_meta: bool, - ) -> Result<()> { - info!("heal objects"); - let opts_clone = *opts; - let heal_entry: HealEntryFn = Arc::new(move |bucket: String, entry: MetaCacheEntry, scan_mode: HealScanMode| { - let opts_clone = opts_clone; - let hs_clone = hs.clone(); - Box::pin(async move { - if entry.is_dir() { - return Ok(()); - } - - if bucket == RUSTFS_META_BUCKET - && Pattern::new("buckets/*/.metacache/*") - .map(|p| p.matches(&entry.name)) - .unwrap_or(false) - || Pattern::new("tmp/*").map(|p| p.matches(&entry.name)).unwrap_or(false) - || Pattern::new("multipart/*").map(|p| p.matches(&entry.name)).unwrap_or(false) - || Pattern::new("tmp-old/*").map(|p| p.matches(&entry.name)).unwrap_or(false) - { - return Ok(()); - } - let fivs = match entry.file_info_versions(&bucket) { - Ok(fivs) => fivs, - Err(_) => { - return if is_meta { - HealSequence::heal_meta_object(hs_clone.clone(), &bucket, &entry.name, "", scan_mode).await - } else { - HealSequence::heal_object(hs_clone.clone(), &bucket, &entry.name, "", scan_mode).await - }; - } - }; - - if opts_clone.remove && !opts_clone.dry_run { - let Some(store) = new_object_layer_fn() else { - return Err(Error::other("errServerNotInitialized")); - }; - - if let Err(err) = store.check_abandoned_parts(&bucket, &entry.name, &opts_clone).await { - info!("unable to check object {}/{} for abandoned data: {}", bucket, entry.name, err.to_string()); - } - } - for version in fivs.versions.iter() { - if is_meta { - if let Err(err) = HealSequence::heal_meta_object( - hs_clone.clone(), - &bucket, - &version.name, - &version.version_id.map(|v| v.to_string()).unwrap_or("".to_string()), - scan_mode, - ) - .await - { - match err { - Error::FileNotFound | Error::FileVersionNotFound => {} - _ => { - return Err(err); - } - } - } - } else if let Err(err) = HealSequence::heal_object( - hs_clone.clone(), - &bucket, - &version.name, - &version.version_id.map(|v| v.to_string()).unwrap_or("".to_string()), - scan_mode, - ) - .await - { - match err { - Error::FileNotFound | Error::FileVersionNotFound => {} - _ => { - return Err(err); - } - } - } - } - Ok(()) - }) - }); - let mut first_err = None; - for (idx, pool) in self.pools.iter().enumerate() { - if opts.pool.is_some() && opts.pool.unwrap() != idx { - continue; - } - //TODO: IsSuspended - - for (idx, set) in pool.disk_set.iter().enumerate() { - if opts.set.is_some() && opts.set.unwrap() != idx { - continue; - } - - if let Err(err) = set.list_and_heal(bucket, prefix, opts, heal_entry.clone()).await { - if first_err.is_none() { - first_err = Some(err) - } - } - } - } - - if first_err.is_some() { - return Err(first_err.unwrap()); - } - - Ok(()) - } - #[tracing::instrument(skip(self))] async fn get_pool_and_set(&self, id: &str) -> Result<(Option, Option, Option)> { for (pool_idx, pool) in self.pools.iter().enumerate() { diff --git a/crates/ecstore/src/store_api.rs b/crates/ecstore/src/store_api.rs index 7ca9cdb8f..5f1ea4e0d 100644 --- a/crates/ecstore/src/store_api.rs +++ b/crates/ecstore/src/store_api.rs @@ -15,16 +15,16 @@ use crate::bucket::metadata_sys::get_versioning_config; use crate::bucket::versioning::VersioningApi as _; use crate::cmd::bucket_replication::{ReplicationStatusType, VersionPurgeStatusType}; +use crate::disk::DiskStore; use crate::error::{Error, Result}; -use crate::heal::heal_ops::HealSequence; use crate::store_utils::clean_metadata; use crate::{ bucket::lifecycle::bucket_lifecycle_audit::LcAuditEvent, bucket::lifecycle::lifecycle::ExpirationOptions, bucket::lifecycle::{bucket_lifecycle_ops::TransitionedObject, lifecycle::TransitionOptions}, }; -use crate::{disk::DiskStore, heal::heal_commands::HealOpts}; use http::{HeaderMap, HeaderValue}; +use rustfs_common::heal_channel::HealOpts; use rustfs_filemeta::headers::RESERVED_METADATA_PREFIX_LOWER; use rustfs_filemeta::{FileInfo, MetaCacheEntriesSorted, ObjectPartInfo, headers::AMZ_OBJECT_TAGGING}; use rustfs_madmin::heal_commands::HealResultItem; @@ -1072,8 +1072,8 @@ pub trait StorageAPI: ObjectIO { version_id: &str, opts: &HealOpts, ) -> Result<(HealResultItem, Option)>; - async fn heal_objects(&self, bucket: &str, prefix: &str, opts: &HealOpts, hs: Arc, is_meta: bool) - -> Result<()>; + // async fn heal_objects(&self, bucket: &str, prefix: &str, opts: &HealOpts, hs: Arc, is_meta: bool) + // -> Result<()>; async fn get_pool_and_set(&self, id: &str) -> Result<(Option, Option, Option)>; async fn check_abandoned_parts(&self, bucket: &str, object: &str, opts: &HealOpts) -> Result<()>; } diff --git a/crates/ecstore/src/store_init.rs b/crates/ecstore/src/store_init.rs index 85ef58988..36bc0f33c 100644 --- a/crates/ecstore/src/store_init.rs +++ b/crates/ecstore/src/store_init.rs @@ -24,7 +24,6 @@ use crate::{ new_disk, }, endpoints::Endpoints, - heal::heal_commands::init_healing_tracker, }; use futures::future::join_all; use std::collections::{HashMap, hash_map::Entry}; @@ -288,7 +287,7 @@ async fn save_format_file_all(disks: &[Option], formats: &[Option], formats: &[Option, format: &Option, heal_id: &str) -> disk::error::Result<()> { +pub async fn save_format_file(disk: &Option, format: &Option) -> disk::error::Result<()> { if disk.is_none() { return Err(DiskError::DiskNotFound); } @@ -331,10 +330,6 @@ pub async fn save_format_file(disk: &Option, format: &Option