gsadmin/rustfs
1
0
Fork 0
mirror of https://github.com/rustfs/rustfs.git synced 2026-08-05 21:07:43 +00:00
Code Issues Packages Projects Releases Wiki Activity

feat(scanner): add runtime scanner controls and status (#3203)

Browse Source 
* 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 <marshawcoco@users.noreply.github.com>
Co-authored-by: houseme <housemecn@gmail.com>
This commit is contained in:
Henry Guo
2026-06-04 13:16:26 +08:00
committed by GitHub
parent f708c22b0e
commit 6850457707
16 changed files with 1394 additions and 126 deletions
Download Patch File Download Diff File Expand all files Collapse all files
crates/config/src/constants/scanner.rs
+84 -6
View File
@@ -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<Self> {
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 {
crates/ecstore/src/config/mod.rs
+15
View File
@@ -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");
}
}
crates/ecstore/src/config/scanner.rs
+106
View File
@@ -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<KVS> = 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,
},
])
});
crates/scanner/src/data_usage_define.rs
+8 -5
View File
@@ -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]
crates/scanner/src/lib.rs
+2
View File
@@ -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};
crates/scanner/src/runtime_config.rs
+819
View File
@@ -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<RwLock<ScannerRuntimeConfig>> =
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<Duration>,
pub(crate) start_delay_source: ScannerRuntimeConfigSource,
pub(crate) cycle_interval: Duration,
pub(crate) cycle_interval_source: ScannerRuntimeConfigSource,
pub(crate) bitrot_cycle: Option<Duration>,
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<T> {
pub value: T,
pub source: ScannerRuntimeConfigSource,
}
#[derive(Clone, Debug, Serialize)]
pub struct ScannerRuntimeConfigStatus {
pub speed: ScannerRuntimeConfigValue<String>,
pub idle_mode: ScannerRuntimeConfigValue<bool>,
pub start_delay_seconds: ScannerRuntimeConfigValue<Option<u64>>,
pub cycle_interval_seconds: ScannerRuntimeConfigValue<u64>,
pub bitrot_cycle_seconds: ScannerRuntimeConfigValue<Option<u64>>,
pub cycle_max_duration_seconds: ScannerRuntimeConfigValue<Option<u64>>,
pub cycle_max_objects: ScannerRuntimeConfigValue<Option<u64>>,
pub cycle_max_directories: ScannerRuntimeConfigValue<Option<u64>>,
pub cache_save_timeout_seconds: ScannerRuntimeConfigValue<u64>,
pub max_concurrent_set_scans: ScannerRuntimeConfigValue<usize>,
pub max_concurrent_disk_scans: ScannerRuntimeConfigValue<usize>,
pub yield_every_n_objects: ScannerRuntimeConfigValue<u64>,
pub alert_excess_versions: ScannerRuntimeConfigValue<u64>,
pub alert_excess_version_size: ScannerRuntimeConfigValue<u64>,
pub alert_excess_folders: ScannerRuntimeConfigValue<u64>,
}
pub(crate) fn set_scanner_default_cycle_secs(secs: Option<u64>) {
SCANNER_DEFAULT_CYCLE_SECS.store(secs.unwrap_or(NO_DEFAULT_CYCLE_OVERRIDE), Ordering::Relaxed);
}
fn scanner_default_cycle_secs() -> Option<u64> {
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<String> {
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<KVS> {
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<String>, reason: &'static str) -> ScannerRuntimeConfigError {
ScannerRuntimeConfigError::InvalidValue {
key,
value: value.into(),
reason,
}
}
fn parse_config_u64(key: &'static str, value: String) -> Result<u64, ScannerRuntimeConfigError> {
value
.parse::<u64>()
.map_err(|_| invalid_value(key, value, "expected unsigned integer seconds or count"))
}
fn parse_config_usize(key: &'static str, value: String) -> Result<usize, ScannerRuntimeConfigError> {
value
.parse::<usize>()
.map_err(|_| invalid_value(key, value, "expected unsigned integer count"))
}
fn parse_config_bool(key: &'static str, value: String) -> Result<bool, ScannerRuntimeConfigError> {
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, ScannerRuntimeConfigError> {
ScannerSpeed::parse_str(&value).ok_or_else(|| invalid_value(SCANNER_SPEED, value, "expected scanner speed preset"))
}
fn parse_config_bitrot_cycle(value: String) -> Result<Option<Duration>, 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<Duration> {
match value.trim().to_ascii_lowercase().as_str() {
"0" | "true" | "on" | "yes" => Some(Duration::ZERO),
"false" | "off" | "no" | "disabled" => None,
value => value.parse::<u64>().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<Duration>, 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<Duration>, 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<u64>, 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<ScannerRuntimeConfig, ScannerRuntimeConfigError> {
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<Duration> {
current_scanner_runtime_config().start_delay
}
pub(crate) fn scanner_bitrot_cycle() -> Option<Duration> {
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();
}
}
crates/scanner/src/scanner.rs
+44 -78
View File
@@ -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<u64> {
match SCANNER_DEFAULT_CYCLE_SECS.load(Ordering::Relaxed) {
NO_DEFAULT_CYCLE_OVERRIDE => None,
secs => Some(secs),
}
}
fn set_scanner_default_cycle_secs(secs: Option<u64>) {
SCANNER_DEFAULT_CYCLE_SECS.store(secs.unwrap_or(NO_DEFAULT_CYCLE_OVERRIDE), Ordering::Relaxed);
}
fn scanner_start_delay_secs() -> Option<u64> {
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<Duration> {
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<u64> {
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<Duration> {
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<ScannerCycleBudgetReason>) -> 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<u64>) -> Duration {
@@ -150,6 +124,9 @@ pub async fn init_data_scanner(ctx: CancellationToken, storeapi: Arc<ECStore>) {
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<ECStore>) {
}
}
#[cfg(test)]
fn bitrot_scan_cycle() -> Option<Duration> {
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::<u64>().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<ECStore>, 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]
crates/scanner/src/scanner_folder.rs
+17 -13
View File
@@ -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]
crates/scanner/src/scanner_io.rs
+8 -12
View File
@@ -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<Error>, 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]
crates/scanner/src/sleeper.rs
+15 -3
View File
@@ -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,
);
}
}
rustfs/src/admin/handlers/config_admin.rs
+153 -5
View File
@@ -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");
rustfs/src/admin/handlers/mod.rs
+2
View File
@@ -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 {};
rustfs/src/admin/handlers/scanner.rs
+96
View File
@@ -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<AdminOperation>) -> 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<Body>) -> S3Result<Credentials> {
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::<Option<RemoteAddr>>()
.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<u8>) -> S3Result<S3Response<(StatusCode, Body)>> {
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<Body>, _params: Params<'_, '_>) -> S3Result<S3Response<(StatusCode, Body)>> {
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)
}
}
rustfs/src/admin/mod.rs
+2 -1
View File
@@ -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<impl S3Route>
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)?;
rustfs/src/admin/route_registration_test.rs
+4 -1
View File
@@ -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<AdminOperation>) {
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),
rustfs/src/admin/service/config.rs
+19 -2
View File
@@ -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<String>) -> 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"));