From 68504577077eb61bf4d83dd234555885840008ad Mon Sep 17 00:00:00 2001 From: Henry Guo Date: Thu, 4 Jun 2026 13:16:26 +0800 Subject: [PATCH] feat(scanner): add runtime scanner controls and status (#3203) * feat(scanner): add runtime scanner controls and status * fix(scanner): validate persisted scanner config * docs(scanner): clarify start delay behavior --------- Co-authored-by: Henry Guo Co-authored-by: houseme --- crates/config/src/constants/scanner.rs | 90 ++- crates/ecstore/src/config/mod.rs | 15 + crates/ecstore/src/config/scanner.rs | 106 +++ crates/scanner/src/data_usage_define.rs | 13 +- crates/scanner/src/lib.rs | 2 + crates/scanner/src/runtime_config.rs | 819 ++++++++++++++++++++ crates/scanner/src/scanner.rs | 122 ++- crates/scanner/src/scanner_folder.rs | 30 +- crates/scanner/src/scanner_io.rs | 20 +- crates/scanner/src/sleeper.rs | 18 +- rustfs/src/admin/handlers/config_admin.rs | 158 +++- rustfs/src/admin/handlers/mod.rs | 2 + rustfs/src/admin/handlers/scanner.rs | 96 +++ rustfs/src/admin/mod.rs | 3 +- rustfs/src/admin/route_registration_test.rs | 5 +- rustfs/src/admin/service/config.rs | 21 +- 16 files changed, 1394 insertions(+), 126 deletions(-) create mode 100644 crates/ecstore/src/config/scanner.rs create mode 100644 crates/scanner/src/runtime_config.rs create mode 100644 rustfs/src/admin/handlers/scanner.rs diff --git a/crates/config/src/constants/scanner.rs b/crates/config/src/constants/scanner.rs index f43f63ae4..8aa4db787 100644 --- a/crates/config/src/constants/scanner.rs +++ b/crates/config/src/constants/scanner.rs @@ -14,6 +14,76 @@ use std::time::Duration; +/// Scanner admin config subsystem name. +pub const SCANNER_SUB_SYS: &str = "scanner"; + +/// Scanner config key selecting the speed preset. +pub const SCANNER_SPEED: &str = "speed"; + +/// Scanner config key overriding the cycle interval in seconds. +pub const SCANNER_CYCLE: &str = "cycle"; + +/// Scanner config key setting the startup delay in seconds. +/// +/// For compatibility, this also acts as the scanner cycle interval when +/// `cycle` is unset. +pub const SCANNER_START_DELAY: &str = "start_delay"; + +/// Scanner config key capping one cycle's runtime in seconds. +pub const SCANNER_CYCLE_MAX_DURATION: &str = "cycle_max_duration"; + +/// Scanner config key capping objects processed by one cycle. +pub const SCANNER_CYCLE_MAX_OBJECTS: &str = "cycle_max_objects"; + +/// Scanner config key capping directories entered by one cycle. +pub const SCANNER_CYCLE_MAX_DIRECTORIES: &str = "cycle_max_directories"; + +/// Scanner config key setting the periodic bitrot scan cycle in seconds. +pub const SCANNER_BITROT_CYCLE: &str = "bitrot_cycle"; + +/// Scanner config key controlling whether scanner throttling is enabled. +pub const SCANNER_IDLE_MODE: &str = "idle_mode"; + +/// Scanner config key controlling scanner cache save timeout in seconds. +pub const SCANNER_CACHE_SAVE_TIMEOUT: &str = "cache_save_timeout"; + +/// Scanner config key capping concurrent scanner set tasks. +pub const SCANNER_MAX_CONCURRENT_SET_SCANS: &str = "max_concurrent_set_scans"; + +/// Scanner config key capping concurrent scanner disk bucket walks per set. +pub const SCANNER_MAX_CONCURRENT_DISK_SCANS: &str = "max_concurrent_disk_scans"; + +/// Scanner config key controlling how often object loops yield. +pub const SCANNER_YIELD_EVERY_N_OBJECTS: &str = "yield_every_n_objects"; + +/// Scanner config key controlling object version count alerts. +pub const SCANNER_ALERT_EXCESS_VERSIONS: &str = "alert_excess_versions"; + +/// Scanner config key controlling retained version size alerts. +pub const SCANNER_ALERT_EXCESS_VERSION_SIZE: &str = "alert_excess_version_size"; + +/// Scanner config key controlling direct subfolder count alerts. +pub const SCANNER_ALERT_EXCESS_FOLDERS: &str = "alert_excess_folders"; + +/// Scanner config keys supported by the admin config subsystem. +pub const SCANNER_KEYS: &[&str] = &[ + SCANNER_SPEED, + SCANNER_CYCLE, + SCANNER_START_DELAY, + SCANNER_CYCLE_MAX_DURATION, + SCANNER_CYCLE_MAX_OBJECTS, + SCANNER_CYCLE_MAX_DIRECTORIES, + SCANNER_BITROT_CYCLE, + SCANNER_IDLE_MODE, + SCANNER_CACHE_SAVE_TIMEOUT, + SCANNER_MAX_CONCURRENT_SET_SCANS, + SCANNER_MAX_CONCURRENT_DISK_SCANS, + SCANNER_YIELD_EVERY_N_OBJECTS, + SCANNER_ALERT_EXCESS_VERSIONS, + SCANNER_ALERT_EXCESS_VERSION_SIZE, + SCANNER_ALERT_EXCESS_FOLDERS, +]; + /// Canonical environment variable name that specifies the scanner start delay in seconds. /// If set, this overrides the cycle interval derived from `RUSTFS_SCANNER_SPEED`. /// - Unit: seconds (u64). @@ -111,6 +181,9 @@ pub const ENV_SCANNER_IDLE_MODE: &str = "RUSTFS_SCANNER_IDLE_MODE"; /// - Example: `export RUSTFS_SCANNER_CACHE_SAVE_TIMEOUT_SECS=30` pub const ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS: &str = "RUSTFS_SCANNER_CACHE_SAVE_TIMEOUT_SECS"; +/// Default scanner cache save timeout in seconds. +pub const DEFAULT_SCANNER_CACHE_SAVE_TIMEOUT_SECS: u64 = 30; + /// Environment variable that caps concurrent scanner set tasks. /// A value of `0` keeps the existing topology-based concurrency. /// - Example: `export RUSTFS_SCANNER_MAX_CONCURRENT_SET_SCANS=2` @@ -203,15 +276,20 @@ impl ScannerSpeed { } } - pub fn from_env_str(s: &str) -> Self { + pub fn parse_str(s: &str) -> Option { match s.trim().to_ascii_lowercase().as_str() { - "fastest" => Self::Fastest, - "fast" => Self::Fast, - "slow" => Self::Slow, - "slowest" => Self::Slowest, - _ => Self::Default, + "fastest" => Some(Self::Fastest), + "fast" => Some(Self::Fast), + "default" => Some(Self::Default), + "slow" => Some(Self::Slow), + "slowest" => Some(Self::Slowest), + _ => None, } } + + pub fn from_env_str(s: &str) -> Self { + Self::parse_str(s).unwrap_or_default() + } } impl std::fmt::Display for ScannerSpeed { diff --git a/crates/ecstore/src/config/mod.rs b/crates/ecstore/src/config/mod.rs index 7b2cb6a51..831f4a449 100644 --- a/crates/ecstore/src/config/mod.rs +++ b/crates/ecstore/src/config/mod.rs @@ -18,6 +18,7 @@ pub mod com; pub mod heal; mod notify; mod oidc; +mod scanner; pub mod storageclass; use crate::error::Result; @@ -254,6 +255,7 @@ pub fn init() { let mut kvs = HashMap::new(); // Load storageclass default configuration kvs.insert(STORAGE_CLASS_SUB_SYS.to_owned(), storageclass::DEFAULT_KVS.clone()); + kvs.insert(rustfs_config::SCANNER_SUB_SYS.to_owned(), scanner::DEFAULT_KVS.clone()); // New: Loading default configurations for notify_webhook and notify_mqtt // Referring subsystem names through constants to improve the readability and maintainability of the code kvs.insert(NOTIFY_WEBHOOK_SUB_SYS.to_owned(), notify::DEFAULT_NOTIFY_WEBHOOK_KVS.clone()); @@ -283,6 +285,7 @@ pub fn init() { #[cfg(test)] mod tests { use super::*; + use rustfs_config::{DEFAULT_DELIMITER, DEFAULT_SCANNER_SPEED, SCANNER_CYCLE_MAX_OBJECTS, SCANNER_SPEED, SCANNER_SUB_SYS}; #[test] fn global_server_config_set_and_get_roundtrip() { @@ -300,4 +303,16 @@ mod tests { .expect("storage_class should exist"); assert_eq!(sc_kvs.get("standard"), "EC:4"); } + + #[test] + fn scanner_defaults_are_registered_for_admin_config() { + init(); + let cfg = Config::new(); + let scanner_kvs = cfg + .get_value(SCANNER_SUB_SYS, DEFAULT_DELIMITER) + .expect("scanner defaults should exist"); + + assert_eq!(scanner_kvs.get(SCANNER_SPEED), DEFAULT_SCANNER_SPEED); + assert_eq!(scanner_kvs.get(SCANNER_CYCLE_MAX_OBJECTS), "0"); + } } diff --git a/crates/ecstore/src/config/scanner.rs b/crates/ecstore/src/config/scanner.rs new file mode 100644 index 000000000..5ba304cdc --- /dev/null +++ b/crates/ecstore/src/config/scanner.rs @@ -0,0 +1,106 @@ +// 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 crate::config::{KV, KVS}; +use rustfs_config::{ + DEFAULT_SCANNER_BITROT_CYCLE_SECS, DEFAULT_SCANNER_CACHE_SAVE_TIMEOUT_SECS, DEFAULT_SCANNER_CYCLE_MAX_DIRECTORIES, + DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS, DEFAULT_SCANNER_CYCLE_MAX_OBJECTS, DEFAULT_SCANNER_IDLE_MODE, + DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS, DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS, DEFAULT_SCANNER_SPEED, + DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS, SCANNER_ALERT_EXCESS_FOLDERS, SCANNER_ALERT_EXCESS_VERSION_SIZE, + SCANNER_ALERT_EXCESS_VERSIONS, SCANNER_BITROT_CYCLE, SCANNER_CACHE_SAVE_TIMEOUT, SCANNER_CYCLE, + SCANNER_CYCLE_MAX_DIRECTORIES, SCANNER_CYCLE_MAX_DURATION, SCANNER_CYCLE_MAX_OBJECTS, SCANNER_IDLE_MODE, + SCANNER_MAX_CONCURRENT_DISK_SCANS, SCANNER_MAX_CONCURRENT_SET_SCANS, SCANNER_SPEED, SCANNER_START_DELAY, + SCANNER_YIELD_EVERY_N_OBJECTS, +}; +use std::sync::LazyLock; + +pub static DEFAULT_KVS: LazyLock = LazyLock::new(|| { + KVS(vec![ + KV { + key: SCANNER_SPEED.to_owned(), + value: DEFAULT_SCANNER_SPEED.to_owned(), + hidden_if_empty: false, + }, + KV { + key: SCANNER_CYCLE.to_owned(), + value: String::new(), + hidden_if_empty: true, + }, + KV { + key: SCANNER_START_DELAY.to_owned(), + value: String::new(), + hidden_if_empty: true, + }, + KV { + key: SCANNER_CYCLE_MAX_DURATION.to_owned(), + value: DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS.to_string(), + hidden_if_empty: false, + }, + KV { + key: SCANNER_CYCLE_MAX_OBJECTS.to_owned(), + value: DEFAULT_SCANNER_CYCLE_MAX_OBJECTS.to_string(), + hidden_if_empty: false, + }, + KV { + key: SCANNER_CYCLE_MAX_DIRECTORIES.to_owned(), + value: DEFAULT_SCANNER_CYCLE_MAX_DIRECTORIES.to_string(), + hidden_if_empty: false, + }, + KV { + key: SCANNER_BITROT_CYCLE.to_owned(), + value: DEFAULT_SCANNER_BITROT_CYCLE_SECS.to_string(), + hidden_if_empty: false, + }, + KV { + key: SCANNER_IDLE_MODE.to_owned(), + value: DEFAULT_SCANNER_IDLE_MODE.to_string(), + hidden_if_empty: false, + }, + KV { + key: SCANNER_CACHE_SAVE_TIMEOUT.to_owned(), + value: DEFAULT_SCANNER_CACHE_SAVE_TIMEOUT_SECS.to_string(), + hidden_if_empty: false, + }, + KV { + key: SCANNER_MAX_CONCURRENT_SET_SCANS.to_owned(), + value: DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS.to_string(), + hidden_if_empty: false, + }, + KV { + key: SCANNER_MAX_CONCURRENT_DISK_SCANS.to_owned(), + value: DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS.to_string(), + hidden_if_empty: false, + }, + KV { + key: SCANNER_YIELD_EVERY_N_OBJECTS.to_owned(), + value: DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS.to_string(), + hidden_if_empty: false, + }, + KV { + key: SCANNER_ALERT_EXCESS_VERSIONS.to_owned(), + value: rustfs_config::DEFAULT_SCANNER_ALERT_EXCESS_VERSIONS.to_string(), + hidden_if_empty: false, + }, + KV { + key: SCANNER_ALERT_EXCESS_VERSION_SIZE.to_owned(), + value: rustfs_config::DEFAULT_SCANNER_ALERT_EXCESS_VERSION_SIZE.to_string(), + hidden_if_empty: false, + }, + KV { + key: SCANNER_ALERT_EXCESS_FOLDERS.to_owned(), + value: rustfs_config::DEFAULT_SCANNER_ALERT_EXCESS_FOLDERS.to_string(), + hidden_if_empty: false, + }, + ]) +}); diff --git a/crates/scanner/src/data_usage_define.rs b/crates/scanner/src/data_usage_define.rs index 2dc849c83..11afdb844 100644 --- a/crates/scanner/src/data_usage_define.rs +++ b/crates/scanner/src/data_usage_define.rs @@ -23,6 +23,7 @@ use std::{ use http::HeaderMap; use metrics::{counter, describe_counter, describe_histogram, histogram}; +#[cfg(test)] use rustfs_config::ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS; pub use rustfs_data_usage::{ BucketTargetUsageInfo, BucketUsageInfo, DataUsageEntry, DataUsageHash, DataUsageHashMap, DataUsageInfo, hash_path, @@ -47,7 +48,6 @@ 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"; -const DATA_USAGE_CACHE_SAVE_TIMEOUT_SECS_DEFAULT: u64 = 30; const DATA_USAGE_CACHE_SAVE_RETRIES: u32 = 2; const DATA_USAGE_CACHE_BACKUP_SAVE_TIMEOUT_SECS_MAX: u64 = 5; const DATA_USAGE_CACHE_BACKUP_SAVE_RETRIES: u32 = 0; @@ -749,9 +749,7 @@ impl DataUsageCache { } fn cache_save_timeout() -> Duration { - Duration::from_secs( - rustfs_utils::get_env_u64(ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, DATA_USAGE_CACHE_SAVE_TIMEOUT_SECS_DEFAULT).max(1), - ) + crate::runtime_config::scanner_cache_save_timeout() } fn backup_cache_save_timeout(timeout_duration: Duration) -> Duration { @@ -1138,22 +1136,27 @@ mod tests { #[test] fn test_cache_save_timeout_uses_default_when_env_missing() { with_var_unset(ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, || { + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); assert_eq!( DataUsageCache::cache_save_timeout(), - Duration::from_secs(DATA_USAGE_CACHE_SAVE_TIMEOUT_SECS_DEFAULT) + Duration::from_secs(rustfs_config::DEFAULT_SCANNER_CACHE_SAVE_TIMEOUT_SECS) ); }); + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); } #[test] fn test_cache_save_timeout_respects_env_and_minimum_bound() { with_var(ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, Some("7"), || { + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); assert_eq!(DataUsageCache::cache_save_timeout(), Duration::from_secs(7)); }); with_var(ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, Some("0"), || { + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); assert_eq!(DataUsageCache::cache_save_timeout(), Duration::from_secs(1)); }); + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); } #[tokio::test] diff --git a/crates/scanner/src/lib.rs b/crates/scanner/src/lib.rs index cd71b0a52..04e83426f 100644 --- a/crates/scanner/src/lib.rs +++ b/crates/scanner/src/lib.rs @@ -22,6 +22,7 @@ pub mod data_usage_define; pub mod error; +pub mod runtime_config; pub mod scanner; pub mod scanner_budget; pub mod scanner_folder; @@ -30,6 +31,7 @@ pub mod sleeper; pub use data_usage_define::*; pub use error::ScannerError; +pub use runtime_config::{apply_scanner_runtime_config, scanner_runtime_config_status, validate_scanner_runtime_config}; pub use rustfs_common::last_minute; pub use scanner::init_data_scanner; pub use sleeper::{DynamicSleeper, SCANNER_IDLE_MODE, SCANNER_SLEEPER}; diff --git a/crates/scanner/src/runtime_config.rs b/crates/scanner/src/runtime_config.rs new file mode 100644 index 000000000..75fec4f69 --- /dev/null +++ b/crates/scanner/src/runtime_config.rs @@ -0,0 +1,819 @@ +// 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 crate::scanner_budget::ScannerCycleBudgetConfig; +use crate::sleeper::{SCANNER_SLEEPER, scanner_default_speed}; +use rustfs_config::{ + DEFAULT_DELIMITER, DEFAULT_SCANNER_ALERT_EXCESS_FOLDERS, DEFAULT_SCANNER_ALERT_EXCESS_VERSION_SIZE, + DEFAULT_SCANNER_ALERT_EXCESS_VERSIONS, DEFAULT_SCANNER_BITROT_CYCLE_SECS, DEFAULT_SCANNER_CACHE_SAVE_TIMEOUT_SECS, + DEFAULT_SCANNER_CYCLE_MAX_DIRECTORIES, DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS, DEFAULT_SCANNER_CYCLE_MAX_OBJECTS, + DEFAULT_SCANNER_IDLE_MODE, DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS, DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS, + DEFAULT_SCANNER_SPEED, DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS, ENV_SCANNER_ALERT_EXCESS_FOLDERS, + ENV_SCANNER_ALERT_EXCESS_VERSION_SIZE, ENV_SCANNER_ALERT_EXCESS_VERSIONS, ENV_SCANNER_BITROT_CYCLE_SECS, + ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, ENV_SCANNER_CYCLE, ENV_SCANNER_CYCLE_MAX_DIRECTORIES, + ENV_SCANNER_CYCLE_MAX_DURATION_SECS, ENV_SCANNER_CYCLE_MAX_OBJECTS, ENV_SCANNER_IDLE_MODE, + ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, ENV_SCANNER_MAX_CONCURRENT_SET_SCANS, ENV_SCANNER_SPEED, ENV_SCANNER_START_DELAY_SECS, + ENV_SCANNER_YIELD_EVERY_N_OBJECTS, SCANNER_ALERT_EXCESS_FOLDERS, SCANNER_ALERT_EXCESS_VERSION_SIZE, + SCANNER_ALERT_EXCESS_VERSIONS, SCANNER_BITROT_CYCLE, SCANNER_CACHE_SAVE_TIMEOUT, SCANNER_CYCLE, + SCANNER_CYCLE_MAX_DIRECTORIES, SCANNER_CYCLE_MAX_DURATION, SCANNER_CYCLE_MAX_OBJECTS, SCANNER_IDLE_MODE, + SCANNER_MAX_CONCURRENT_DISK_SCANS, SCANNER_MAX_CONCURRENT_SET_SCANS, SCANNER_SPEED, SCANNER_START_DELAY, SCANNER_SUB_SYS, + SCANNER_YIELD_EVERY_N_OBJECTS, ScannerSpeed, +}; +use rustfs_ecstore::config::{Config as ServerConfig, KVS}; +use serde::Serialize; +use std::fmt; +use std::sync::atomic::{AtomicU64, Ordering}; +use std::sync::{LazyLock, RwLock}; +use std::time::Duration; +use tracing::warn; + +const ENV_SCANNER_START_DELAY_SECS_DEPRECATED: &str = "RUSTFS_DATA_SCANNER_START_DELAY_SECS"; +const NO_DEFAULT_CYCLE_OVERRIDE: u64 = 0; + +static SCANNER_DEFAULT_CYCLE_SECS: AtomicU64 = AtomicU64::new(NO_DEFAULT_CYCLE_OVERRIDE); + +static SCANNER_RUNTIME_CONFIG: LazyLock> = + LazyLock::new(|| RwLock::new(lookup_scanner_runtime_config(None).unwrap_or_default())); + +#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)] +#[serde(rename_all = "snake_case")] +pub enum ScannerRuntimeConfigSource { + Env, + Config, + Default, +} + +#[derive(Clone, Debug, PartialEq, Eq)] +pub(crate) struct ScannerRuntimeConfig { + pub(crate) speed: ScannerSpeed, + pub(crate) speed_source: ScannerRuntimeConfigSource, + pub(crate) idle_mode: bool, + pub(crate) idle_mode_source: ScannerRuntimeConfigSource, + pub(crate) start_delay: Option, + pub(crate) start_delay_source: ScannerRuntimeConfigSource, + pub(crate) cycle_interval: Duration, + pub(crate) cycle_interval_source: ScannerRuntimeConfigSource, + pub(crate) bitrot_cycle: Option, + pub(crate) bitrot_cycle_source: ScannerRuntimeConfigSource, + pub(crate) cycle_budget: ScannerCycleBudgetConfig, + pub(crate) cycle_max_duration_source: ScannerRuntimeConfigSource, + pub(crate) cycle_max_objects_source: ScannerRuntimeConfigSource, + pub(crate) cycle_max_directories_source: ScannerRuntimeConfigSource, + pub(crate) cache_save_timeout: Duration, + pub(crate) cache_save_timeout_source: ScannerRuntimeConfigSource, + pub(crate) max_concurrent_set_scans: usize, + pub(crate) max_concurrent_set_scans_source: ScannerRuntimeConfigSource, + pub(crate) max_concurrent_disk_scans: usize, + pub(crate) max_concurrent_disk_scans_source: ScannerRuntimeConfigSource, + pub(crate) yield_every_n_objects: u64, + pub(crate) yield_every_n_objects_source: ScannerRuntimeConfigSource, + pub(crate) alert_excess_versions: u64, + pub(crate) alert_excess_versions_source: ScannerRuntimeConfigSource, + pub(crate) alert_excess_version_size: u64, + pub(crate) alert_excess_version_size_source: ScannerRuntimeConfigSource, + pub(crate) alert_excess_folders: u64, + pub(crate) alert_excess_folders_source: ScannerRuntimeConfigSource, +} + +impl Default for ScannerRuntimeConfig { + fn default() -> Self { + Self { + speed: scanner_default_speed(), + speed_source: ScannerRuntimeConfigSource::Default, + idle_mode: DEFAULT_SCANNER_IDLE_MODE, + idle_mode_source: ScannerRuntimeConfigSource::Default, + start_delay: None, + start_delay_source: ScannerRuntimeConfigSource::Default, + cycle_interval: scanner_default_cycle_secs() + .map(Duration::from_secs) + .unwrap_or_else(|| scanner_default_speed().cycle_interval()), + cycle_interval_source: ScannerRuntimeConfigSource::Default, + bitrot_cycle: Some(Duration::from_secs(DEFAULT_SCANNER_BITROT_CYCLE_SECS)), + bitrot_cycle_source: ScannerRuntimeConfigSource::Default, + cycle_budget: ScannerCycleBudgetConfig::default(), + cycle_max_duration_source: ScannerRuntimeConfigSource::Default, + cycle_max_objects_source: ScannerRuntimeConfigSource::Default, + cycle_max_directories_source: ScannerRuntimeConfigSource::Default, + cache_save_timeout: Duration::from_secs(DEFAULT_SCANNER_CACHE_SAVE_TIMEOUT_SECS), + cache_save_timeout_source: ScannerRuntimeConfigSource::Default, + max_concurrent_set_scans: DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS, + max_concurrent_set_scans_source: ScannerRuntimeConfigSource::Default, + max_concurrent_disk_scans: DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS, + max_concurrent_disk_scans_source: ScannerRuntimeConfigSource::Default, + yield_every_n_objects: DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS, + yield_every_n_objects_source: ScannerRuntimeConfigSource::Default, + alert_excess_versions: DEFAULT_SCANNER_ALERT_EXCESS_VERSIONS, + alert_excess_versions_source: ScannerRuntimeConfigSource::Default, + alert_excess_version_size: DEFAULT_SCANNER_ALERT_EXCESS_VERSION_SIZE, + alert_excess_version_size_source: ScannerRuntimeConfigSource::Default, + alert_excess_folders: DEFAULT_SCANNER_ALERT_EXCESS_FOLDERS, + alert_excess_folders_source: ScannerRuntimeConfigSource::Default, + } + } +} + +#[derive(Debug, Clone, PartialEq, Eq, thiserror::Error)] +pub enum ScannerRuntimeConfigError { + #[error("invalid scanner config value for {key}: {value} ({reason})")] + InvalidValue { + key: &'static str, + value: String, + reason: &'static str, + }, +} + +#[derive(Clone, Debug, Serialize)] +pub struct ScannerRuntimeConfigValue { + pub value: T, + pub source: ScannerRuntimeConfigSource, +} + +#[derive(Clone, Debug, Serialize)] +pub struct ScannerRuntimeConfigStatus { + pub speed: ScannerRuntimeConfigValue, + pub idle_mode: ScannerRuntimeConfigValue, + pub start_delay_seconds: ScannerRuntimeConfigValue>, + pub cycle_interval_seconds: ScannerRuntimeConfigValue, + pub bitrot_cycle_seconds: ScannerRuntimeConfigValue>, + pub cycle_max_duration_seconds: ScannerRuntimeConfigValue>, + pub cycle_max_objects: ScannerRuntimeConfigValue>, + pub cycle_max_directories: ScannerRuntimeConfigValue>, + pub cache_save_timeout_seconds: ScannerRuntimeConfigValue, + pub max_concurrent_set_scans: ScannerRuntimeConfigValue, + pub max_concurrent_disk_scans: ScannerRuntimeConfigValue, + pub yield_every_n_objects: ScannerRuntimeConfigValue, + pub alert_excess_versions: ScannerRuntimeConfigValue, + pub alert_excess_version_size: ScannerRuntimeConfigValue, + pub alert_excess_folders: ScannerRuntimeConfigValue, +} + +pub(crate) fn set_scanner_default_cycle_secs(secs: Option) { + SCANNER_DEFAULT_CYCLE_SECS.store(secs.unwrap_or(NO_DEFAULT_CYCLE_OVERRIDE), Ordering::Relaxed); +} + +fn scanner_default_cycle_secs() -> Option { + match SCANNER_DEFAULT_CYCLE_SECS.load(Ordering::Relaxed) { + NO_DEFAULT_CYCLE_OVERRIDE => None, + secs => Some(secs), + } +} + +fn config_value(kvs: Option<&KVS>, key: &'static str, default: impl fmt::Display) -> Option { + let value = kvs?.lookup(key)?; + let value = value.trim(); + if value.is_empty() || value == default.to_string() { + None + } else { + Some(value.to_string()) + } +} + +fn scanner_kvs(config: Option<&ServerConfig>) -> Option { + config.and_then(|config| config.get_value(SCANNER_SUB_SYS, DEFAULT_DELIMITER)) +} + +fn validate_default_scanner_target(config: &ServerConfig) -> Result<(), ScannerRuntimeConfigError> { + let Some(targets) = config.0.get(SCANNER_SUB_SYS) else { + return Ok(()); + }; + + for target in targets.keys() { + if target != DEFAULT_DELIMITER { + return Err(invalid_value("target", target.clone(), "scanner config only supports the default target")); + } + } + Ok(()) +} + +fn invalid_value(key: &'static str, value: impl Into, reason: &'static str) -> ScannerRuntimeConfigError { + ScannerRuntimeConfigError::InvalidValue { + key, + value: value.into(), + reason, + } +} + +fn parse_config_u64(key: &'static str, value: String) -> Result { + value + .parse::() + .map_err(|_| invalid_value(key, value, "expected unsigned integer seconds or count")) +} + +fn parse_config_usize(key: &'static str, value: String) -> Result { + value + .parse::() + .map_err(|_| invalid_value(key, value, "expected unsigned integer count")) +} + +fn parse_config_bool(key: &'static str, value: String) -> Result { + match value.trim().to_ascii_lowercase().as_str() { + "1" | "t" | "true" | "on" | "yes" | "ok" | "success" | "active" | "enabled" => Ok(true), + "0" | "f" | "false" | "off" | "no" | "not_ok" | "failure" | "inactive" | "disabled" => Ok(false), + _ => Err(invalid_value(key, value, "expected boolean value")), + } +} + +fn parse_config_speed(value: String) -> Result { + ScannerSpeed::parse_str(&value).ok_or_else(|| invalid_value(SCANNER_SPEED, value, "expected scanner speed preset")) +} + +fn parse_config_bitrot_cycle(value: String) -> Result, ScannerRuntimeConfigError> { + match value.trim().to_ascii_lowercase().as_str() { + "0" | "true" | "on" | "yes" => Ok(Some(Duration::ZERO)), + "false" | "off" | "no" | "disabled" => Ok(None), + _ => parse_config_u64(SCANNER_BITROT_CYCLE, value).map(|secs| Some(Duration::from_secs(secs))), + } +} + +fn parse_env_bitrot_cycle(value: String) -> Option { + match value.trim().to_ascii_lowercase().as_str() { + "0" | "true" | "on" | "yes" => Some(Duration::ZERO), + "false" | "off" | "no" | "disabled" => None, + value => value.parse::().ok().map(Duration::from_secs).or_else(|| { + warn!( + env = ENV_SCANNER_BITROT_CYCLE_SECS, + value, + default_secs = DEFAULT_SCANNER_BITROT_CYCLE_SECS, + "Invalid scanner bitrot cycle, using default" + ); + Some(Duration::from_secs(DEFAULT_SCANNER_BITROT_CYCLE_SECS)) + }), + } +} + +fn validate_optional_config_u64( + kvs: Option<&KVS>, + key: &'static str, + default: impl fmt::Display, +) -> Result<(), ScannerRuntimeConfigError> { + if let Some(value) = config_value(kvs, key, default) { + parse_config_u64(key, value)?; + } + Ok(()) +} + +fn validate_optional_config_usize( + kvs: Option<&KVS>, + key: &'static str, + default: impl fmt::Display, +) -> Result<(), ScannerRuntimeConfigError> { + if let Some(value) = config_value(kvs, key, default) { + parse_config_usize(key, value)?; + } + Ok(()) +} + +fn validate_persisted_scanner_runtime_config(config: &ServerConfig) -> Result<(), ScannerRuntimeConfigError> { + validate_default_scanner_target(config)?; + + let kvs = scanner_kvs(Some(config)); + let kvs = kvs.as_ref(); + + if let Some(value) = config_value(kvs, SCANNER_SPEED, DEFAULT_SCANNER_SPEED) { + parse_config_speed(value)?; + } + if let Some(value) = config_value(kvs, SCANNER_IDLE_MODE, DEFAULT_SCANNER_IDLE_MODE) { + parse_config_bool(SCANNER_IDLE_MODE, value)?; + } + validate_optional_config_u64(kvs, SCANNER_START_DELAY, "")?; + validate_optional_config_u64(kvs, SCANNER_CYCLE, "")?; + validate_optional_config_u64(kvs, SCANNER_CYCLE_MAX_DURATION, DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS)?; + validate_optional_config_u64(kvs, SCANNER_CYCLE_MAX_OBJECTS, DEFAULT_SCANNER_CYCLE_MAX_OBJECTS)?; + validate_optional_config_u64(kvs, SCANNER_CYCLE_MAX_DIRECTORIES, DEFAULT_SCANNER_CYCLE_MAX_DIRECTORIES)?; + if let Some(value) = config_value(kvs, SCANNER_BITROT_CYCLE, DEFAULT_SCANNER_BITROT_CYCLE_SECS) { + parse_config_bitrot_cycle(value)?; + } + validate_optional_config_u64(kvs, SCANNER_CACHE_SAVE_TIMEOUT, DEFAULT_SCANNER_CACHE_SAVE_TIMEOUT_SECS)?; + validate_optional_config_usize(kvs, SCANNER_MAX_CONCURRENT_SET_SCANS, DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS)?; + validate_optional_config_usize(kvs, SCANNER_MAX_CONCURRENT_DISK_SCANS, DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS)?; + validate_optional_config_u64(kvs, SCANNER_YIELD_EVERY_N_OBJECTS, DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS)?; + validate_optional_config_u64(kvs, SCANNER_ALERT_EXCESS_VERSIONS, DEFAULT_SCANNER_ALERT_EXCESS_VERSIONS)?; + validate_optional_config_u64(kvs, SCANNER_ALERT_EXCESS_VERSION_SIZE, DEFAULT_SCANNER_ALERT_EXCESS_VERSION_SIZE)?; + validate_optional_config_u64(kvs, SCANNER_ALERT_EXCESS_FOLDERS, DEFAULT_SCANNER_ALERT_EXCESS_FOLDERS)?; + + Ok(()) +} + +fn lookup_speed(kvs: Option<&KVS>) -> Result<(ScannerSpeed, ScannerRuntimeConfigSource), ScannerRuntimeConfigError> { + if let Some(value) = rustfs_utils::get_env_opt_str(ENV_SCANNER_SPEED) { + return Ok((ScannerSpeed::from_env_str(&value), ScannerRuntimeConfigSource::Env)); + } + + if let Some(value) = config_value(kvs, SCANNER_SPEED, DEFAULT_SCANNER_SPEED) { + return parse_config_speed(value).map(|speed| (speed, ScannerRuntimeConfigSource::Config)); + } + + Ok((scanner_default_speed(), ScannerRuntimeConfigSource::Default)) +} + +fn lookup_optional_seconds( + kvs: Option<&KVS>, + key: &'static str, + env_key: &'static str, + default: u64, +) -> Result<(Option, ScannerRuntimeConfigSource), ScannerRuntimeConfigError> { + if let Some(secs) = rustfs_utils::get_env_opt_u64(env_key) { + return Ok((Some(Duration::from_secs(secs)), ScannerRuntimeConfigSource::Env)); + } + if let Some(value) = config_value(kvs, key, default) { + return parse_config_u64(key, value).map(|secs| (Some(Duration::from_secs(secs)), ScannerRuntimeConfigSource::Config)); + } + Ok((None, ScannerRuntimeConfigSource::Default)) +} + +fn lookup_start_delay(kvs: Option<&KVS>) -> Result<(Option, ScannerRuntimeConfigSource), ScannerRuntimeConfigError> { + let aliases = [ENV_SCANNER_START_DELAY_SECS_DEPRECATED]; + if let Some(secs) = rustfs_utils::get_env_opt_u64_with_aliases(ENV_SCANNER_START_DELAY_SECS, &aliases) { + return Ok((Some(Duration::from_secs(secs)), ScannerRuntimeConfigSource::Env)); + } + if let Some(value) = config_value(kvs, SCANNER_START_DELAY, "") { + return parse_config_u64(SCANNER_START_DELAY, value) + .map(|secs| (Some(Duration::from_secs(secs)), ScannerRuntimeConfigSource::Config)); + } + Ok((None, ScannerRuntimeConfigSource::Default)) +} + +fn lookup_count_budget( + kvs: Option<&KVS>, + key: &'static str, + env_key: &'static str, + default: u64, +) -> Result<(Option, ScannerRuntimeConfigSource), ScannerRuntimeConfigError> { + if let Some(value) = rustfs_utils::get_env_opt_u64(env_key) { + return Ok(((value != 0).then_some(value), ScannerRuntimeConfigSource::Env)); + } + if let Some(value) = config_value(kvs, key, default) { + let count = parse_config_u64(key, value)?; + return Ok(((count != 0).then_some(count), ScannerRuntimeConfigSource::Config)); + } + Ok((None, ScannerRuntimeConfigSource::Default)) +} + +fn lookup_u64( + kvs: Option<&KVS>, + key: &'static str, + env_key: &'static str, + default: u64, +) -> Result<(u64, ScannerRuntimeConfigSource), ScannerRuntimeConfigError> { + if let Some(value) = rustfs_utils::get_env_opt_u64(env_key) { + return Ok((value, ScannerRuntimeConfigSource::Env)); + } + if let Some(value) = config_value(kvs, key, default) { + return parse_config_u64(key, value).map(|value| (value, ScannerRuntimeConfigSource::Config)); + } + Ok((default, ScannerRuntimeConfigSource::Default)) +} + +fn lookup_usize( + kvs: Option<&KVS>, + key: &'static str, + env_key: &'static str, + default: usize, +) -> Result<(usize, ScannerRuntimeConfigSource), ScannerRuntimeConfigError> { + if let Some(value) = rustfs_utils::get_env_opt_usize(env_key) { + return Ok((value, ScannerRuntimeConfigSource::Env)); + } + if let Some(value) = config_value(kvs, key, default) { + return parse_config_usize(key, value).map(|value| (value, ScannerRuntimeConfigSource::Config)); + } + Ok((default, ScannerRuntimeConfigSource::Default)) +} + +fn lookup_bool( + kvs: Option<&KVS>, + key: &'static str, + env_key: &'static str, + default: bool, +) -> Result<(bool, ScannerRuntimeConfigSource), ScannerRuntimeConfigError> { + if let Some(value) = rustfs_utils::get_env_opt_bool(env_key) { + return Ok((value, ScannerRuntimeConfigSource::Env)); + } + if let Some(value) = config_value(kvs, key, default) { + return parse_config_bool(key, value).map(|value| (value, ScannerRuntimeConfigSource::Config)); + } + Ok((default, ScannerRuntimeConfigSource::Default)) +} + +pub(crate) fn lookup_scanner_runtime_config( + config: Option<&ServerConfig>, +) -> Result { + let kvs = scanner_kvs(config); + let kvs = kvs.as_ref(); + let (speed, speed_source) = lookup_speed(kvs)?; + let (idle_mode, idle_mode_source) = lookup_bool(kvs, SCANNER_IDLE_MODE, ENV_SCANNER_IDLE_MODE, DEFAULT_SCANNER_IDLE_MODE)?; + let (start_delay, start_delay_source) = lookup_start_delay(kvs)?; + + let (cycle_interval, cycle_interval_source) = if let Some(secs) = rustfs_utils::get_env_opt_u64(ENV_SCANNER_CYCLE) { + (Duration::from_secs(secs), ScannerRuntimeConfigSource::Env) + } else if let Some(value) = config_value(kvs, SCANNER_CYCLE, "") { + ( + Duration::from_secs(parse_config_u64(SCANNER_CYCLE, value)?), + ScannerRuntimeConfigSource::Config, + ) + } else if let Some(start_delay) = start_delay { + (start_delay, start_delay_source) + } else if let Some(secs) = scanner_default_cycle_secs() { + (Duration::from_secs(secs), ScannerRuntimeConfigSource::Default) + } else { + (speed.cycle_interval(), speed_source) + }; + + let (cycle_max_duration, cycle_max_duration_source) = lookup_optional_seconds( + kvs, + SCANNER_CYCLE_MAX_DURATION, + ENV_SCANNER_CYCLE_MAX_DURATION_SECS, + DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS, + )?; + let (cycle_max_objects, cycle_max_objects_source) = lookup_count_budget( + kvs, + SCANNER_CYCLE_MAX_OBJECTS, + ENV_SCANNER_CYCLE_MAX_OBJECTS, + DEFAULT_SCANNER_CYCLE_MAX_OBJECTS, + )?; + let (cycle_max_directories, cycle_max_directories_source) = lookup_count_budget( + kvs, + SCANNER_CYCLE_MAX_DIRECTORIES, + ENV_SCANNER_CYCLE_MAX_DIRECTORIES, + DEFAULT_SCANNER_CYCLE_MAX_DIRECTORIES, + )?; + let cycle_budget = ScannerCycleBudgetConfig { + max_duration: cycle_max_duration.filter(|duration| !duration.is_zero()), + max_objects: cycle_max_objects, + max_directories: cycle_max_directories, + }; + + let (bitrot_cycle, bitrot_cycle_source) = if let Some(value) = rustfs_utils::get_env_opt_str(ENV_SCANNER_BITROT_CYCLE_SECS) { + (parse_env_bitrot_cycle(value), ScannerRuntimeConfigSource::Env) + } else if let Some(value) = config_value(kvs, SCANNER_BITROT_CYCLE, DEFAULT_SCANNER_BITROT_CYCLE_SECS) { + (parse_config_bitrot_cycle(value)?, ScannerRuntimeConfigSource::Config) + } else { + ( + Some(Duration::from_secs(DEFAULT_SCANNER_BITROT_CYCLE_SECS)), + ScannerRuntimeConfigSource::Default, + ) + }; + + let (cache_save_timeout, cache_save_timeout_source) = lookup_u64( + kvs, + SCANNER_CACHE_SAVE_TIMEOUT, + ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, + DEFAULT_SCANNER_CACHE_SAVE_TIMEOUT_SECS, + )?; + let (max_concurrent_set_scans, max_concurrent_set_scans_source) = lookup_usize( + kvs, + SCANNER_MAX_CONCURRENT_SET_SCANS, + ENV_SCANNER_MAX_CONCURRENT_SET_SCANS, + DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS, + )?; + let (max_concurrent_disk_scans, max_concurrent_disk_scans_source) = lookup_usize( + kvs, + SCANNER_MAX_CONCURRENT_DISK_SCANS, + ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, + DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS, + )?; + let (yield_every_n_objects, yield_every_n_objects_source) = lookup_u64( + kvs, + SCANNER_YIELD_EVERY_N_OBJECTS, + ENV_SCANNER_YIELD_EVERY_N_OBJECTS, + DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS, + )?; + let (alert_excess_versions, alert_excess_versions_source) = lookup_u64( + kvs, + SCANNER_ALERT_EXCESS_VERSIONS, + ENV_SCANNER_ALERT_EXCESS_VERSIONS, + DEFAULT_SCANNER_ALERT_EXCESS_VERSIONS, + )?; + let (alert_excess_version_size, alert_excess_version_size_source) = lookup_u64( + kvs, + SCANNER_ALERT_EXCESS_VERSION_SIZE, + ENV_SCANNER_ALERT_EXCESS_VERSION_SIZE, + DEFAULT_SCANNER_ALERT_EXCESS_VERSION_SIZE, + )?; + let (alert_excess_folders, alert_excess_folders_source) = lookup_u64( + kvs, + SCANNER_ALERT_EXCESS_FOLDERS, + ENV_SCANNER_ALERT_EXCESS_FOLDERS, + DEFAULT_SCANNER_ALERT_EXCESS_FOLDERS, + )?; + + Ok(ScannerRuntimeConfig { + speed, + speed_source, + idle_mode, + idle_mode_source, + start_delay, + start_delay_source, + cycle_interval, + cycle_interval_source, + bitrot_cycle, + bitrot_cycle_source, + cycle_budget, + cycle_max_duration_source, + cycle_max_objects_source, + cycle_max_directories_source, + cache_save_timeout: Duration::from_secs(cache_save_timeout.max(1)), + cache_save_timeout_source, + max_concurrent_set_scans, + max_concurrent_set_scans_source, + max_concurrent_disk_scans, + max_concurrent_disk_scans_source, + yield_every_n_objects, + yield_every_n_objects_source, + alert_excess_versions, + alert_excess_versions_source, + alert_excess_version_size, + alert_excess_version_size_source, + alert_excess_folders, + alert_excess_folders_source, + }) +} + +fn apply_resolved_runtime_config(config: ScannerRuntimeConfig) { + SCANNER_SLEEPER.update_from_runtime_config(config.speed, config.idle_mode, config.yield_every_n_objects); + if let Ok(mut guard) = SCANNER_RUNTIME_CONFIG.write() { + *guard = config; + } +} + +pub fn validate_scanner_runtime_config(config: &ServerConfig) -> Result<(), ScannerRuntimeConfigError> { + validate_persisted_scanner_runtime_config(config) +} + +pub fn apply_scanner_runtime_config(config: &ServerConfig) -> Result<(), ScannerRuntimeConfigError> { + validate_scanner_runtime_config(config)?; + let resolved = lookup_scanner_runtime_config(Some(config))?; + apply_resolved_runtime_config(resolved); + Ok(()) +} + +pub(crate) fn refresh_scanner_runtime_config_from_global() -> Result<(), ScannerRuntimeConfigError> { + let config = rustfs_ecstore::config::get_global_server_config(); + let resolved = lookup_scanner_runtime_config(config.as_ref())?; + apply_resolved_runtime_config(resolved); + Ok(()) +} + +#[cfg(test)] +pub(crate) fn refresh_scanner_runtime_config_for_tests() { + if let Ok(resolved) = lookup_scanner_runtime_config(None) { + apply_resolved_runtime_config(resolved); + } +} + +pub(crate) fn current_scanner_runtime_config() -> ScannerRuntimeConfig { + SCANNER_RUNTIME_CONFIG.read().map(|guard| guard.clone()).unwrap_or_default() +} + +pub fn scanner_runtime_config_status() -> ScannerRuntimeConfigStatus { + let config = current_scanner_runtime_config(); + ScannerRuntimeConfigStatus { + speed: ScannerRuntimeConfigValue { + value: config.speed.to_string(), + source: config.speed_source, + }, + idle_mode: ScannerRuntimeConfigValue { + value: config.idle_mode, + source: config.idle_mode_source, + }, + start_delay_seconds: ScannerRuntimeConfigValue { + value: config.start_delay.map(|duration| duration.as_secs()), + source: config.start_delay_source, + }, + cycle_interval_seconds: ScannerRuntimeConfigValue { + value: config.cycle_interval.as_secs(), + source: config.cycle_interval_source, + }, + bitrot_cycle_seconds: ScannerRuntimeConfigValue { + value: config.bitrot_cycle.map(|duration| duration.as_secs()), + source: config.bitrot_cycle_source, + }, + cycle_max_duration_seconds: ScannerRuntimeConfigValue { + value: config.cycle_budget.max_duration.map(|duration| duration.as_secs()), + source: config.cycle_max_duration_source, + }, + cycle_max_objects: ScannerRuntimeConfigValue { + value: config.cycle_budget.max_objects, + source: config.cycle_max_objects_source, + }, + cycle_max_directories: ScannerRuntimeConfigValue { + value: config.cycle_budget.max_directories, + source: config.cycle_max_directories_source, + }, + cache_save_timeout_seconds: ScannerRuntimeConfigValue { + value: config.cache_save_timeout.as_secs(), + source: config.cache_save_timeout_source, + }, + max_concurrent_set_scans: ScannerRuntimeConfigValue { + value: config.max_concurrent_set_scans, + source: config.max_concurrent_set_scans_source, + }, + max_concurrent_disk_scans: ScannerRuntimeConfigValue { + value: config.max_concurrent_disk_scans, + source: config.max_concurrent_disk_scans_source, + }, + yield_every_n_objects: ScannerRuntimeConfigValue { + value: config.yield_every_n_objects, + source: config.yield_every_n_objects_source, + }, + alert_excess_versions: ScannerRuntimeConfigValue { + value: config.alert_excess_versions, + source: config.alert_excess_versions_source, + }, + alert_excess_version_size: ScannerRuntimeConfigValue { + value: config.alert_excess_version_size, + source: config.alert_excess_version_size_source, + }, + alert_excess_folders: ScannerRuntimeConfigValue { + value: config.alert_excess_folders, + source: config.alert_excess_folders_source, + }, + } +} + +pub(crate) fn scanner_cycle_interval() -> Duration { + current_scanner_runtime_config().cycle_interval +} + +pub(crate) fn scanner_start_delay() -> Option { + current_scanner_runtime_config().start_delay +} + +pub(crate) fn scanner_bitrot_cycle() -> Option { + current_scanner_runtime_config().bitrot_cycle +} + +pub(crate) fn scanner_yield_every_n_objects() -> u64 { + current_scanner_runtime_config().yield_every_n_objects +} + +pub(crate) fn scanner_cache_save_timeout() -> Duration { + current_scanner_runtime_config().cache_save_timeout +} + +pub(crate) fn scanner_max_concurrent_set_scans_configured() -> usize { + current_scanner_runtime_config().max_concurrent_set_scans +} + +pub(crate) fn scanner_max_concurrent_disk_scans_configured() -> usize { + current_scanner_runtime_config().max_concurrent_disk_scans +} + +pub(crate) fn scanner_alert_excess_versions() -> u64 { + current_scanner_runtime_config().alert_excess_versions +} + +pub(crate) fn scanner_alert_excess_version_size() -> u64 { + current_scanner_runtime_config().alert_excess_version_size +} + +pub(crate) fn scanner_alert_excess_folders() -> u64 { + current_scanner_runtime_config().alert_excess_folders +} + +#[cfg(test)] +mod tests { + use super::{ScannerRuntimeConfigSource, lookup_scanner_runtime_config, validate_scanner_runtime_config}; + use rustfs_config::{ + DEFAULT_DELIMITER, ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, ENV_SCANNER_CYCLE, ENV_SCANNER_CYCLE_MAX_OBJECTS, + ENV_SCANNER_SPEED, SCANNER_CACHE_SAVE_TIMEOUT, SCANNER_CYCLE, SCANNER_CYCLE_MAX_DIRECTORIES, SCANNER_CYCLE_MAX_DURATION, + SCANNER_CYCLE_MAX_OBJECTS, SCANNER_IDLE_MODE, SCANNER_SPEED, SCANNER_SUB_SYS, ScannerSpeed, + }; + use rustfs_ecstore::config::{Config as ServerConfig, KVS}; + use serial_test::serial; + use std::collections::HashMap; + use std::time::Duration; + use temp_env::{with_var, with_var_unset}; + + fn server_config_with_scanner(entries: &[(&str, &str)]) -> ServerConfig { + rustfs_ecstore::config::init(); + let mut config = ServerConfig::new(); + let mut kvs = KVS::new(); + for (key, value) in entries { + kvs.insert((*key).to_string(), (*value).to_string()); + } + config + .0 + .insert(SCANNER_SUB_SYS.to_string(), HashMap::from([(DEFAULT_DELIMITER.to_string(), kvs)])); + config.set_defaults(); + config + } + + fn server_config_with_scanner_target(target: &str, entries: &[(&str, &str)]) -> ServerConfig { + let mut config = server_config_with_scanner(&[]); + let mut kvs = KVS::new(); + for (key, value) in entries { + kvs.insert((*key).to_string(), (*value).to_string()); + } + config + .0 + .entry(SCANNER_SUB_SYS.to_string()) + .or_default() + .insert(target.to_string(), kvs); + config.set_defaults(); + config + } + + #[test] + #[serial] + fn scanner_runtime_config_uses_persisted_values_when_env_is_unset() { + let config = server_config_with_scanner(&[ + (SCANNER_SPEED, "slow"), + (SCANNER_CYCLE, "120"), + (SCANNER_CYCLE_MAX_DURATION, "30"), + (SCANNER_CYCLE_MAX_OBJECTS, "1000"), + (SCANNER_CYCLE_MAX_DIRECTORIES, "25"), + (SCANNER_IDLE_MODE, "off"), + ]); + + with_var_unset(ENV_SCANNER_SPEED, || { + with_var_unset(ENV_SCANNER_CYCLE, || { + let resolved = lookup_scanner_runtime_config(Some(&config)).expect("scanner runtime config"); + + assert_eq!(resolved.speed, ScannerSpeed::Slow); + assert_eq!(resolved.speed_source, ScannerRuntimeConfigSource::Config); + assert_eq!(resolved.cycle_interval, Duration::from_secs(120)); + assert_eq!(resolved.cycle_interval_source, ScannerRuntimeConfigSource::Config); + assert_eq!(resolved.cycle_budget.max_duration, Some(Duration::from_secs(30))); + assert_eq!(resolved.cycle_budget.max_objects, Some(1000)); + assert_eq!(resolved.cycle_budget.max_directories, Some(25)); + assert!(!resolved.idle_mode); + }); + }); + } + + #[test] + #[serial] + fn scanner_runtime_config_prefers_env_over_persisted_config() { + let config = server_config_with_scanner(&[(SCANNER_SPEED, "slowest"), (SCANNER_CYCLE, "600")]); + + with_var(ENV_SCANNER_SPEED, Some("fast"), || { + with_var(ENV_SCANNER_CYCLE, Some("45"), || { + let resolved = lookup_scanner_runtime_config(Some(&config)).expect("scanner runtime config"); + + assert_eq!(resolved.speed, ScannerSpeed::Fast); + assert_eq!(resolved.speed_source, ScannerRuntimeConfigSource::Env); + assert_eq!(resolved.cycle_interval, Duration::from_secs(45)); + assert_eq!(resolved.cycle_interval_source, ScannerRuntimeConfigSource::Env); + }); + }); + } + + #[test] + fn scanner_runtime_config_rejects_invalid_persisted_speed() { + let config = server_config_with_scanner(&[(SCANNER_SPEED, "warp")]); + + let err = lookup_scanner_runtime_config(Some(&config)).expect_err("invalid scanner speed should fail"); + assert!(err.to_string().contains("speed")); + } + + #[test] + #[serial] + fn scanner_runtime_config_validation_rejects_invalid_persisted_speed_with_env_override() { + let config = server_config_with_scanner(&[(SCANNER_SPEED, "warp")]); + + with_var(ENV_SCANNER_SPEED, Some("fast"), || { + let err = validate_scanner_runtime_config(&config).expect_err("persisted scanner speed should be validated"); + assert!(err.to_string().contains("speed")); + }); + } + + #[test] + fn scanner_runtime_config_validation_rejects_non_default_target() { + let config = server_config_with_scanner_target("analytics", &[(SCANNER_SPEED, "slow")]); + + let err = validate_scanner_runtime_config(&config).expect_err("scanner targets should be rejected"); + assert!(err.to_string().contains("target")); + } + + #[test] + #[serial] + fn scanner_runtime_config_status_reports_value_sources() { + let config = server_config_with_scanner(&[(SCANNER_CYCLE_MAX_OBJECTS, "100"), (SCANNER_CACHE_SAVE_TIMEOUT, "5")]); + + with_var_unset(ENV_SCANNER_CYCLE_MAX_OBJECTS, || { + with_var_unset(ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, || { + let resolved = lookup_scanner_runtime_config(Some(&config)).expect("scanner runtime config"); + super::apply_resolved_runtime_config(resolved); + + let status = super::scanner_runtime_config_status(); + + assert_eq!(status.cycle_max_objects.value, Some(100)); + assert_eq!(status.cycle_max_objects.source, ScannerRuntimeConfigSource::Config); + assert_eq!(status.cache_save_timeout_seconds.value, 5); + assert_eq!(status.cache_save_timeout_seconds.source, ScannerRuntimeConfigSource::Config); + }); + }); + super::refresh_scanner_runtime_config_for_tests(); + } +} diff --git a/crates/scanner/src/scanner.rs b/crates/scanner/src/scanner.rs index 99b99a11e..2b17268ad 100644 --- a/crates/scanner/src/scanner.rs +++ b/crates/scanner/src/scanner.rs @@ -12,16 +12,17 @@ // See the License for the specific language governing permissions and // limitations under the License. -use std::sync::{ - Arc, - atomic::{AtomicU64, Ordering}, -}; +use std::sync::Arc; use crate::data_usage_define::{BACKGROUND_HEAL_INFO_PATH, DATA_USAGE_BLOOM_NAME_PATH, DATA_USAGE_OBJ_NAME_PATH}; +use crate::runtime_config::{ + current_scanner_runtime_config, lookup_scanner_runtime_config, refresh_scanner_runtime_config_from_global, + scanner_bitrot_cycle, scanner_cycle_interval, scanner_start_delay, set_scanner_default_cycle_secs, +}; use crate::scanner_budget::{ScannerCycleBudget, ScannerCycleBudgetConfig, ScannerCycleBudgetReason}; use crate::scanner_folder::{data_usage_update_dir_cycles, heal_object_select_prob}; use crate::scanner_io::ScannerIO; -use crate::sleeper::{SCANNER_SLEEPER, scanner_speed_from_env_or_default, set_scanner_default_speed}; +use crate::sleeper::{SCANNER_SLEEPER, set_scanner_default_speed}; use crate::{DataUsageInfo, ScannerActivityGuard, ScannerError}; use chrono::{DateTime, Utc}; use rustfs_common::heal_channel::HealScanMode; @@ -29,11 +30,12 @@ use rustfs_common::metrics::{ CurrentCycle, Metric, Metrics, ScanCyclePartialReason, emit_scan_cycle_complete, emit_scan_cycle_partial, global_metrics, }; use rustfs_config::ScannerSpeed; +#[cfg(test)] use rustfs_config::{ - DEFAULT_SCANNER_BITROT_CYCLE_SECS, DEFAULT_SCANNER_CYCLE_MAX_DIRECTORIES, DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS, - DEFAULT_SCANNER_CYCLE_MAX_OBJECTS, ENV_SCANNER_BITROT_CYCLE_SECS, ENV_SCANNER_CYCLE, ENV_SCANNER_CYCLE_MAX_DIRECTORIES, - ENV_SCANNER_CYCLE_MAX_DURATION_SECS, ENV_SCANNER_CYCLE_MAX_OBJECTS, ENV_SCANNER_SPEED, ENV_SCANNER_START_DELAY_SECS, + ENV_SCANNER_BITROT_CYCLE_SECS, ENV_SCANNER_CYCLE_MAX_DIRECTORIES, ENV_SCANNER_CYCLE_MAX_DURATION_SECS, + ENV_SCANNER_CYCLE_MAX_OBJECTS, }; +use rustfs_config::{ENV_SCANNER_CYCLE, ENV_SCANNER_SPEED, ENV_SCANNER_START_DELAY_SECS}; use rustfs_ecstore::StorageAPI as _; use rustfs_ecstore::bucket::lifecycle::lifecycle::Lifecycle as _; use rustfs_ecstore::bucket::metadata_sys::{get_lifecycle_config, get_replication_config}; @@ -50,11 +52,9 @@ use tokio::time::{Duration, Instant}; 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); +#[cfg(test)] +const ENV_SCANNER_START_DELAY_SECS_DEPRECATED: &str = "RUSTFS_DATA_SCANNER_START_DELAY_SECS"; /// Returns the base cycle interval. /// Priority order: @@ -62,54 +62,28 @@ static SCANNER_DEFAULT_CYCLE_SECS: AtomicU64 = AtomicU64::new(NO_DEFAULT_CYCLE_O /// 2. RUSTFS_SCANNER_START_DELAY_SECS (for backward compatibility) /// 3. Deployment-specific default cycle override /// 4. RUSTFS_SCANNER_SPEED preset +#[cfg(test)] fn cycle_interval() -> Duration { - if let Some(secs) = rustfs_utils::get_env_opt_u64(ENV_SCANNER_CYCLE) { - return Duration::from_secs(secs); - } - if let Some(secs) = scanner_start_delay_secs() { - return Duration::from_secs(secs); - } - 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 { - let deprecated = [ENV_SCANNER_START_DELAY_SECS_DEPRECATED]; - rustfs_utils::get_env_opt_u64_with_aliases(ENV_SCANNER_START_DELAY_SECS, &deprecated) -} - -fn scanner_cycle_max_duration() -> Option { - match rustfs_utils::get_env_u64(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS) { - 0 => None, - secs => Some(Duration::from_secs(secs)), - } -} - -fn scanner_cycle_count_budget(env: &str, default: u64) -> Option { - match rustfs_utils::get_env_u64(env, default) { - 0 => None, - count => Some(count), - } + resolve_scanner_runtime_config().cycle_interval } fn scanner_cycle_budget_config() -> ScannerCycleBudgetConfig { - ScannerCycleBudgetConfig { - max_duration: scanner_cycle_max_duration(), - max_objects: scanner_cycle_count_budget(ENV_SCANNER_CYCLE_MAX_OBJECTS, DEFAULT_SCANNER_CYCLE_MAX_OBJECTS), - max_directories: scanner_cycle_count_budget(ENV_SCANNER_CYCLE_MAX_DIRECTORIES, DEFAULT_SCANNER_CYCLE_MAX_DIRECTORIES), + resolve_scanner_runtime_config().cycle_budget +} + +#[cfg(test)] +fn scanner_cycle_max_duration() -> Option { + resolve_scanner_runtime_config().cycle_budget.max_duration +} + +fn resolve_scanner_runtime_config() -> crate::runtime_config::ScannerRuntimeConfig { + let config = rustfs_ecstore::config::get_global_server_config(); + match lookup_scanner_runtime_config(config.as_ref()) { + Ok(config) => config, + Err(err) => { + warn!(error = %err, "Failed to resolve scanner runtime config, using last applied config"); + current_scanner_runtime_config() + } } } @@ -125,7 +99,7 @@ fn scan_cycle_partial_reason(reason: Option) -> ScanCy /// Compute a randomized inter-cycle sleep. // Delay is scan interval +- 10%, with a floor of 1 second. fn randomized_cycle_delay() -> Duration { - randomized_cycle_delay_for(cycle_interval()) + randomized_cycle_delay_for(scanner_cycle_interval()) } fn randomized_cycle_delay_for(interval: Duration) -> Duration { @@ -137,7 +111,7 @@ fn randomized_cycle_delay_for(interval: Duration) -> Duration { } fn initial_scanner_delay() -> Duration { - initial_scanner_delay_for(scanner_start_delay_secs()) + initial_scanner_delay_for(scanner_start_delay().map(|duration| duration.as_secs())) } fn initial_scanner_delay_for(start_delay_secs: Option) -> Duration { @@ -150,6 +124,9 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc) { configure_scanner_defaults(&storeapi).await; // Force init global sleeper so config is read once at startup. let _ = &*SCANNER_SLEEPER; + if let Err(err) = refresh_scanner_runtime_config_from_global() { + warn!(error = %err, "Failed to apply scanner runtime config at startup"); + } let ctx_clone = ctx; let storeapi_clone = storeapi; @@ -281,24 +258,9 @@ async fn configure_scanner_defaults(storeapi: &Arc) { } } +#[cfg(test)] 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)); - }; - - match value.trim().to_ascii_lowercase().as_str() { - "0" | "true" | "on" | "yes" => Some(Duration::ZERO), - "false" | "off" | "no" | "disabled" => None, - value => value.parse::().ok().map(Duration::from_secs).or_else(|| { - warn!( - env = ENV_SCANNER_BITROT_CYCLE_SECS, - value, - default_secs = DEFAULT_SCANNER_BITROT_CYCLE_SECS, - "Invalid scanner bitrot cycle, using default" - ); - Some(Duration::from_secs(DEFAULT_SCANNER_BITROT_CYCLE_SECS)) - }), - } + resolve_scanner_runtime_config().bitrot_cycle } fn get_cycle_scan_mode( @@ -476,9 +438,11 @@ async fn mark_scan_cycle_idle(cycle_info: &mut CurrentCycle) { #[instrument(skip_all)] async fn run_data_scanner_cycle(ctx: &CancellationToken, storeapi: &Arc, cycle_info: &mut CurrentCycle) { let _activity_guard = ScannerActivityGuard::new(); - SCANNER_SLEEPER.refresh_from_env(); - let configured_cycle_interval = cycle_interval(); - let configured_bitrot_cycle = bitrot_scan_cycle(); + if let Err(err) = refresh_scanner_runtime_config_from_global() { + warn!(error = %err, "Failed to refresh scanner runtime config, using last applied config"); + } + let configured_cycle_interval = scanner_cycle_interval(); + let configured_bitrot_cycle = scanner_bitrot_cycle(); let cycle_budget_config = scanner_cycle_budget_config(); global_metrics().record_scanner_cycle_config( configured_cycle_interval, @@ -790,10 +754,12 @@ mod tests { #[serial] fn test_initial_scanner_delay_uses_cycle_without_explicit_start_delay() { with_var(ENV_SCANNER_CYCLE, Some("120"), || { + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); let delay = initial_scanner_delay_for(None); assert!(delay >= Duration::from_secs(108)); assert!(delay <= Duration::from_secs(132)); }); + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); } #[test] diff --git a/crates/scanner/src/scanner_folder.rs b/crates/scanner/src/scanner_folder.rs index cfabc9d0a..d57ac5b3d 100644 --- a/crates/scanner/src/scanner_folder.rs +++ b/crates/scanner/src/scanner_folder.rs @@ -21,9 +21,12 @@ use std::time::{Duration, Instant, SystemTime}; use crate::ReplTargetSizeSummary; use crate::data_usage_define::{DataUsageCache, DataUsageEntry, DataUsageHash, DataUsageHashMap, SizeSummary, hash_path}; use crate::error::ScannerError; +use crate::runtime_config::{ + scanner_alert_excess_folders, scanner_alert_excess_version_size, scanner_alert_excess_versions, scanner_yield_every_n_objects, +}; use crate::scanner_budget::ScannerCycleBudget; use crate::scanner_io::ScannerIODisk as _; -use crate::sleeper::{DynamicSleeper, scanner_yield_every_n_objects}; +use crate::sleeper::DynamicSleeper; use metrics::{counter, describe_counter}; use rustfs_common::heal_channel::{ HEAL_DELETE_DANGLING, HealAdmissionResult, HealChannelPriority, HealChannelRequest, HealScanMode, @@ -124,24 +127,15 @@ fn ensure_scanner_alert_metrics_registered() { } fn scanner_excess_versions_threshold() -> u64 { - rustfs_utils::get_env_u64( - rustfs_config::ENV_SCANNER_ALERT_EXCESS_VERSIONS, - rustfs_config::DEFAULT_SCANNER_ALERT_EXCESS_VERSIONS, - ) + scanner_alert_excess_versions() } fn scanner_excess_version_size_threshold() -> u64 { - rustfs_utils::get_env_u64( - rustfs_config::ENV_SCANNER_ALERT_EXCESS_VERSION_SIZE, - rustfs_config::DEFAULT_SCANNER_ALERT_EXCESS_VERSION_SIZE, - ) + scanner_alert_excess_version_size() } fn scanner_excess_folders_threshold() -> u64 { - rustfs_utils::get_env_u64( - rustfs_config::ENV_SCANNER_ALERT_EXCESS_FOLDERS, - rustfs_config::DEFAULT_SCANNER_ALERT_EXCESS_FOLDERS, - ) + scanner_alert_excess_folders() } fn should_yield_after_object(object_count: u64, yield_every: u64) -> bool { @@ -1850,44 +1844,54 @@ mod tests { fn test_excessive_version_alert_thresholds_use_env() { with_var(rustfs_config::ENV_SCANNER_ALERT_EXCESS_VERSIONS, Some("3"), || { with_var(rustfs_config::ENV_SCANNER_ALERT_EXCESS_VERSION_SIZE, Some("100"), || { + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); assert_eq!(should_alert_excessive_versions(2, 99), (false, false)); assert_eq!(should_alert_excessive_versions(3, 99), (true, false)); assert_eq!(should_alert_excessive_versions(2, 100), (false, true)); assert_eq!(should_alert_excessive_versions(3, 100), (true, true)); }); }); + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); } #[test] #[serial] fn test_excessive_folders_threshold_uses_env() { with_var(rustfs_config::ENV_SCANNER_ALERT_EXCESS_FOLDERS, Some("3"), || { + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); assert_eq!(scanner_excess_folders_threshold(), 3); }); + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); } #[test] #[serial] fn test_excessive_folders_threshold_default_supports_pbs_layout() { with_var_unset(rustfs_config::ENV_SCANNER_ALERT_EXCESS_FOLDERS, || { + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); assert_eq!(scanner_excess_folders_threshold(), 65_538); }); + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); } #[test] #[serial] fn test_scanner_yield_every_n_objects_uses_env() { with_var(rustfs_config::ENV_SCANNER_YIELD_EVERY_N_OBJECTS, Some("32"), || { + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); assert_eq!(scanner_yield_every_n_objects(), 32); }); + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); } #[test] #[serial] fn test_scanner_yield_every_n_objects_uses_default() { with_var_unset(rustfs_config::ENV_SCANNER_YIELD_EVERY_N_OBJECTS, || { + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); assert_eq!(scanner_yield_every_n_objects(), rustfs_config::DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS); }); + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); } #[test] diff --git a/crates/scanner/src/scanner_io.rs b/crates/scanner/src/scanner_io.rs index 6231ed438..db716e41f 100644 --- a/crates/scanner/src/scanner_io.rs +++ b/crates/scanner/src/scanner_io.rs @@ -24,10 +24,8 @@ use metrics::counter; use rand::seq::SliceRandom as _; use rustfs_common::heal_channel::HealScanMode; use rustfs_common::metrics::{Metric, Metrics, emit_scan_bucket_drive_complete, emit_scan_bucket_drive_partial, global_metrics}; -use rustfs_config::{ - DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS, DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS, ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, - ENV_SCANNER_MAX_CONCURRENT_SET_SCANS, -}; +#[cfg(test)] +use rustfs_config::{ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, ENV_SCANNER_MAX_CONCURRENT_SET_SCANS}; use rustfs_ecstore::bucket::bucket_target_sys::BucketTargetSys; use rustfs_ecstore::bucket::lifecycle::bucket_lifecycle_ops::GLOBAL_ExpiryState; use rustfs_ecstore::bucket::lifecycle::lifecycle::Lifecycle; @@ -229,17 +227,11 @@ fn scanner_concurrency_limit(configured: usize, available: usize) -> usize { } fn scanner_max_concurrent_set_scans(available: usize) -> usize { - scanner_concurrency_limit( - rustfs_utils::get_env_usize(ENV_SCANNER_MAX_CONCURRENT_SET_SCANS, DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS), - available, - ) + scanner_concurrency_limit(crate::runtime_config::scanner_max_concurrent_set_scans_configured(), available) } fn scanner_max_concurrent_disk_scans(available: usize) -> usize { - scanner_concurrency_limit( - rustfs_utils::get_env_usize(ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS), - available, - ) + scanner_concurrency_limit(crate::runtime_config::scanner_max_concurrent_disk_scans_configured(), available) } fn record_set_scan_failure(first_err: &mut Option, err: Error) { @@ -1168,16 +1160,20 @@ mod tests { #[serial] fn scanner_max_concurrent_set_scans_uses_env_cap() { with_var(ENV_SCANNER_MAX_CONCURRENT_SET_SCANS, Some("2"), || { + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); assert_eq!(scanner_max_concurrent_set_scans(4), 2); }); + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); } #[test] #[serial] fn scanner_max_concurrent_disk_scans_uses_env_cap() { with_var(ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, Some("1"), || { + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); assert_eq!(scanner_max_concurrent_disk_scans(4), 1); }); + crate::runtime_config::refresh_scanner_runtime_config_for_tests(); } #[test] diff --git a/crates/scanner/src/sleeper.rs b/crates/scanner/src/sleeper.rs index 51170fd40..b58c1313f 100644 --- a/crates/scanner/src/sleeper.rs +++ b/crates/scanner/src/sleeper.rs @@ -56,10 +56,14 @@ pub(crate) fn set_scanner_default_speed(speed: ScannerSpeed) { SCANNER_DEFAULT_SPEED_PRESET.store(scanner_speed_code(speed), Ordering::Relaxed); } +pub(crate) fn scanner_default_speed() -> ScannerSpeed { + scanner_speed_from_code(SCANNER_DEFAULT_SPEED_PRESET.load(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))) + .unwrap_or_else(scanner_default_speed) } fn scanner_env_config() -> (ScannerSpeed, bool) { @@ -159,18 +163,26 @@ impl DynamicSleeper { /// Reload speed and idle-mode settings from the current environment. pub fn refresh_from_env(&self) { let (speed, idle_mode) = scanner_env_config(); + self.update_from_runtime_config(speed, idle_mode, scanner_yield_every_n_objects()); + } + + pub(crate) fn update_from_runtime_config(&self, speed: ScannerSpeed, idle_mode: bool, yield_every_n_objects: u64) { self.update(speed); SCANNER_IDLE_MODE.store(idle_mode, Ordering::Relaxed); - self.record_throttle_config(); + self.record_throttle_config_with_yield(yield_every_n_objects); } fn record_throttle_config(&self) { + self.record_throttle_config_with_yield(scanner_yield_every_n_objects()); + } + + fn record_throttle_config_with_yield(&self, yield_every_n_objects: u64) { let (factor, max_sleep) = self.read_params(); global_metrics().record_scanner_throttle_config( SCANNER_IDLE_MODE.load(Ordering::Relaxed), factor, max_sleep, - scanner_yield_every_n_objects(), + yield_every_n_objects, ); } } diff --git a/rustfs/src/admin/handlers/config_admin.rs b/rustfs/src/admin/handlers/config_admin.rs index 726cdfb43..4e7ba5bc8 100644 --- a/rustfs/src/admin/handlers/config_admin.rs +++ b/rustfs/src/admin/handlers/config_admin.rs @@ -49,10 +49,18 @@ use rustfs_config::oidc::{ OIDC_REDIRECT_URI, OIDC_REDIRECT_URI_DYNAMIC, OIDC_ROLE_POLICY, OIDC_SCOPES, OIDC_USERNAME_CLAIM, }; use rustfs_config::{ - COMMENT_KEY, DEFAULT_DELIMITER, ENABLE_KEY, ENV_PREFIX, MAX_ADMIN_REQUEST_BODY_SIZE, MQTT_BROKER, MQTT_KEEP_ALIVE_INTERVAL, - MQTT_PASSWORD, MQTT_QOS, MQTT_QUEUE_DIR, MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TOPIC, MQTT_USERNAME, - WEBHOOK_AUTH_TOKEN, WEBHOOK_BATCH_SIZE, WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_KEY, WEBHOOK_ENDPOINT, WEBHOOK_HTTP_TIMEOUT, - WEBHOOK_MAX_RETRY, WEBHOOK_QUEUE_DIR, WEBHOOK_QUEUE_LIMIT, WEBHOOK_RETRY_INTERVAL, + COMMENT_KEY, DEFAULT_DELIMITER, ENABLE_KEY, ENV_PREFIX, ENV_SCANNER_ALERT_EXCESS_FOLDERS, + ENV_SCANNER_ALERT_EXCESS_VERSION_SIZE, ENV_SCANNER_ALERT_EXCESS_VERSIONS, ENV_SCANNER_BITROT_CYCLE_SECS, + ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, ENV_SCANNER_CYCLE, ENV_SCANNER_CYCLE_MAX_DIRECTORIES, + ENV_SCANNER_CYCLE_MAX_DURATION_SECS, ENV_SCANNER_CYCLE_MAX_OBJECTS, ENV_SCANNER_IDLE_MODE, + ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, ENV_SCANNER_MAX_CONCURRENT_SET_SCANS, ENV_SCANNER_SPEED, ENV_SCANNER_START_DELAY_SECS, + ENV_SCANNER_YIELD_EVERY_N_OBJECTS, MAX_ADMIN_REQUEST_BODY_SIZE, MQTT_BROKER, MQTT_KEEP_ALIVE_INTERVAL, MQTT_PASSWORD, + MQTT_QOS, MQTT_QUEUE_DIR, MQTT_QUEUE_LIMIT, MQTT_RECONNECT_INTERVAL, MQTT_TOPIC, MQTT_USERNAME, SCANNER_ALERT_EXCESS_FOLDERS, + SCANNER_ALERT_EXCESS_VERSION_SIZE, SCANNER_ALERT_EXCESS_VERSIONS, SCANNER_BITROT_CYCLE, SCANNER_CACHE_SAVE_TIMEOUT, + SCANNER_CYCLE, SCANNER_CYCLE_MAX_DIRECTORIES, SCANNER_CYCLE_MAX_DURATION, SCANNER_CYCLE_MAX_OBJECTS, SCANNER_IDLE_MODE, + SCANNER_MAX_CONCURRENT_DISK_SCANS, SCANNER_MAX_CONCURRENT_SET_SCANS, SCANNER_SPEED, SCANNER_START_DELAY, SCANNER_SUB_SYS, + SCANNER_YIELD_EVERY_N_OBJECTS, WEBHOOK_AUTH_TOKEN, WEBHOOK_BATCH_SIZE, WEBHOOK_CLIENT_CERT, WEBHOOK_CLIENT_KEY, + WEBHOOK_ENDPOINT, WEBHOOK_HTTP_TIMEOUT, WEBHOOK_MAX_RETRY, WEBHOOK_QUEUE_DIR, WEBHOOK_QUEUE_LIMIT, WEBHOOK_RETRY_INTERVAL, }; use rustfs_credentials::Credentials; use rustfs_ecstore::config::com::STORAGE_CLASS_SUB_SYS; @@ -176,6 +184,99 @@ const STORAGE_CLASS_HELP_KEYS: &[HelpKeyMetadata] = &[ }, ]; +const SCANNER_HELP_KEYS: &[HelpKeyMetadata] = &[ + HelpKeyMetadata { + key: SCANNER_SPEED, + type_name: "fastest|fast|default|slow|slowest", + description: "set scanner throttling preset", + optional: true, + }, + HelpKeyMetadata { + key: SCANNER_CYCLE, + type_name: "seconds", + description: "override scanner cycle interval in seconds", + optional: true, + }, + HelpKeyMetadata { + key: SCANNER_START_DELAY, + type_name: "seconds", + description: "set scanner startup delay in seconds; used as legacy cycle interval when cycle is unset", + optional: true, + }, + HelpKeyMetadata { + key: SCANNER_CYCLE_MAX_DURATION, + type_name: "seconds", + description: "cap one scanner cycle runtime in seconds, 0 disables the cap", + optional: true, + }, + HelpKeyMetadata { + key: SCANNER_CYCLE_MAX_OBJECTS, + type_name: "number", + description: "cap objects processed by one scanner cycle, 0 disables the cap", + optional: true, + }, + HelpKeyMetadata { + key: SCANNER_CYCLE_MAX_DIRECTORIES, + type_name: "number", + description: "cap directories entered by one scanner cycle, 0 disables the cap", + optional: true, + }, + HelpKeyMetadata { + key: SCANNER_BITROT_CYCLE, + type_name: "seconds|off", + description: "set periodic deep bitrot scan cycle, 0 scans deeply every cycle, off disables periodic deep scans", + optional: true, + }, + HelpKeyMetadata { + key: SCANNER_IDLE_MODE, + type_name: "on|off", + description: "enable scanner throttling sleeps between operations", + optional: true, + }, + HelpKeyMetadata { + key: SCANNER_CACHE_SAVE_TIMEOUT, + type_name: "seconds", + description: "set scanner data-usage cache save timeout in seconds", + optional: true, + }, + HelpKeyMetadata { + key: SCANNER_MAX_CONCURRENT_SET_SCANS, + type_name: "number", + description: "cap concurrent scanner set tasks, 0 uses topology defaults", + optional: true, + }, + HelpKeyMetadata { + key: SCANNER_MAX_CONCURRENT_DISK_SCANS, + type_name: "number", + description: "cap concurrent disk bucket walks per set, 0 uses disk-count defaults", + optional: true, + }, + HelpKeyMetadata { + key: SCANNER_YIELD_EVERY_N_OBJECTS, + type_name: "number", + description: "yield to the async runtime after this many scanned objects, 0 disables extra object-count yields", + optional: true, + }, + HelpKeyMetadata { + key: SCANNER_ALERT_EXCESS_VERSIONS, + type_name: "number", + description: "object version count threshold for scanner alerts", + optional: true, + }, + HelpKeyMetadata { + key: SCANNER_ALERT_EXCESS_VERSION_SIZE, + type_name: "bytes", + description: "retained object version size threshold for scanner alerts", + optional: true, + }, + HelpKeyMetadata { + key: SCANNER_ALERT_EXCESS_FOLDERS, + type_name: "number", + description: "direct subfolder count threshold for scanner alerts", + optional: true, + }, +]; + const OIDC_HELP_KEYS: &[HelpKeyMetadata] = &[ HelpKeyMetadata { key: OIDC_CONFIG_URL, @@ -423,6 +524,12 @@ const HELP_SUBSYSTEMS: &[HelpSubSystemMetadata] = &[ multiple_targets: false, keys: STORAGE_CLASS_HELP_KEYS, }, + HelpSubSystemMetadata { + key: SCANNER_SUB_SYS, + description: "configure background data scanner scheduling, throttling, bitrot, and observability thresholds", + multiple_targets: false, + keys: SCANNER_HELP_KEYS, + }, HelpSubSystemMetadata { key: IDENTITY_OPENID_SUB_SYS, description: "enable OpenID SSO support", @@ -1200,6 +1307,21 @@ fn env_help_key(sub_system: &str, key: &str) -> String { (STORAGE_CLASS_SUB_SYS, "rrs") => RRS_ENV.to_string(), (STORAGE_CLASS_SUB_SYS, "optimize") => OPTIMIZE_ENV.to_string(), (STORAGE_CLASS_SUB_SYS, "inline_block") => INLINE_BLOCK_ENV.to_string(), + (SCANNER_SUB_SYS, SCANNER_SPEED) => ENV_SCANNER_SPEED.to_string(), + (SCANNER_SUB_SYS, SCANNER_CYCLE) => ENV_SCANNER_CYCLE.to_string(), + (SCANNER_SUB_SYS, SCANNER_START_DELAY) => ENV_SCANNER_START_DELAY_SECS.to_string(), + (SCANNER_SUB_SYS, SCANNER_CYCLE_MAX_DURATION) => ENV_SCANNER_CYCLE_MAX_DURATION_SECS.to_string(), + (SCANNER_SUB_SYS, SCANNER_CYCLE_MAX_OBJECTS) => ENV_SCANNER_CYCLE_MAX_OBJECTS.to_string(), + (SCANNER_SUB_SYS, SCANNER_CYCLE_MAX_DIRECTORIES) => ENV_SCANNER_CYCLE_MAX_DIRECTORIES.to_string(), + (SCANNER_SUB_SYS, SCANNER_BITROT_CYCLE) => ENV_SCANNER_BITROT_CYCLE_SECS.to_string(), + (SCANNER_SUB_SYS, SCANNER_IDLE_MODE) => ENV_SCANNER_IDLE_MODE.to_string(), + (SCANNER_SUB_SYS, SCANNER_CACHE_SAVE_TIMEOUT) => ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS.to_string(), + (SCANNER_SUB_SYS, SCANNER_MAX_CONCURRENT_SET_SCANS) => ENV_SCANNER_MAX_CONCURRENT_SET_SCANS.to_string(), + (SCANNER_SUB_SYS, SCANNER_MAX_CONCURRENT_DISK_SCANS) => ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS.to_string(), + (SCANNER_SUB_SYS, SCANNER_YIELD_EVERY_N_OBJECTS) => ENV_SCANNER_YIELD_EVERY_N_OBJECTS.to_string(), + (SCANNER_SUB_SYS, SCANNER_ALERT_EXCESS_VERSIONS) => ENV_SCANNER_ALERT_EXCESS_VERSIONS.to_string(), + (SCANNER_SUB_SYS, SCANNER_ALERT_EXCESS_VERSION_SIZE) => ENV_SCANNER_ALERT_EXCESS_VERSION_SIZE.to_string(), + (SCANNER_SUB_SYS, SCANNER_ALERT_EXCESS_FOLDERS) => ENV_SCANNER_ALERT_EXCESS_FOLDERS.to_string(), (IDENTITY_OPENID_SUB_SYS, ENABLE_KEY) => ENV_IDENTITY_OPENID_ENABLE.to_string(), (IDENTITY_OPENID_SUB_SYS, OIDC_CONFIG_URL) => ENV_IDENTITY_OPENID_CONFIG_URL.to_string(), (IDENTITY_OPENID_SUB_SYS, OIDC_CLIENT_ID) => ENV_IDENTITY_OPENID_CLIENT_ID.to_string(), @@ -1388,7 +1510,12 @@ fn build_help_response(sub_system: Option<&str>, key: Option<&str>, env_only: bo /// the entire config and must ensure runtime state (e.g. GLOBAL_STORAGE_CLASS) is /// refreshed on both the leader and all peers. async fn apply_and_signal_dynamic_subsystems(config: &ServerConfig) { - for sub_system in [STORAGE_CLASS_SUB_SYS, AUDIT_WEBHOOK_SUB_SYS, AUDIT_MQTT_SUB_SYS] { + for sub_system in [ + STORAGE_CLASS_SUB_SYS, + AUDIT_WEBHOOK_SUB_SYS, + AUDIT_MQTT_SUB_SYS, + SCANNER_SUB_SYS, + ] { if apply_dynamic_config_for_subsystem(config, sub_system).await.unwrap_or(false) { signal_dynamic_config_reload(sub_system).await; } @@ -1804,6 +1931,27 @@ identity_openid config_url="https://issuer.example" client_id="console""#, assert_eq!(response.keys_help[1].key, "endpoint"); } + #[test] + fn build_help_response_reports_scanner_keys() { + let response = build_help_response(Some("scanner"), Some("speed"), false).expect("scanner help response"); + + assert_eq!(response.sub_sys, "scanner"); + assert!(!response.multiple_targets); + assert_eq!(response.keys_help.len(), 1); + assert_eq!(response.keys_help[0].key, "speed"); + assert_eq!(response.keys_help[0].type_name, "fastest|fast|default|slow|slowest"); + } + + #[test] + fn build_help_response_reports_scanner_start_delay_legacy_cycle_behavior() { + let response = build_help_response(Some("scanner"), Some("start_delay"), false).expect("scanner help response"); + + assert_eq!(response.sub_sys, "scanner"); + assert_eq!(response.keys_help.len(), 1); + assert_eq!(response.keys_help[0].key, "start_delay"); + assert!(response.keys_help[0].description.contains("cycle")); + } + #[test] fn build_top_level_help_response_uses_empty_type_names() { let response = build_help_response(None, None, false).expect("top level help response"); diff --git a/rustfs/src/admin/handlers/mod.rs b/rustfs/src/admin/handlers/mod.rs index 245ecfbd9..0b246e7fc 100644 --- a/rustfs/src/admin/handlers/mod.rs +++ b/rustfs/src/admin/handlers/mod.rs @@ -40,6 +40,7 @@ pub mod profile_admin; pub mod quota; pub mod rebalance; pub mod replication; +pub mod scanner; pub mod service_account; pub mod site_replication; pub mod sts; @@ -90,6 +91,7 @@ mod tests { let _set_remote_target_handler = replication::SetRemoteTargetHandler {}; let _list_remote_target_handler = replication::ListRemoteTargetHandler {}; let _remove_remote_target_handler = replication::RemoveRemoteTargetHandler {}; + let _scanner_status_handler = scanner::ScannerStatusHandler {}; let _site_replication_add_handler = site_replication::SiteReplicationAddHandler {}; let _site_replication_info_handler = site_replication::SiteReplicationInfoHandler {}; let _site_replication_status_handler = site_replication::SiteReplicationStatusHandler {}; diff --git a/rustfs/src/admin/handlers/scanner.rs b/rustfs/src/admin/handlers/scanner.rs new file mode 100644 index 000000000..b6b8d149c --- /dev/null +++ b/rustfs/src/admin/handlers/scanner.rs @@ -0,0 +1,96 @@ +// 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 crate::admin::auth::validate_admin_request; +use crate::admin::router::{AdminOperation, Operation, S3Router}; +use crate::auth::{check_key_valid, get_session_token}; +use crate::server::{ADMIN_PREFIX, RemoteAddr}; +use http::{HeaderMap, HeaderValue}; +use hyper::{Method, StatusCode}; +use matchit::Params; +use rustfs_common::metrics::{ScannerMetricsReport, global_metrics}; +use rustfs_credentials::Credentials; +use rustfs_policy::policy::action::{Action, AdminAction}; +use s3s::header::CONTENT_TYPE; +use s3s::{Body, S3Error, S3ErrorCode, S3Request, S3Response, S3Result, s3_error}; +use serde::Serialize; + +const JSON_CONTENT_TYPE: &str = "application/json"; + +#[derive(Debug, Serialize)] +struct ScannerStatusResponse { + metrics: ScannerMetricsReport, + runtime_config: rustfs_scanner::runtime_config::ScannerRuntimeConfigStatus, +} + +pub fn register_scanner_route(r: &mut S3Router) -> std::io::Result<()> { + r.insert( + Method::GET, + format!("{ADMIN_PREFIX}/v3/scanner/status").as_str(), + AdminOperation(&ScannerStatusHandler {}), + )?; + + Ok(()) +} + +async fn validate_scanner_status_request(req: &S3Request) -> S3Result { + let Some(input_cred) = req.credentials.as_ref() else { + return Err(s3_error!(InvalidRequest, "missing credentials")); + }; + + let (cred, owner) = + check_key_valid(get_session_token(&req.uri, &req.headers).unwrap_or_default(), &input_cred.access_key).await?; + + let remote_addr = req + .extensions + .get::>() + .and_then(|opt| opt.map(|addr| addr.0)); + validate_admin_request( + &req.headers, + &cred, + owner, + false, + vec![Action::AdminAction(AdminAction::ServerInfoAdminAction)], + remote_addr, + ) + .await?; + + Ok(cred) +} + +fn json_response(body: Vec) -> S3Result> { + let mut headers = HeaderMap::new(); + let content_type = HeaderValue::from_str(JSON_CONTENT_TYPE) + .map_err(|err| S3Error::with_message(S3ErrorCode::InternalError, format!("invalid content type: {err}")))?; + headers.insert(CONTENT_TYPE, content_type); + Ok(S3Response::with_headers((StatusCode::OK, Body::from(body)), headers)) +} + +pub struct ScannerStatusHandler {} + +#[async_trait::async_trait] +impl Operation for ScannerStatusHandler { + async fn call(&self, req: S3Request, _params: Params<'_, '_>) -> S3Result> { + let _cred = validate_scanner_status_request(&req).await?; + let response = ScannerStatusResponse { + metrics: global_metrics().report().await, + runtime_config: rustfs_scanner::scanner_runtime_config_status(), + }; + let body = serde_json::to_vec(&response).map_err(|err| { + S3Error::with_message(S3ErrorCode::InternalError, format!("failed to encode scanner status: {err}")) + })?; + + json_response(body) + } +} diff --git a/rustfs/src/admin/mod.rs b/rustfs/src/admin/mod.rs index f8c3f09c4..81e54d93e 100644 --- a/rustfs/src/admin/mod.rs +++ b/rustfs/src/admin/mod.rs @@ -28,7 +28,7 @@ mod route_registration_test; use handlers::{ audit, bucket_meta, config_admin, heal, health, kms, module_switch, oidc, plugins_catalog, plugins_instances, pools, - profile_admin, quota, rebalance, replication, site_replication, sts, system, tier, tls_debug, user, + profile_admin, quota, rebalance, replication, scanner, site_replication, sts, system, tier, tls_debug, user, }; use router::{AdminOperation, S3Router}; use s3s::route::S3Route; @@ -58,6 +58,7 @@ pub fn make_admin_route(console_enabled: bool) -> std::io::Result quota::register_quota_route(&mut r)?; bucket_meta::register_bucket_meta_route(&mut r)?; config_admin::register_config_route(&mut r)?; + scanner::register_scanner_route(&mut r)?; audit::register_audit_target_route(&mut r)?; module_switch::register_module_switch_route(&mut r)?; plugins_catalog::register_plugin_catalog_route(&mut r)?; diff --git a/rustfs/src/admin/route_registration_test.rs b/rustfs/src/admin/route_registration_test.rs index 77c72f3f5..46c5d4364 100644 --- a/rustfs/src/admin/route_registration_test.rs +++ b/rustfs/src/admin/route_registration_test.rs @@ -15,7 +15,7 @@ use crate::admin::{ handlers::{ audit, bucket_meta, config_admin, heal, health, kms, module_switch, oidc, plugins_catalog, plugins_instances, pools, - profile_admin, quota, rebalance, replication, site_replication, sts, system, tier, tls_debug, user, + profile_admin, quota, rebalance, replication, scanner, site_replication, sts, system, tier, tls_debug, user, }, router::{AdminOperation, S3Router}, }; @@ -53,6 +53,7 @@ fn register_admin_routes(router: &mut S3Router) { quota::register_quota_route(router).expect("register quota route"); bucket_meta::register_bucket_meta_route(router).expect("register bucket meta route"); config_admin::register_config_route(router).expect("register config admin route"); + scanner::register_scanner_route(router).expect("register scanner route"); audit::register_audit_target_route(router).expect("register audit target route"); module_switch::register_module_switch_route(router).expect("register module switch route"); plugins_catalog::register_plugin_catalog_route(router).expect("register plugin catalog route"); @@ -123,6 +124,7 @@ fn test_register_routes_cover_representative_admin_paths() { assert_route(&router, Method::PUT, &admin_path("/v3/restore-config-history-kv")); assert_route(&router, Method::GET, &admin_path("/v3/config")); assert_route(&router, Method::PUT, &admin_path("/v3/config")); + assert_route(&router, Method::GET, &admin_path("/v3/scanner/status")); assert_route(&router, Method::POST, &admin_path("/v3/service")); assert_route(&router, Method::GET, &admin_path("/v3/info")); @@ -257,6 +259,7 @@ fn test_admin_alias_paths_match_existing_admin_routes() { (Method::PUT, compat_admin_alias_path("/v3/restore-config-history-kv")), (Method::GET, compat_admin_alias_path("/v3/config")), (Method::PUT, compat_admin_alias_path("/v3/config")), + (Method::GET, compat_admin_alias_path("/v3/scanner/status")), ] { assert!( router.contains_compatible_route(method.clone(), &path), diff --git a/rustfs/src/admin/service/config.rs b/rustfs/src/admin/service/config.rs index 25c14b17b..6f48f80a7 100644 --- a/rustfs/src/admin/service/config.rs +++ b/rustfs/src/admin/service/config.rs @@ -13,6 +13,7 @@ // limitations under the License. use rustfs_audit::reload_audit_config; +use rustfs_config::SCANNER_SUB_SYS; use rustfs_config::audit::{AUDIT_MQTT_SUB_SYS, AUDIT_REDIS_DEFAULT_CHANNEL, AUDIT_WEBHOOK_SUB_SYS}; use rustfs_config::notify::{NOTIFY_MQTT_SUB_SYS, NOTIFY_REDIS_DEFAULT_CHANNEL, NOTIFY_WEBHOOK_SUB_SYS}; use rustfs_config::oidc::IDENTITY_OPENID_SUB_SYS; @@ -34,7 +35,10 @@ use tracing::warn; use url::Url; pub fn is_dynamic_config_subsystem(sub_system: &str) -> bool { - matches!(sub_system, STORAGE_CLASS_SUB_SYS | AUDIT_WEBHOOK_SUB_SYS | AUDIT_MQTT_SUB_SYS) + matches!( + sub_system, + STORAGE_CLASS_SUB_SYS | AUDIT_WEBHOOK_SUB_SYS | AUDIT_MQTT_SUB_SYS | SCANNER_SUB_SYS + ) } fn internal_error(message: impl Into) -> S3Error { @@ -231,6 +235,8 @@ pub async fn validate_server_config(config: &ServerConfig, sub_system: Option<&s Some(AUDIT_WEBHOOK_SUB_SYS) => validate_audit_subsystem_config(config, AUDIT_WEBHOOK_SUB_SYS), Some(AUDIT_MQTT_SUB_SYS) => validate_audit_subsystem_config(config, AUDIT_MQTT_SUB_SYS), Some(IDENTITY_OPENID_SUB_SYS) => validate_identity_openid_config(config), + Some(SCANNER_SUB_SYS) => rustfs_scanner::validate_scanner_runtime_config(config) + .map_err(|err| invalid_request(format!("invalid scanner config: {err}"))), Some(_) => Ok(()), None => { validate_storage_class_config(config).await?; @@ -239,6 +245,8 @@ pub async fn validate_server_config(config: &ServerConfig, sub_system: Option<&s validate_audit_subsystem_config(config, AUDIT_WEBHOOK_SUB_SYS)?; validate_audit_subsystem_config(config, AUDIT_MQTT_SUB_SYS)?; validate_identity_openid_config(config)?; + rustfs_scanner::validate_scanner_runtime_config(config) + .map_err(|err| invalid_request(format!("invalid scanner config: {err}")))?; Ok(()) } } @@ -254,6 +262,8 @@ pub async fn apply_dynamic_config_for_subsystem(config: &ServerConfig, sub_syste AUDIT_WEBHOOK_SUB_SYS | AUDIT_MQTT_SUB_SYS => reload_audit_config(config.clone()) .await .map_err(|err| internal_error(format!("failed to reload audit config: {err}")))?, + SCANNER_SUB_SYS => rustfs_scanner::apply_scanner_runtime_config(config) + .map_err(|err| internal_error(format!("failed to reload scanner config: {err}")))?, _ => return Ok(false), } @@ -292,7 +302,12 @@ pub async fn reload_runtime_config_snapshot() -> S3Result<()> { // Re-apply dynamic subsystems before publishing the snapshot, so that // runtime state (e.g. GLOBAL_STORAGE_CLASS) is refreshed on this peer. - for sub_system in [STORAGE_CLASS_SUB_SYS, AUDIT_WEBHOOK_SUB_SYS, AUDIT_MQTT_SUB_SYS] { + for sub_system in [ + STORAGE_CLASS_SUB_SYS, + AUDIT_WEBHOOK_SUB_SYS, + AUDIT_MQTT_SUB_SYS, + SCANNER_SUB_SYS, + ] { if let Err(err) = apply_dynamic_config_for_subsystem(&config, sub_system).await { warn!("peer reload_runtime_config_snapshot: failed to apply {sub_system}: {err}"); } @@ -333,6 +348,7 @@ pub async fn signal_config_snapshot_reload() { #[cfg(test)] mod tests { use super::*; + use rustfs_config::SCANNER_SUB_SYS; use rustfs_config::notify::NOTIFY_WEBHOOK_SUB_SYS; use rustfs_config::oidc::{OIDC_CLIENT_ID, OIDC_CONFIG_URL, OIDC_SCOPES}; use rustfs_config::{MQTT_BROKER, MQTT_QUEUE_DIR, MQTT_TOPIC, WEBHOOK_ENDPOINT, WEBHOOK_QUEUE_DIR}; @@ -341,6 +357,7 @@ mod tests { fn dynamic_config_subsystems_match_runtime_apply_support() { assert!(is_dynamic_config_subsystem(AUDIT_WEBHOOK_SUB_SYS)); assert!(is_dynamic_config_subsystem(AUDIT_MQTT_SUB_SYS)); + assert!(is_dynamic_config_subsystem(SCANNER_SUB_SYS)); assert!(is_dynamic_config_subsystem(STORAGE_CLASS_SUB_SYS)); assert!(!is_dynamic_config_subsystem("identity_openid")); assert!(!is_dynamic_config_subsystem("notify_webhook"));