diff --git a/crates/scanner/src/data_usage_define.rs b/crates/scanner/src/data_usage_define.rs index bd01ca06d..b50f5a403 100644 --- a/crates/scanner/src/data_usage_define.rs +++ b/crates/scanner/src/data_usage_define.rs @@ -15,7 +15,7 @@ use s3s::dto::BucketLifecycleConfiguration; use serde::{Deserialize, Serialize}; use std::{ - collections::HashMap, + collections::{HashMap, HashSet}, future::Future, sync::{Arc, LazyLock, Once}, time::SystemTime, @@ -27,7 +27,6 @@ use rustfs_config::ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS; pub use rustfs_data_usage::{ BucketTargetUsageInfo, BucketUsageInfo, DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageInfo, hash_path, }; -use rustfs_data_usage::{DataUsageCache as SharedDataUsageCache, DataUsageCacheInfo as SharedDataUsageCacheInfo}; use rustfs_ecstore::{ StorageAPI, bucket::{lifecycle::lifecycle::TRANSITION_COMPLETE, replication::ReplicationConfig}, @@ -238,28 +237,6 @@ pub struct DataUsageCache { } impl DataUsageCache { - fn as_shared(&self) -> SharedDataUsageCache { - SharedDataUsageCache { - info: SharedDataUsageCacheInfo { - name: self.info.name.clone(), - next_cycle: self.info.next_cycle, - last_update: self.info.last_update, - skip_healing: self.info.skip_healing, - failed_objects: self.info.failed_objects.clone(), - }, - cache: self.cache.clone(), - } - } - - fn apply_shared_state(&mut self, shared: SharedDataUsageCache) { - self.info.name = shared.info.name; - self.info.next_cycle = shared.info.next_cycle; - self.info.last_update = shared.info.last_update; - self.info.skip_healing = shared.info.skip_healing; - self.info.failed_objects = shared.info.failed_objects; - self.cache = shared.cache; - } - fn ensure_cache_save_metrics_registered() { CACHE_SAVE_METRICS_ONCE.call_once(|| { describe_counter!( @@ -286,15 +263,19 @@ impl DataUsageCache { } pub fn replace(&mut self, path: &str, parent: &str, e: DataUsageEntry) { - let mut shared = self.as_shared(); - shared.replace(path, parent, e); - self.apply_shared_state(shared); + let hash = hash_path(path); + self.cache.insert(hash.key(), e); + if !parent.is_empty() { + let parent_hash = hash_path(parent); + self.cache.entry(parent_hash.key()).or_default().add_child(&hash); + } } pub fn replace_hashed(&mut self, hash: &DataUsageHash, parent: &Option, e: &DataUsageEntry) { - let mut shared = self.as_shared(); - shared.replace_hashed(hash, parent, e); - self.apply_shared_state(shared); + 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<&DataUsageEntry> { @@ -302,73 +283,278 @@ impl DataUsageCache { } pub fn find_children_copy(&mut self, h: DataUsageHash) -> DataUsageHashMap { - let mut shared = self.as_shared(); - let children = shared.find_children_copy(h); - self.apply_shared_state(shared); - children + self.cache.entry(h.string()).or_default().children.clone() } pub fn flatten(&self, root: &DataUsageEntry) -> DataUsageEntry { - self.as_shared().flatten(root) + 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) { - let mut shared = self.as_shared(); - shared.copy_with_children(&src.as_shared(), hash, parent); - self.apply_shared_state(shared); + 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 { + self.cache.entry(parent.key()).or_default().add_child(hash); + } + } } pub fn delete_recursive(&mut self, hash: &DataUsageHash) { - let mut shared = self.as_shared(); - shared.delete_recursive(hash); - self.apply_shared_state(shared); + 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()); + for child in need_remove { + self.delete_recursive(&DataUsageHash(child)); + } } pub fn size_recursive(&self, path: &str) -> Option { - self.as_shared().size_recursive(path) + match self.find(path) { + Some(root) => { + if root.children.is_empty() { + return Some(root.clone()); + } + let mut flat = self.flatten(root); + if flat.replication_stats.as_ref().is_some_and(|stats| stats.empty()) { + flat.replication_stats = None; + } + Some(flat) + } + None => None, + } } pub fn search_parent(&self, hash: &DataUsageHash) -> Option { - self.as_shared().search_parent(hash) + let want = hash.key(); + if let Some(last_index) = want.rfind('/') + && let Some(v) = self.find(&want[0..last_index]) + && v.children.contains(&want) + { + return Some(hash_path(&want[0..last_index])); + } + + for (k, v) in self.cache.iter() { + if v.children.contains(&want) { + return Some(DataUsageHash(k.clone())); + } + } + None } pub fn is_compacted(&self, hash: &DataUsageHash) -> bool { - self.as_shared().is_compacted(hash) + self.cache.get(&hash.key()).is_some_and(|due| due.compacted) } pub fn force_compact(&mut self, limit: usize) { - let mut shared = self.as_shared(); - shared.force_compact(limit); - self.apply_shared_state(shared); + if self.cache.len() < limit { + return; + } + let top = hash_path(&self.info.name).key(); + let Some(top_e) = self.find(&top).cloned() else { + return; + }; + + if top_e.children.len() > 250_000 { + 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, _| found.contains(k)); } pub fn reduce_children_of(&mut self, path: &DataUsageHash, limit: usize, compact_self: bool) { - let mut shared = self.as_shared(); - shared.reduce_children_of(path, limit, compact_self); - self.apply_shared_state(shared); + let Some(e) = self.cache.get(&path.key()).cloned() else { + 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_key(|a| a.objects); + + while remove > 0 && !leaves.is_empty() { + let Some(e) = leaves.first() else { + break; + }; + 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 { - self.as_shared().total_children_rec(path) + let Some(root) = self.find(path) else { + return 0; + }; + 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 shared = self.as_shared(); - shared.merge(&o.as_shared()); - self.apply_shared_state(shared); + let Some(mut existing_root) = self.root() else { + if o.root().is_none() { + return; + } + *self = o.clone(); + return; + }; + + let Some(other_root) = o.root() else { + return; + }; + + if o.info.last_update > self.info.last_update { + self.info.last_update = o.info.last_update; + } + + existing_root.merge(&other_root); + self.cache.insert(hash_path(&self.info.name).key(), existing_root); + + let root_hash = self.root_hash(); + for key in other_root.children.iter() { + let Some(entry) = o.cache.get(key) else { + continue; + }; + let flat = o.flatten(entry); + if let Some(existing) = self.cache.get_mut(key) { + existing.merge(&flat); + } else { + self.replace_hashed(&DataUsageHash(key.clone()), &Some(root_hash.clone()), &flat); + } + } } pub fn root_hash(&self) -> DataUsageHash { - self.as_shared().root_hash() + hash_path(&self.info.name) } pub fn root(&self) -> Option { - self.as_shared().root() + self.find(&self.info.name).cloned() } /// Convert cache to DataUsageInfo for a specific path pub fn dui(&self, path: &str, buckets: &[String]) -> DataUsageInfo { - self.as_shared().dui(path, buckets) + 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> { @@ -672,6 +858,40 @@ impl DataUsageCache { } } +#[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); + } + } +} + /// Trait for storage-specific operations on DataUsageCache #[async_trait::async_trait] pub trait DataUsageCacheStorage { @@ -803,6 +1023,111 @@ mod tests { assert_eq!(decoded.failed_objects, 0); } + #[test] + fn test_data_usage_cache_mutations_update_in_place() { + let mut cache = DataUsageCache { + info: DataUsageCacheInfo { + name: "bucket".to_string(), + failed_objects: HashMap::from([("bad-object".to_string(), 7)]), + ..Default::default() + }, + ..Default::default() + }; + + let root_hash = hash_path("bucket"); + let child_hash = hash_path("bucket/a"); + let grandchild_hash = hash_path("bucket/a/b"); + + cache.replace_hashed(&root_hash, &None, &DataUsageEntry::default()); + cache.replace_hashed( + &child_hash, + &Some(root_hash.clone()), + &DataUsageEntry { + objects: 2, + size: 20, + ..Default::default() + }, + ); + cache.replace_hashed( + &grandchild_hash, + &Some(child_hash.clone()), + &DataUsageEntry { + objects: 3, + size: 30, + ..Default::default() + }, + ); + + assert!(cache.find("bucket").unwrap().children.contains(&child_hash.key())); + assert!(cache.find("bucket/a").unwrap().children.contains(&grandchild_hash.key())); + assert_eq!(cache.search_parent(&grandchild_hash), Some(child_hash.clone())); + assert_eq!(cache.info.failed_objects.get("bad-object"), Some(&7)); + + let flat = cache.size_recursive("bucket").unwrap(); + assert_eq!(flat.objects, 5); + assert_eq!(flat.size, 50); + assert!(flat.children.is_empty()); + } + + #[test] + fn test_data_usage_cache_copy_and_delete_recursive() { + let root_hash = hash_path("bucket"); + let child_hash = hash_path("bucket/a"); + let grandchild_hash = hash_path("bucket/a/b"); + + let mut src = DataUsageCache { + info: DataUsageCacheInfo { + name: "bucket".to_string(), + ..Default::default() + }, + ..Default::default() + }; + src.replace_hashed(&root_hash, &None, &DataUsageEntry::default()); + src.replace_hashed( + &child_hash, + &Some(root_hash.clone()), + &DataUsageEntry { + objects: 1, + ..Default::default() + }, + ); + src.replace_hashed( + &grandchild_hash, + &Some(child_hash.clone()), + &DataUsageEntry { + objects: 1, + ..Default::default() + }, + ); + + let mut dst = DataUsageCache { + info: src.info.clone(), + ..Default::default() + }; + dst.replace_hashed(&root_hash, &None, &DataUsageEntry::default()); + dst.copy_with_children(&src, &child_hash, &Some(root_hash.clone())); + + assert!(dst.cache.contains_key(&child_hash.key())); + assert!(dst.cache.contains_key(&grandchild_hash.key())); + assert!(dst.find("bucket").unwrap().children.contains(&child_hash.key())); + assert!(dst.find("bucket/a").unwrap().children.contains(&grandchild_hash.key())); + + dst.delete_recursive(&child_hash); + + assert!(!dst.cache.contains_key(&child_hash.key())); + assert!(!dst.cache.contains_key(&grandchild_hash.key())); + assert!(dst.cache.contains_key(&root_hash.key())); + } + + #[test] + fn test_find_children_copy_preserves_missing_entry_behavior() { + let mut cache = DataUsageCache::default(); + let missing_hash = hash_path("missing"); + + assert!(cache.find_children_copy(missing_hash.clone()).is_empty()); + assert!(cache.cache.contains_key(&missing_hash.key())); + } + #[test] fn test_cache_path_type_distinguishes_main_and_backup() { assert_eq!(DataUsageCache::cache_path_type("buckets/.usage-cache.bin"), "main"); diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index f15fe48f3..308045491 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -12,20 +12,23 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::sync::Arc; +use std::sync::{ + Arc, + atomic::{AtomicU64, Ordering}, +}; use crate::data_usage_define::{BACKGROUND_HEAL_INFO_PATH, DATA_USAGE_BLOOM_NAME_PATH, DATA_USAGE_OBJ_NAME_PATH}; use crate::scanner_folder::{data_usage_update_dir_cycles, heal_object_select_prob}; use crate::scanner_io::ScannerIO; -use crate::sleeper::SCANNER_SLEEPER; +use crate::sleeper::{SCANNER_SLEEPER, scanner_speed_from_env_or_default, set_scanner_default_speed}; use crate::{DataUsageInfo, ScannerActivityGuard, ScannerError}; use chrono::{DateTime, Utc}; use rustfs_common::heal_channel::HealScanMode; use rustfs_common::metrics::{CurrentCycle, Metric, Metrics, emit_scan_cycle_complete, global_metrics}; use rustfs_config::ScannerSpeed; use rustfs_config::{ - DEFAULT_SCANNER_BITROT_CYCLE_SECS, DEFAULT_SCANNER_SPEED, ENV_SCANNER_BITROT_CYCLE_SECS, ENV_SCANNER_CYCLE, - ENV_SCANNER_SPEED, ENV_SCANNER_START_DELAY_SECS, + DEFAULT_SCANNER_BITROT_CYCLE_SECS, ENV_SCANNER_BITROT_CYCLE_SECS, ENV_SCANNER_CYCLE, ENV_SCANNER_SPEED, + ENV_SCANNER_START_DELAY_SECS, }; use rustfs_ecstore::StorageAPI as _; use rustfs_ecstore::config::com::{read_config, save_config}; @@ -40,12 +43,17 @@ use tokio_util::sync::CancellationToken; use tracing::{debug, error, info, instrument, warn}; const ENV_SCANNER_START_DELAY_SECS_DEPRECATED: &str = "RUSTFS_DATA_SCANNER_START_DELAY_SECS"; +const SINGLE_DISK_SCANNER_CYCLE_SECS: u64 = 24 * 60 * 60; +const NO_DEFAULT_CYCLE_OVERRIDE: u64 = 0; + +static SCANNER_DEFAULT_CYCLE_SECS: AtomicU64 = AtomicU64::new(NO_DEFAULT_CYCLE_OVERRIDE); /// Returns the base cycle interval. /// Priority order: /// 1. RUSTFS_SCANNER_CYCLE (if set, overrides everything) /// 2. RUSTFS_SCANNER_START_DELAY_SECS (for backward compatibility) -/// 3. RUSTFS_SCANNER_SPEED preset +/// 3. Deployment-specific default cycle override +/// 4. RUSTFS_SCANNER_SPEED preset fn cycle_interval() -> Duration { if let Some(secs) = rustfs_utils::get_env_opt_u64(ENV_SCANNER_CYCLE) { return Duration::from_secs(secs); @@ -53,8 +61,21 @@ fn cycle_interval() -> Duration { if let Some(secs) = scanner_start_delay_secs() { return Duration::from_secs(secs); } - let speed_str = rustfs_utils::get_env_str(ENV_SCANNER_SPEED, DEFAULT_SCANNER_SPEED); - ScannerSpeed::from_env_str(&speed_str).cycle_interval() + if let Some(secs) = scanner_default_cycle_secs() { + return Duration::from_secs(secs); + } + scanner_speed_from_env_or_default().cycle_interval() +} + +fn scanner_default_cycle_secs() -> Option { + match SCANNER_DEFAULT_CYCLE_SECS.load(Ordering::Relaxed) { + NO_DEFAULT_CYCLE_OVERRIDE => None, + secs => Some(secs), + } +} + +fn set_scanner_default_cycle_secs(secs: Option) { + SCANNER_DEFAULT_CYCLE_SECS.store(secs.unwrap_or(NO_DEFAULT_CYCLE_OVERRIDE), Ordering::Relaxed); } fn scanner_start_delay_secs() -> Option { @@ -87,6 +108,7 @@ fn initial_scanner_delay_for(start_delay_secs: Option) -> Duration { } pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc) { + configure_scanner_defaults().await; // Force init global sleeper so config is read once at startup. let _ = &*SCANNER_SLEEPER; @@ -114,6 +136,23 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc) { }); } +async fn configure_scanner_defaults() { + if is_erasure_sd().await { + set_scanner_default_speed(ScannerSpeed::Slowest); + set_scanner_default_cycle_secs(Some(SINGLE_DISK_SCANNER_CYCLE_SECS)); + info!( + env_speed = ENV_SCANNER_SPEED, + env_cycle = ENV_SCANNER_CYCLE, + env_start_delay = ENV_SCANNER_START_DELAY_SECS, + default_cycle_secs = SINGLE_DISK_SCANNER_CYCLE_SECS, + "Using slower scanner defaults for single-disk deployments; explicit scanner cycle or start-delay settings still take precedence" + ); + } else { + set_scanner_default_speed(ScannerSpeed::Default); + set_scanner_default_cycle_secs(None); + } +} + fn bitrot_scan_cycle() -> Option { let Ok(value) = std::env::var(ENV_SCANNER_BITROT_CYCLE_SECS) else { return Some(Duration::from_secs(DEFAULT_SCANNER_BITROT_CYCLE_SECS)); @@ -471,6 +510,50 @@ mod tests { use serial_test::serial; use temp_env::{with_var, with_var_unset}; + struct ScannerDefaultSpeedGuard; + + impl ScannerDefaultSpeedGuard { + fn set(speed: ScannerSpeed) -> Self { + set_scanner_default_speed(speed); + Self + } + } + + impl Drop for ScannerDefaultSpeedGuard { + fn drop(&mut self) { + set_scanner_default_speed(ScannerSpeed::Default); + } + } + + struct ScannerDefaultCycleGuard; + + impl ScannerDefaultCycleGuard { + fn set(secs: u64) -> Self { + set_scanner_default_cycle_secs(Some(secs)); + Self + } + } + + impl Drop for ScannerDefaultCycleGuard { + fn drop(&mut self) { + set_scanner_default_cycle_secs(None); + } + } + + fn with_unset_scanner_timing_env(f: impl FnOnce()) { + with_var_unset(ENV_SCANNER_SPEED, || { + with_var_unset("MINIO_SCANNER_SPEED", || { + with_var_unset(ENV_SCANNER_CYCLE, || { + with_var_unset("MINIO_SCANNER_CYCLE", || { + with_var_unset(ENV_SCANNER_START_DELAY_SECS, || { + with_var_unset(ENV_SCANNER_START_DELAY_SECS_DEPRECATED, f); + }); + }); + }); + }); + }); + } + #[test] #[serial] fn test_randomized_cycle_delay_keeps_configured_start_delay() { @@ -510,6 +593,86 @@ mod tests { }); } + #[test] + #[serial] + fn test_cycle_interval_prefers_explicit_cycle_over_default_cycle() { + let _guard = ScannerDefaultCycleGuard::set(SINGLE_DISK_SCANNER_CYCLE_SECS); + + with_var(ENV_SCANNER_CYCLE, Some("42"), || { + assert_eq!(cycle_interval(), Duration::from_secs(42)); + }); + } + + #[test] + #[serial] + fn test_cycle_interval_uses_scanner_default_speed_override_when_unconfigured() { + let _guard = ScannerDefaultSpeedGuard::set(ScannerSpeed::Slowest); + + with_unset_scanner_timing_env(|| { + assert_eq!(cycle_interval(), Duration::from_secs(30 * 60)); + }); + } + + #[test] + #[serial] + fn test_cycle_interval_prefers_explicit_speed_over_default_speed_override() { + let _guard = ScannerDefaultSpeedGuard::set(ScannerSpeed::Slowest); + + with_var_unset(ENV_SCANNER_CYCLE, || { + with_var_unset("MINIO_SCANNER_CYCLE", || { + with_var_unset(ENV_SCANNER_START_DELAY_SECS, || { + with_var_unset(ENV_SCANNER_START_DELAY_SECS_DEPRECATED, || { + with_var(ENV_SCANNER_SPEED, Some("fastest"), || { + assert_eq!(cycle_interval(), Duration::from_secs(1)); + }); + }); + }); + }); + }); + } + + #[test] + #[serial] + fn test_cycle_interval_uses_default_cycle_override_when_unconfigured() { + let _guard = ScannerDefaultCycleGuard::set(SINGLE_DISK_SCANNER_CYCLE_SECS); + + with_unset_scanner_timing_env(|| { + assert_eq!(cycle_interval(), Duration::from_secs(SINGLE_DISK_SCANNER_CYCLE_SECS)); + }); + } + + #[test] + #[serial] + fn test_cycle_interval_keeps_single_disk_cycle_with_explicit_speed() { + let _guard = ScannerDefaultCycleGuard::set(SINGLE_DISK_SCANNER_CYCLE_SECS); + + with_var_unset(ENV_SCANNER_CYCLE, || { + with_var_unset("MINIO_SCANNER_CYCLE", || { + with_var_unset(ENV_SCANNER_START_DELAY_SECS, || { + with_var_unset(ENV_SCANNER_START_DELAY_SECS_DEPRECATED, || { + with_var(ENV_SCANNER_SPEED, Some("slowest"), || { + assert_eq!(cycle_interval(), Duration::from_secs(SINGLE_DISK_SCANNER_CYCLE_SECS)); + }); + }); + }); + }); + }); + } + + #[test] + #[serial] + fn test_cycle_interval_prefers_explicit_start_delay_over_default_cycle() { + let _guard = ScannerDefaultCycleGuard::set(SINGLE_DISK_SCANNER_CYCLE_SECS); + + with_var_unset(ENV_SCANNER_CYCLE, || { + with_var_unset("MINIO_SCANNER_CYCLE", || { + with_var(ENV_SCANNER_START_DELAY_SECS, Some("120"), || { + assert_eq!(cycle_interval(), Duration::from_secs(120)); + }); + }); + }); + } + #[test] #[serial] fn test_cycle_interval_supports_minio_speed_alias() { diff --git a/crates/scanner/src/sleeper.rs b/crates/scanner/src/sleeper.rs index 43434da14..ab9b97ba4 100644 --- a/crates/scanner/src/sleeper.rs +++ b/crates/scanner/src/sleeper.rs @@ -12,18 +12,54 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::sync::atomic::{AtomicBool, Ordering}; +use std::sync::atomic::{AtomicBool, AtomicU8, Ordering}; use std::sync::{Arc, LazyLock, RwLock}; use std::time::Instant; -use rustfs_config::{DEFAULT_SCANNER_IDLE_MODE, DEFAULT_SCANNER_SPEED, ENV_SCANNER_IDLE_MODE, ENV_SCANNER_SPEED, ScannerSpeed}; +use rustfs_config::{DEFAULT_SCANNER_IDLE_MODE, ENV_SCANNER_IDLE_MODE, ENV_SCANNER_SPEED, ScannerSpeed}; use tokio::time::Duration; const MIN_SLEEP: Duration = Duration::from_millis(1); +const SCANNER_SPEED_FASTEST: u8 = 0; +const SCANNER_SPEED_FAST: u8 = 1; +const SCANNER_SPEED_DEFAULT: u8 = 2; +const SCANNER_SPEED_SLOW: u8 = 3; +const SCANNER_SPEED_SLOWEST: u8 = 4; + +static SCANNER_DEFAULT_SPEED_PRESET: AtomicU8 = AtomicU8::new(SCANNER_SPEED_DEFAULT); + +const fn scanner_speed_code(speed: ScannerSpeed) -> u8 { + match speed { + ScannerSpeed::Fastest => SCANNER_SPEED_FASTEST, + ScannerSpeed::Fast => SCANNER_SPEED_FAST, + ScannerSpeed::Default => SCANNER_SPEED_DEFAULT, + ScannerSpeed::Slow => SCANNER_SPEED_SLOW, + ScannerSpeed::Slowest => SCANNER_SPEED_SLOWEST, + } +} + +fn scanner_speed_from_code(value: u8) -> ScannerSpeed { + match value { + SCANNER_SPEED_FASTEST => ScannerSpeed::Fastest, + SCANNER_SPEED_FAST => ScannerSpeed::Fast, + SCANNER_SPEED_SLOW => ScannerSpeed::Slow, + SCANNER_SPEED_SLOWEST => ScannerSpeed::Slowest, + _ => ScannerSpeed::Default, + } +} + +pub(crate) fn set_scanner_default_speed(speed: ScannerSpeed) { + SCANNER_DEFAULT_SPEED_PRESET.store(scanner_speed_code(speed), Ordering::Relaxed); +} + +pub(crate) fn scanner_speed_from_env_or_default() -> ScannerSpeed { + rustfs_utils::get_env_opt_str(ENV_SCANNER_SPEED) + .map(|speed| ScannerSpeed::from_env_str(&speed)) + .unwrap_or_else(|| scanner_speed_from_code(SCANNER_DEFAULT_SPEED_PRESET.load(Ordering::Relaxed))) +} fn scanner_env_config() -> (ScannerSpeed, bool) { - let speed_str = rustfs_utils::get_env_str(ENV_SCANNER_SPEED, DEFAULT_SCANNER_SPEED); - let speed = ScannerSpeed::from_env_str(&speed_str); + let speed = scanner_speed_from_env_or_default(); let idle_mode = rustfs_utils::get_env_bool(ENV_SCANNER_IDLE_MODE, DEFAULT_SCANNER_IDLE_MODE); (speed, idle_mode) } @@ -149,7 +185,22 @@ impl SleepTimer { mod tests { use super::*; use serial_test::serial; - use temp_env::with_var; + use temp_env::{with_var, with_var_unset}; + + struct ScannerDefaultSpeedGuard; + + impl ScannerDefaultSpeedGuard { + fn set(speed: ScannerSpeed) -> Self { + set_scanner_default_speed(speed); + Self + } + } + + impl Drop for ScannerDefaultSpeedGuard { + fn drop(&mut self) { + set_scanner_default_speed(ScannerSpeed::Default); + } + } #[test] fn test_scanner_speed_presets() { @@ -198,6 +249,22 @@ mod tests { SCANNER_IDLE_MODE.store(prev_mode, Ordering::Relaxed); } + #[test] + #[serial] + fn test_refresh_from_env_uses_default_speed_override_when_speed_unset() { + let _guard = ScannerDefaultSpeedGuard::set(ScannerSpeed::Slowest); + let s = DynamicSleeper::new(ScannerSpeed::Default); + + with_var_unset(ENV_SCANNER_SPEED, || { + with_var_unset("MINIO_SCANNER_SPEED", || { + s.refresh_from_env(); + let (factor, max_sleep) = s.read_params(); + assert_eq!(factor, 100.0); + assert_eq!(max_sleep, Duration::from_secs(15)); + }); + }); + } + #[tokio::test(start_paused = true)] #[serial] async fn test_fastest_never_sleeps() {