mirror of
https://github.com/rustfs/rustfs.git
synced 2026-10-31 16:22:06 +00:00
Hauser
900d082764
* fix(scanner): bound implicit startup delay and publish config early Limit warm-cache startup to the smaller of the cycle interval and one minute before jitter, while preserving explicit startup settings. Publish configuration gauges when applying runtime settings and retain timeout diagnostics until an actual new cycle begins. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(scanner): unify startup waits and skip unchanged configuration Bound read-only startup inspection and make startup and pre-leadership retries respond to cancellation and configuration changes. Preserve one startup deadline and discover replication behind lifecycle-only buckets. Avoid irrelevant bucket listings and repeated per-disk reconfiguration. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> * fix(scanner): retain effective startup policy and fallback metrics Keep the sampled startup deadline when configuration changes leave its normalized policy unchanged. Initialize retained runtime metrics after an invalid first refresh without suppressing the validation error or replacing a newer accepted configuration. Co-Authored-By: heihutu <heihutu@gmail.com> Co-Authored-By: zhi22915 <qiuzgang@gmail.com> --------- Co-authored-by: heihutu <heihutu@gmail.com> Co-authored-by: zhi22915 <qiuzgang@gmail.com>
1796 lines
81 KiB
Rust
1796 lines
81 KiB
Rust
// 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::resolve_scanner_server_config;
|
|
use crate::scanner_budget::ScannerCycleBudgetConfig;
|
|
use crate::sleeper::{SCANNER_SLEEPER, ScannerDiskIoBudgetConfig, ScannerIoBudgetLimit, scanner_default_speed};
|
|
use rustfs_config::{
|
|
DEFAULT_DELIMITER, DEFAULT_HEAL_BITROT_CYCLE_SECS, DEFAULT_SCANNER_ALERT_EXCESS_FOLDERS,
|
|
DEFAULT_SCANNER_ALERT_EXCESS_VERSION_SIZE, DEFAULT_SCANNER_ALERT_EXCESS_VERSIONS, DEFAULT_SCANNER_CACHE_SAVE_TIMEOUT_SECS,
|
|
DEFAULT_SCANNER_CYCLE_MAX_DIRECTORIES, DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS, DEFAULT_SCANNER_CYCLE_MAX_OBJECTS,
|
|
DEFAULT_SCANNER_DISK_IO_BUDGET, DEFAULT_SCANNER_IDLE_MODE, DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS,
|
|
DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS, DEFAULT_SCANNER_RECOVERY_BACKOFF_CAP_SECS, DEFAULT_SCANNER_SPEED,
|
|
DEFAULT_SCANNER_WALKERS_PER_DISK, 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_DELAY,
|
|
ENV_SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC, ENV_SCANNER_DISK_IO_BUDGET_IOPS, ENV_SCANNER_IDLE_MODE,
|
|
ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS, ENV_SCANNER_MAX_CONCURRENT_SET_SCANS, ENV_SCANNER_MAX_WAIT_SECS,
|
|
ENV_SCANNER_RECOVERY_BACKOFF_CAP_SECS, ENV_SCANNER_SPEED, ENV_SCANNER_START_DELAY_SECS, ENV_SCANNER_WALKERS_PER_DISK,
|
|
ENV_SCANNER_YIELD_EVERY_N_OBJECTS, HEAL_BITROT_CYCLE, HEAL_SUB_SYS, MAX_SCANNER_WALKERS_PER_DISK,
|
|
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_DELAY, SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC, SCANNER_DISK_IO_BUDGET_IOPS,
|
|
SCANNER_IDLE_MODE, SCANNER_MAX_CONCURRENT_DISK_SCANS, SCANNER_MAX_CONCURRENT_SET_SCANS, SCANNER_MAX_WAIT,
|
|
SCANNER_RECOVERY_BACKOFF_CAP, SCANNER_SPEED, SCANNER_START_DELAY, SCANNER_SUB_SYS, SCANNER_WALKERS_PER_DISK,
|
|
SCANNER_YIELD_EVERY_N_OBJECTS, ScannerSpeed,
|
|
server_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 tokio::sync::Notify;
|
|
use tracing::warn;
|
|
|
|
const LOG_COMPONENT_SCANNER: &str = "scanner";
|
|
#[cfg(test)]
|
|
const LOG_SUBSYSTEM_RUNTIME: &str = "runtime";
|
|
const LOG_SUBSYSTEM_RUNTIME_CONFIG: &str = "runtime_config";
|
|
#[cfg(test)]
|
|
const EVENT_SCANNER_RUNTIME_CONFIG: &str = "scanner_runtime_config";
|
|
const EVENT_SCANNER_RUNTIME_CONFIG_PARSE: &str = "scanner_runtime_config_parse";
|
|
|
|
const ENV_SCANNER_START_DELAY_SECS_DEPRECATED: &str = "RUSTFS_DATA_SCANNER_START_DELAY_SECS";
|
|
const NO_DEFAULT_CYCLE_OVERRIDE: u64 = 0;
|
|
const MAX_SCANNER_DELAY_FACTOR: f64 = 10_000.0;
|
|
const SCANNER_DELAY_RANGE_REASON: &str = "expected scanner delay between 0 and 10000";
|
|
const SCANNER_WALKERS_PER_DISK_RANGE_REASON: &str = "expected walkers per disk between 1 and 16";
|
|
const SCANNER_DISK_IO_BUDGET_REASON: &str = "expected auto, off, or a non-negative integer rate";
|
|
|
|
static SCANNER_DEFAULT_CYCLE_SECS: AtomicU64 = AtomicU64::new(NO_DEFAULT_CYCLE_OVERRIDE);
|
|
static SCANNER_RUNTIME_CONFIG_GENERATION: AtomicU64 = AtomicU64::new(0);
|
|
static SCANNER_RUNTIME_CONFIG_NOTIFY: LazyLock<Notify> = LazyLock::new(Notify::new);
|
|
|
|
struct ScannerRuntimeConfigState {
|
|
config: ScannerRuntimeConfig,
|
|
applied: bool,
|
|
}
|
|
|
|
static SCANNER_RUNTIME_CONFIG: LazyLock<RwLock<ScannerRuntimeConfigState>> = LazyLock::new(|| {
|
|
RwLock::new(ScannerRuntimeConfigState {
|
|
config: lookup_scanner_runtime_config(None).unwrap_or_default(),
|
|
applied: false,
|
|
})
|
|
});
|
|
|
|
#[derive(Clone, Copy, Debug, PartialEq, Eq, Serialize)]
|
|
#[serde(rename_all = "snake_case")]
|
|
pub enum ScannerRuntimeConfigSource {
|
|
Env,
|
|
Config,
|
|
ScannerCompatConfig,
|
|
Default,
|
|
}
|
|
|
|
#[derive(Clone, Debug, PartialEq)]
|
|
pub(crate) struct ScannerRuntimeConfig {
|
|
pub(crate) speed: ScannerSpeed,
|
|
pub(crate) speed_source: ScannerRuntimeConfigSource,
|
|
pub(crate) delay: f64,
|
|
pub(crate) delay_source: ScannerRuntimeConfigSource,
|
|
pub(crate) max_wait: Duration,
|
|
pub(crate) max_wait_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) recovery_backoff_cap: Duration,
|
|
pub(crate) recovery_backoff_cap_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) walkers_per_disk: usize,
|
|
pub(crate) walkers_per_disk_source: ScannerRuntimeConfigSource,
|
|
pub(crate) disk_io_budget_iops: ScannerIoBudgetLimit,
|
|
pub(crate) disk_io_budget_iops_source: ScannerRuntimeConfigSource,
|
|
pub(crate) disk_io_budget_bytes_per_sec: ScannerIoBudgetLimit,
|
|
pub(crate) disk_io_budget_bytes_per_sec_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,
|
|
delay: scanner_default_speed().sleep_factor(),
|
|
delay_source: ScannerRuntimeConfigSource::Default,
|
|
max_wait: scanner_default_speed().max_sleep(),
|
|
max_wait_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_HEAL_BITROT_CYCLE_SECS)),
|
|
bitrot_cycle_source: ScannerRuntimeConfigSource::Default,
|
|
cycle_budget: ScannerCycleBudgetConfig {
|
|
max_duration: Some(Duration::from_secs(DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS)),
|
|
..Default::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,
|
|
recovery_backoff_cap: Duration::from_secs(DEFAULT_SCANNER_RECOVERY_BACKOFF_CAP_SECS),
|
|
recovery_backoff_cap_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,
|
|
walkers_per_disk: DEFAULT_SCANNER_WALKERS_PER_DISK,
|
|
walkers_per_disk_source: ScannerRuntimeConfigSource::Default,
|
|
disk_io_budget_iops: ScannerIoBudgetLimit::Auto,
|
|
disk_io_budget_iops_source: ScannerRuntimeConfigSource::Default,
|
|
disk_io_budget_bytes_per_sec: ScannerIoBudgetLimit::Auto,
|
|
disk_io_budget_bytes_per_sec_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 delay: ScannerRuntimeConfigValue<f64>,
|
|
pub max_wait_seconds: ScannerRuntimeConfigValue<f64>,
|
|
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 recovery_backoff_cap_seconds: ScannerRuntimeConfigValue<u64>,
|
|
pub max_concurrent_set_scans: ScannerRuntimeConfigValue<usize>,
|
|
pub max_concurrent_disk_scans: ScannerRuntimeConfigValue<usize>,
|
|
pub walkers_per_disk: ScannerRuntimeConfigValue<usize>,
|
|
/// `auto`, `off`, or an explicit rate, as configured.
|
|
pub disk_io_budget_iops: ScannerRuntimeConfigValue<String>,
|
|
pub disk_io_budget_bytes_per_sec: ScannerRuntimeConfigValue<String>,
|
|
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 heal_kvs(config: Option<&ServerConfig>) -> Option<KVS> {
|
|
config.and_then(|config| config.get_value(HEAL_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 validate_default_heal_target(config: &ServerConfig) -> Result<(), ScannerRuntimeConfigError> {
|
|
let Some(targets) = config.0.get(HEAL_SUB_SYS) else {
|
|
return Ok(());
|
|
};
|
|
|
|
for target in targets.keys() {
|
|
if target != DEFAULT_DELIMITER {
|
|
return Err(invalid_value("target", target.clone(), "heal 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_io_budget(key: &'static str, value: String) -> Result<ScannerIoBudgetLimit, ScannerRuntimeConfigError> {
|
|
ScannerIoBudgetLimit::parse(&value).ok_or_else(|| invalid_value(key, value, SCANNER_DISK_IO_BUDGET_REASON))
|
|
}
|
|
|
|
fn lookup_io_budget(
|
|
kvs: Option<&KVS>,
|
|
key: &'static str,
|
|
env_key: &'static str,
|
|
) -> Result<(ScannerIoBudgetLimit, ScannerRuntimeConfigSource), ScannerRuntimeConfigError> {
|
|
if let Some(value) = rustfs_utils::get_env_opt_str(env_key) {
|
|
return parse_config_io_budget(env_key, value).map(|limit| (limit, ScannerRuntimeConfigSource::Env));
|
|
}
|
|
if let Some(value) = config_value(kvs, key, DEFAULT_SCANNER_DISK_IO_BUDGET) {
|
|
return parse_config_io_budget(key, value).map(|limit| (limit, ScannerRuntimeConfigSource::Config));
|
|
}
|
|
Ok((ScannerIoBudgetLimit::Auto, ScannerRuntimeConfigSource::Default))
|
|
}
|
|
|
|
fn parse_config_delay(key: &'static str, value: String) -> Result<f64, ScannerRuntimeConfigError> {
|
|
let parsed = value
|
|
.parse::<f64>()
|
|
.map_err(|_| invalid_value(key, value.clone(), SCANNER_DELAY_RANGE_REASON))?;
|
|
if parsed.is_finite() && (0.0..=MAX_SCANNER_DELAY_FACTOR).contains(&parsed) {
|
|
Ok(parsed)
|
|
} else {
|
|
Err(invalid_value(key, value, SCANNER_DELAY_RANGE_REASON))
|
|
}
|
|
}
|
|
|
|
fn parse_env_delay(key: &'static str, value: String, fallback: f64) -> f64 {
|
|
match parse_config_delay(key, value.clone()) {
|
|
Ok(parsed) => parsed,
|
|
Err(_) => {
|
|
warn!(
|
|
target: "rustfs::scanner::runtime_config",
|
|
event = EVENT_SCANNER_RUNTIME_CONFIG_PARSE,
|
|
component = LOG_COMPONENT_SCANNER,
|
|
subsystem = LOG_SUBSYSTEM_RUNTIME_CONFIG,
|
|
env = key,
|
|
value,
|
|
fallback,
|
|
state = "invalid_delay",
|
|
"Scanner runtime config used derived delay fallback"
|
|
);
|
|
fallback
|
|
}
|
|
}
|
|
}
|
|
|
|
fn parse_config_bitrot_cycle(key: &'static str, 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(key, 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!(
|
|
target: "rustfs::scanner::runtime_config",
|
|
event = EVENT_SCANNER_RUNTIME_CONFIG_PARSE,
|
|
component = LOG_COMPONENT_SCANNER,
|
|
subsystem = LOG_SUBSYSTEM_RUNTIME_CONFIG,
|
|
env = ENV_SCANNER_BITROT_CYCLE_SECS,
|
|
value,
|
|
default_secs = DEFAULT_HEAL_BITROT_CYCLE_SECS,
|
|
state = "invalid_bitrot_cycle",
|
|
"Scanner runtime config used default bitrot cycle"
|
|
);
|
|
Some(Duration::from_secs(DEFAULT_HEAL_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)?;
|
|
validate_default_heal_target(config)?;
|
|
|
|
let scanner_kvs = scanner_kvs(Some(config));
|
|
let scanner_kvs = scanner_kvs.as_ref();
|
|
let heal_kvs = heal_kvs(Some(config));
|
|
let heal_kvs = heal_kvs.as_ref();
|
|
|
|
if let Some(value) = config_value(scanner_kvs, SCANNER_SPEED, DEFAULT_SCANNER_SPEED) {
|
|
parse_config_speed(value)?;
|
|
}
|
|
if let Some(value) = config_value(scanner_kvs, SCANNER_DELAY, "") {
|
|
parse_config_delay(SCANNER_DELAY, value)?;
|
|
}
|
|
validate_optional_config_u64(scanner_kvs, SCANNER_MAX_WAIT, "")?;
|
|
if let Some(value) = config_value(scanner_kvs, SCANNER_IDLE_MODE, DEFAULT_SCANNER_IDLE_MODE) {
|
|
parse_config_bool(SCANNER_IDLE_MODE, value)?;
|
|
}
|
|
validate_optional_config_u64(scanner_kvs, SCANNER_START_DELAY, "")?;
|
|
validate_optional_config_u64(scanner_kvs, SCANNER_CYCLE, "")?;
|
|
if let Some(value) = config_value(scanner_kvs, SCANNER_CYCLE_MAX_DURATION, DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS) {
|
|
let secs = parse_config_u64(SCANNER_CYCLE_MAX_DURATION, value)?;
|
|
cycle_duration_from_secs(SCANNER_CYCLE_MAX_DURATION, secs)?;
|
|
}
|
|
validate_optional_config_u64(scanner_kvs, SCANNER_CYCLE_MAX_OBJECTS, DEFAULT_SCANNER_CYCLE_MAX_OBJECTS)?;
|
|
validate_optional_config_u64(scanner_kvs, SCANNER_CYCLE_MAX_DIRECTORIES, DEFAULT_SCANNER_CYCLE_MAX_DIRECTORIES)?;
|
|
if let Some(value) = config_value(heal_kvs, HEAL_BITROT_CYCLE, DEFAULT_HEAL_BITROT_CYCLE_SECS) {
|
|
parse_config_bitrot_cycle(HEAL_BITROT_CYCLE, value)?;
|
|
}
|
|
if let Some(value) = config_value(scanner_kvs, SCANNER_BITROT_CYCLE, DEFAULT_HEAL_BITROT_CYCLE_SECS) {
|
|
parse_config_bitrot_cycle(SCANNER_BITROT_CYCLE, value)?;
|
|
}
|
|
validate_optional_config_u64(scanner_kvs, SCANNER_CACHE_SAVE_TIMEOUT, DEFAULT_SCANNER_CACHE_SAVE_TIMEOUT_SECS)?;
|
|
if let Some(value) = config_value(scanner_kvs, SCANNER_RECOVERY_BACKOFF_CAP, DEFAULT_SCANNER_RECOVERY_BACKOFF_CAP_SECS) {
|
|
recovery_backoff_cap_from_secs(SCANNER_RECOVERY_BACKOFF_CAP, parse_config_u64(SCANNER_RECOVERY_BACKOFF_CAP, value)?)?;
|
|
}
|
|
validate_optional_config_usize(scanner_kvs, SCANNER_MAX_CONCURRENT_SET_SCANS, DEFAULT_SCANNER_MAX_CONCURRENT_SET_SCANS)?;
|
|
validate_optional_config_usize(scanner_kvs, SCANNER_MAX_CONCURRENT_DISK_SCANS, DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS)?;
|
|
if let Some(value) = config_value(scanner_kvs, SCANNER_WALKERS_PER_DISK, DEFAULT_SCANNER_WALKERS_PER_DISK) {
|
|
walkers_per_disk_from_count(SCANNER_WALKERS_PER_DISK, parse_config_usize(SCANNER_WALKERS_PER_DISK, value)?)?;
|
|
}
|
|
if let Some(value) = config_value(scanner_kvs, SCANNER_DISK_IO_BUDGET_IOPS, DEFAULT_SCANNER_DISK_IO_BUDGET) {
|
|
parse_config_io_budget(SCANNER_DISK_IO_BUDGET_IOPS, value)?;
|
|
}
|
|
if let Some(value) = config_value(scanner_kvs, SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC, DEFAULT_SCANNER_DISK_IO_BUDGET) {
|
|
parse_config_io_budget(SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC, value)?;
|
|
}
|
|
validate_optional_config_u64(scanner_kvs, SCANNER_YIELD_EVERY_N_OBJECTS, DEFAULT_SCANNER_YIELD_EVERY_N_OBJECTS)?;
|
|
validate_optional_config_u64(scanner_kvs, SCANNER_ALERT_EXCESS_VERSIONS, DEFAULT_SCANNER_ALERT_EXCESS_VERSIONS)?;
|
|
validate_optional_config_u64(scanner_kvs, SCANNER_ALERT_EXCESS_VERSION_SIZE, DEFAULT_SCANNER_ALERT_EXCESS_VERSION_SIZE)?;
|
|
validate_optional_config_u64(scanner_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_delay(
|
|
kvs: Option<&KVS>,
|
|
speed: ScannerSpeed,
|
|
speed_source: ScannerRuntimeConfigSource,
|
|
) -> Result<(f64, ScannerRuntimeConfigSource), ScannerRuntimeConfigError> {
|
|
let derived = speed.sleep_factor();
|
|
if let Some(value) = rustfs_utils::get_env_opt_str(ENV_SCANNER_DELAY) {
|
|
return Ok((parse_env_delay(ENV_SCANNER_DELAY, value, derived), ScannerRuntimeConfigSource::Env));
|
|
}
|
|
if let Some(value) = config_value(kvs, SCANNER_DELAY, "") {
|
|
return parse_config_delay(SCANNER_DELAY, value).map(|delay| (delay, ScannerRuntimeConfigSource::Config));
|
|
}
|
|
Ok((derived, speed_source))
|
|
}
|
|
|
|
fn lookup_max_wait(
|
|
kvs: Option<&KVS>,
|
|
speed: ScannerSpeed,
|
|
speed_source: ScannerRuntimeConfigSource,
|
|
) -> Result<(Duration, ScannerRuntimeConfigSource), ScannerRuntimeConfigError> {
|
|
if let Some(secs) = rustfs_utils::get_env_opt_u64(ENV_SCANNER_MAX_WAIT_SECS) {
|
|
return Ok((Duration::from_secs(secs), ScannerRuntimeConfigSource::Env));
|
|
}
|
|
if let Some(value) = config_value(kvs, SCANNER_MAX_WAIT, "") {
|
|
return parse_config_u64(SCANNER_MAX_WAIT, value)
|
|
.map(|secs| (Duration::from_secs(secs), ScannerRuntimeConfigSource::Config));
|
|
}
|
|
Ok((speed.max_sleep(), speed_source))
|
|
}
|
|
|
|
fn lookup_cycle_duration(kvs: Option<&KVS>) -> Result<(Option<Duration>, ScannerRuntimeConfigSource), ScannerRuntimeConfigError> {
|
|
match rustfs_utils::get_env_parse_outcome::<u64>(ENV_SCANNER_CYCLE_MAX_DURATION_SECS) {
|
|
rustfs_utils::EnvParseOutcome::Parsed(secs) => {
|
|
return cycle_duration_from_secs(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, secs)
|
|
.map(|duration| (duration, ScannerRuntimeConfigSource::Env));
|
|
}
|
|
rustfs_utils::EnvParseOutcome::Invalid => {
|
|
// Do not include the raw environment value in the typed error:
|
|
// deployments occasionally put sensitive material in inherited
|
|
// environment snapshots. The key still identifies the control.
|
|
return Err(invalid_value(
|
|
ENV_SCANNER_CYCLE_MAX_DURATION_SECS,
|
|
"<invalid>",
|
|
"expected unsigned integer seconds",
|
|
));
|
|
}
|
|
rustfs_utils::EnvParseOutcome::Absent => {}
|
|
}
|
|
|
|
if let Some(value) = config_value(kvs, SCANNER_CYCLE_MAX_DURATION, DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS) {
|
|
let secs = parse_config_u64(SCANNER_CYCLE_MAX_DURATION, value)?;
|
|
return cycle_duration_from_secs(SCANNER_CYCLE_MAX_DURATION, secs)
|
|
.map(|duration| (duration, ScannerRuntimeConfigSource::Config));
|
|
}
|
|
|
|
Ok((
|
|
Some(Duration::from_secs(DEFAULT_SCANNER_CYCLE_MAX_DURATION_SECS)),
|
|
ScannerRuntimeConfigSource::Default,
|
|
))
|
|
}
|
|
|
|
fn cycle_duration_from_secs(key: &'static str, secs: u64) -> Result<Option<Duration>, ScannerRuntimeConfigError> {
|
|
if secs == 0 {
|
|
return Ok(None);
|
|
}
|
|
let duration = Duration::from_secs(secs);
|
|
if std::time::Instant::now().checked_add(duration).is_none() {
|
|
return Err(invalid_value(key, "<overflow>", "duration exceeds the timer range"));
|
|
}
|
|
Ok(Some(duration))
|
|
}
|
|
|
|
fn recovery_backoff_cap_from_secs(key: &'static str, secs: u64) -> Result<Duration, ScannerRuntimeConfigError> {
|
|
if secs == 0 {
|
|
return Err(invalid_value(key, "0", "expected positive integer seconds"));
|
|
}
|
|
let duration = Duration::from_secs(secs);
|
|
if std::time::Instant::now().checked_add(duration).is_none() {
|
|
return Err(invalid_value(key, "<overflow>", "duration exceeds the timer range"));
|
|
}
|
|
Ok(duration)
|
|
}
|
|
|
|
fn lookup_recovery_backoff_cap(kvs: Option<&KVS>) -> Result<(Duration, ScannerRuntimeConfigSource), ScannerRuntimeConfigError> {
|
|
// Validate durable input even when an environment override takes precedence.
|
|
let persisted = config_value(kvs, SCANNER_RECOVERY_BACKOFF_CAP, DEFAULT_SCANNER_RECOVERY_BACKOFF_CAP_SECS)
|
|
.map(|value| {
|
|
recovery_backoff_cap_from_secs(SCANNER_RECOVERY_BACKOFF_CAP, parse_config_u64(SCANNER_RECOVERY_BACKOFF_CAP, value)?)
|
|
})
|
|
.transpose()?;
|
|
match rustfs_utils::get_env_parse_outcome::<u64>(ENV_SCANNER_RECOVERY_BACKOFF_CAP_SECS) {
|
|
rustfs_utils::EnvParseOutcome::Parsed(secs) => {
|
|
return recovery_backoff_cap_from_secs(ENV_SCANNER_RECOVERY_BACKOFF_CAP_SECS, secs)
|
|
.map(|duration| (duration, ScannerRuntimeConfigSource::Env));
|
|
}
|
|
rustfs_utils::EnvParseOutcome::Invalid => {
|
|
return Err(invalid_value(
|
|
ENV_SCANNER_RECOVERY_BACKOFF_CAP_SECS,
|
|
"<invalid>",
|
|
"expected positive integer seconds",
|
|
));
|
|
}
|
|
rustfs_utils::EnvParseOutcome::Absent => {}
|
|
}
|
|
if let Some(duration) = persisted {
|
|
return Ok((duration, ScannerRuntimeConfigSource::Config));
|
|
}
|
|
Ok((
|
|
Duration::from_secs(DEFAULT_SCANNER_RECOVERY_BACKOFF_CAP_SECS),
|
|
ScannerRuntimeConfigSource::Default,
|
|
))
|
|
}
|
|
|
|
fn walkers_per_disk_from_count(key: &'static str, count: usize) -> Result<usize, ScannerRuntimeConfigError> {
|
|
if (1..=MAX_SCANNER_WALKERS_PER_DISK).contains(&count) {
|
|
Ok(count)
|
|
} else {
|
|
Err(invalid_value(key, count.to_string(), SCANNER_WALKERS_PER_DISK_RANGE_REASON))
|
|
}
|
|
}
|
|
|
|
fn lookup_walkers_per_disk(kvs: Option<&KVS>) -> Result<(usize, ScannerRuntimeConfigSource), ScannerRuntimeConfigError> {
|
|
match rustfs_utils::get_env_parse_outcome::<usize>(ENV_SCANNER_WALKERS_PER_DISK) {
|
|
rustfs_utils::EnvParseOutcome::Parsed(count) => {
|
|
return walkers_per_disk_from_count(ENV_SCANNER_WALKERS_PER_DISK, count)
|
|
.map(|count| (count, ScannerRuntimeConfigSource::Env));
|
|
}
|
|
rustfs_utils::EnvParseOutcome::Invalid => {
|
|
return Err(invalid_value(
|
|
ENV_SCANNER_WALKERS_PER_DISK,
|
|
"<invalid>",
|
|
"expected unsigned integer count",
|
|
));
|
|
}
|
|
rustfs_utils::EnvParseOutcome::Absent => {}
|
|
}
|
|
|
|
if let Some(value) = config_value(kvs, SCANNER_WALKERS_PER_DISK, DEFAULT_SCANNER_WALKERS_PER_DISK) {
|
|
let count = parse_config_usize(SCANNER_WALKERS_PER_DISK, value)?;
|
|
return walkers_per_disk_from_count(SCANNER_WALKERS_PER_DISK, count)
|
|
.map(|count| (count, ScannerRuntimeConfigSource::Config));
|
|
}
|
|
|
|
Ok((DEFAULT_SCANNER_WALKERS_PER_DISK, 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 scanner_kvs = scanner_kvs(config);
|
|
let scanner_kvs = scanner_kvs.as_ref();
|
|
let heal_kvs = heal_kvs(config);
|
|
let heal_kvs = heal_kvs.as_ref();
|
|
let (speed, speed_source) = lookup_speed(scanner_kvs)?;
|
|
let (delay, delay_source) = lookup_delay(scanner_kvs, speed, speed_source)?;
|
|
let (max_wait, max_wait_source) = lookup_max_wait(scanner_kvs, speed, speed_source)?;
|
|
let (idle_mode, idle_mode_source) =
|
|
lookup_bool(scanner_kvs, SCANNER_IDLE_MODE, ENV_SCANNER_IDLE_MODE, DEFAULT_SCANNER_IDLE_MODE)?;
|
|
let (start_delay, start_delay_source) = lookup_start_delay(scanner_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(scanner_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_cycle_duration(scanner_kvs)?;
|
|
let (cycle_max_objects, cycle_max_objects_source) = lookup_count_budget(
|
|
scanner_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(
|
|
scanner_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(heal_kvs, HEAL_BITROT_CYCLE, DEFAULT_HEAL_BITROT_CYCLE_SECS) {
|
|
(parse_config_bitrot_cycle(HEAL_BITROT_CYCLE, value)?, ScannerRuntimeConfigSource::Config)
|
|
} else if let Some(value) = config_value(scanner_kvs, SCANNER_BITROT_CYCLE, DEFAULT_HEAL_BITROT_CYCLE_SECS) {
|
|
(
|
|
parse_config_bitrot_cycle(SCANNER_BITROT_CYCLE, value)?,
|
|
ScannerRuntimeConfigSource::ScannerCompatConfig,
|
|
)
|
|
} else {
|
|
(
|
|
Some(Duration::from_secs(DEFAULT_HEAL_BITROT_CYCLE_SECS)),
|
|
ScannerRuntimeConfigSource::Default,
|
|
)
|
|
};
|
|
|
|
let (cache_save_timeout, cache_save_timeout_source) = lookup_u64(
|
|
scanner_kvs,
|
|
SCANNER_CACHE_SAVE_TIMEOUT,
|
|
ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS,
|
|
DEFAULT_SCANNER_CACHE_SAVE_TIMEOUT_SECS,
|
|
)?;
|
|
let (recovery_backoff_cap, recovery_backoff_cap_source) = lookup_recovery_backoff_cap(scanner_kvs)?;
|
|
let (max_concurrent_set_scans, max_concurrent_set_scans_source) = lookup_usize(
|
|
scanner_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(
|
|
scanner_kvs,
|
|
SCANNER_MAX_CONCURRENT_DISK_SCANS,
|
|
ENV_SCANNER_MAX_CONCURRENT_DISK_SCANS,
|
|
DEFAULT_SCANNER_MAX_CONCURRENT_DISK_SCANS,
|
|
)?;
|
|
let (walkers_per_disk, walkers_per_disk_source) = lookup_walkers_per_disk(scanner_kvs)?;
|
|
let (disk_io_budget_iops, disk_io_budget_iops_source) =
|
|
lookup_io_budget(scanner_kvs, SCANNER_DISK_IO_BUDGET_IOPS, ENV_SCANNER_DISK_IO_BUDGET_IOPS)?;
|
|
let (disk_io_budget_bytes_per_sec, disk_io_budget_bytes_per_sec_source) = lookup_io_budget(
|
|
scanner_kvs,
|
|
SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC,
|
|
ENV_SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC,
|
|
)?;
|
|
let (yield_every_n_objects, yield_every_n_objects_source) = lookup_u64(
|
|
scanner_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(
|
|
scanner_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(
|
|
scanner_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(
|
|
scanner_kvs,
|
|
SCANNER_ALERT_EXCESS_FOLDERS,
|
|
ENV_SCANNER_ALERT_EXCESS_FOLDERS,
|
|
DEFAULT_SCANNER_ALERT_EXCESS_FOLDERS,
|
|
)?;
|
|
|
|
Ok(ScannerRuntimeConfig {
|
|
speed,
|
|
speed_source,
|
|
delay,
|
|
delay_source,
|
|
max_wait,
|
|
max_wait_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,
|
|
recovery_backoff_cap,
|
|
recovery_backoff_cap_source,
|
|
max_concurrent_set_scans,
|
|
max_concurrent_set_scans_source,
|
|
max_concurrent_disk_scans,
|
|
max_concurrent_disk_scans_source,
|
|
walkers_per_disk,
|
|
walkers_per_disk_source,
|
|
disk_io_budget_iops,
|
|
disk_io_budget_iops_source,
|
|
disk_io_budget_bytes_per_sec,
|
|
disk_io_budget_bytes_per_sec_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: Option<ScannerRuntimeConfig>) -> bool {
|
|
let Ok(mut state) = SCANNER_RUNTIME_CONFIG.write() else {
|
|
return false;
|
|
};
|
|
// Failed resolution can initialize the retained startup snapshot, but
|
|
// must not replace a newer configuration accepted by another caller.
|
|
let config = config.unwrap_or_else(|| state.config.clone());
|
|
if state.applied && state.config == config {
|
|
return false;
|
|
}
|
|
// Lock order: runtime configuration -> sleeper parameters / disk-budget registry.
|
|
SCANNER_SLEEPER.update_from_runtime_config(config.delay, config.max_wait, config.idle_mode, config.yield_every_n_objects);
|
|
SCANNER_SLEEPER.update_disk_io_budget(ScannerDiskIoBudgetConfig {
|
|
iops: config.disk_io_budget_iops,
|
|
bytes_per_sec: config.disk_io_budget_bytes_per_sec,
|
|
speed: config.speed,
|
|
});
|
|
rustfs_scanner_metrics::metrics::global_metrics().record_scanner_cycle_config(
|
|
config.cycle_interval,
|
|
config.bitrot_cycle,
|
|
config.cycle_budget.max_duration,
|
|
config.cycle_budget.max_objects,
|
|
config.cycle_budget.max_directories,
|
|
);
|
|
state.config = config;
|
|
state.applied = true;
|
|
true
|
|
}
|
|
|
|
fn current_server_config() -> Option<ServerConfig> {
|
|
resolve_scanner_server_config()
|
|
}
|
|
|
|
#[cfg(test)]
|
|
pub(crate) fn resolve_scanner_runtime_config_from_global() -> ScannerRuntimeConfig {
|
|
let config = current_server_config();
|
|
match lookup_scanner_runtime_config(config.as_ref()) {
|
|
Ok(config) => config,
|
|
Err(err) => {
|
|
warn!(
|
|
target: "rustfs::scanner",
|
|
event = EVENT_SCANNER_RUNTIME_CONFIG,
|
|
component = LOG_COMPONENT_SCANNER,
|
|
subsystem = LOG_SUBSYSTEM_RUNTIME,
|
|
state = "resolve_failed",
|
|
error = %err,
|
|
"Scanner runtime config fallback applied"
|
|
);
|
|
current_scanner_runtime_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))?;
|
|
if apply_resolved_runtime_config(Some(resolved)) {
|
|
SCANNER_RUNTIME_CONFIG_GENERATION.fetch_add(1, Ordering::AcqRel);
|
|
SCANNER_RUNTIME_CONFIG_NOTIFY.notify_one();
|
|
}
|
|
Ok(())
|
|
}
|
|
|
|
pub(crate) fn scanner_runtime_config_generation() -> u64 {
|
|
SCANNER_RUNTIME_CONFIG_GENERATION.load(Ordering::Acquire)
|
|
}
|
|
|
|
pub(crate) async fn scanner_runtime_config_changed() {
|
|
SCANNER_RUNTIME_CONFIG_NOTIFY.notified().await;
|
|
}
|
|
|
|
pub(crate) fn refresh_scanner_runtime_config_from_global() -> Result<(), ScannerRuntimeConfigError> {
|
|
let config = current_server_config();
|
|
match lookup_scanner_runtime_config(config.as_ref()) {
|
|
Ok(resolved) => {
|
|
apply_resolved_runtime_config(Some(resolved));
|
|
Ok(())
|
|
}
|
|
Err(error) => {
|
|
apply_resolved_runtime_config(None);
|
|
Err(error)
|
|
}
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
pub(crate) fn refresh_scanner_runtime_config_for_tests() {
|
|
if let Ok(resolved) = lookup_scanner_runtime_config(None) {
|
|
apply_resolved_runtime_config(Some(resolved));
|
|
}
|
|
}
|
|
|
|
pub(crate) fn current_scanner_runtime_config() -> ScannerRuntimeConfig {
|
|
SCANNER_RUNTIME_CONFIG
|
|
.read()
|
|
.map(|state| state.config.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,
|
|
},
|
|
delay: ScannerRuntimeConfigValue {
|
|
value: config.delay,
|
|
source: config.delay_source,
|
|
},
|
|
max_wait_seconds: ScannerRuntimeConfigValue {
|
|
value: config.max_wait.as_secs_f64(),
|
|
source: config.max_wait_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,
|
|
},
|
|
recovery_backoff_cap_seconds: ScannerRuntimeConfigValue {
|
|
value: config.recovery_backoff_cap.as_secs(),
|
|
source: config.recovery_backoff_cap_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,
|
|
},
|
|
walkers_per_disk: ScannerRuntimeConfigValue {
|
|
value: config.walkers_per_disk,
|
|
source: config.walkers_per_disk_source,
|
|
},
|
|
disk_io_budget_iops: ScannerRuntimeConfigValue {
|
|
value: config.disk_io_budget_iops.to_string(),
|
|
source: config.disk_io_budget_iops_source,
|
|
},
|
|
disk_io_budget_bytes_per_sec: ScannerRuntimeConfigValue {
|
|
value: config.disk_io_budget_bytes_per_sec.to_string(),
|
|
source: config.disk_io_budget_bytes_per_sec_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_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_recovery_backoff_cap() -> Duration {
|
|
current_scanner_runtime_config().recovery_backoff_cap
|
|
}
|
|
|
|
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_walkers_per_disk_configured() -> usize {
|
|
current_scanner_runtime_config().walkers_per_disk
|
|
}
|
|
|
|
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 crate::init_ecstore_config_for_scanner_tests;
|
|
use crate::sleeper::ScannerIoBudgetLimit;
|
|
use rustfs_config::server_config::{Config as ServerConfig, KVS};
|
|
use rustfs_config::{
|
|
DEFAULT_DELIMITER, DEFAULT_HEAL_BITROT_CYCLE_SECS, DEFAULT_SCANNER_RECOVERY_BACKOFF_CAP_SECS,
|
|
ENV_SCANNER_BITROT_CYCLE_SECS, ENV_SCANNER_CACHE_SAVE_TIMEOUT_SECS, ENV_SCANNER_CYCLE,
|
|
ENV_SCANNER_CYCLE_MAX_DURATION_SECS, ENV_SCANNER_CYCLE_MAX_OBJECTS, ENV_SCANNER_DELAY,
|
|
ENV_SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC, ENV_SCANNER_DISK_IO_BUDGET_IOPS, ENV_SCANNER_MAX_WAIT_SECS,
|
|
ENV_SCANNER_RECOVERY_BACKOFF_CAP_SECS, ENV_SCANNER_SPEED, ENV_SCANNER_WALKERS_PER_DISK, HEAL_BITROT_CYCLE, HEAL_SUB_SYS,
|
|
SCANNER_BITROT_CYCLE, SCANNER_CACHE_SAVE_TIMEOUT, SCANNER_CYCLE, SCANNER_CYCLE_MAX_DIRECTORIES,
|
|
SCANNER_CYCLE_MAX_DURATION, SCANNER_CYCLE_MAX_OBJECTS, SCANNER_DELAY, SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC,
|
|
SCANNER_DISK_IO_BUDGET_IOPS, SCANNER_IDLE_MODE, SCANNER_RECOVERY_BACKOFF_CAP, SCANNER_SPEED, SCANNER_SUB_SYS,
|
|
SCANNER_WALKERS_PER_DISK, ScannerSpeed,
|
|
};
|
|
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 {
|
|
init_ecstore_config_for_scanner_tests();
|
|
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_and_heal(scanner_entries: &[(&str, &str)], heal_entries: &[(&str, &str)]) -> ServerConfig {
|
|
let mut config = server_config_with_scanner(scanner_entries);
|
|
let mut kvs = KVS::new();
|
|
for (key, value) in heal_entries {
|
|
kvs.insert((*key).to_string(), (*value).to_string());
|
|
}
|
|
config
|
|
.0
|
|
.insert(HEAL_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
|
|
}
|
|
|
|
#[tokio::test]
|
|
#[serial]
|
|
async fn applying_runtime_config_publishes_cycle_metrics_before_first_cycle() {
|
|
let metrics = rustfs_scanner_metrics::metrics::global_metrics();
|
|
metrics.record_scanner_cycle_config(Duration::ZERO, None, None, None, None);
|
|
let before = metrics.report().await;
|
|
let mut config = super::ScannerRuntimeConfig {
|
|
cycle_interval: Duration::from_secs(21600),
|
|
bitrot_cycle: Some(Duration::from_secs(2592000)),
|
|
cycle_budget: crate::scanner_budget::ScannerCycleBudgetConfig {
|
|
max_duration: Some(Duration::from_secs(1800)),
|
|
max_objects: Some(1000),
|
|
max_directories: Some(8000),
|
|
},
|
|
..Default::default()
|
|
};
|
|
super::apply_resolved_runtime_config(Some(config.clone()));
|
|
let report = metrics.report().await;
|
|
assert_eq!(report.cycle_interval_seconds, 21600.0);
|
|
assert!(report.bitrot_cycle_enabled);
|
|
assert_eq!(report.bitrot_cycle_seconds, 2592000.0);
|
|
assert_eq!(report.cycle_max_duration_seconds, 1800.0);
|
|
assert_eq!(report.cycle_max_objects, 1000);
|
|
assert_eq!(report.cycle_max_directories, 8000);
|
|
assert_eq!(report.current_cycle, before.current_cycle);
|
|
assert_eq!(report.current_cycle_active, before.current_cycle_active);
|
|
assert_eq!(report.cycles_completed_at, before.cycles_completed_at);
|
|
|
|
config.bitrot_cycle = None;
|
|
config.cycle_budget = Default::default();
|
|
super::apply_resolved_runtime_config(Some(config));
|
|
let report = metrics.report().await;
|
|
assert!(!report.bitrot_cycle_enabled);
|
|
assert_eq!(report.bitrot_cycle_seconds, 0.0);
|
|
assert_eq!(report.cycle_max_duration_seconds, 0.0);
|
|
assert_eq!(report.cycle_max_objects, 0);
|
|
assert_eq!(report.cycle_max_directories, 0);
|
|
super::refresh_scanner_runtime_config_for_tests();
|
|
}
|
|
|
|
#[tokio::test]
|
|
#[serial]
|
|
async fn invalid_first_refresh_publishes_retained_config_without_reverting_newer_state() {
|
|
temp_env::async_with_vars([(ENV_SCANNER_WALKERS_PER_DISK, Some("0"))], async {
|
|
let retained = super::ScannerRuntimeConfig::default();
|
|
{
|
|
let mut state = super::SCANNER_RUNTIME_CONFIG.write().expect("runtime config state");
|
|
state.config = retained.clone();
|
|
state.applied = false;
|
|
}
|
|
let metrics = rustfs_scanner_metrics::metrics::global_metrics();
|
|
metrics.record_scanner_cycle_config(Duration::ZERO, None, None, None, None);
|
|
assert!(matches!(
|
|
super::refresh_scanner_runtime_config_from_global(),
|
|
Err(super::ScannerRuntimeConfigError::InvalidValue {
|
|
key: ENV_SCANNER_WALKERS_PER_DISK,
|
|
..
|
|
})
|
|
));
|
|
let report = metrics.report().await;
|
|
assert_eq!(report.cycle_interval_seconds, retained.cycle_interval.as_secs_f64());
|
|
assert_eq!(report.bitrot_cycle_enabled, retained.bitrot_cycle.is_some());
|
|
assert_eq!(
|
|
report.cycle_max_duration_seconds,
|
|
retained.cycle_budget.max_duration.unwrap_or_default().as_secs_f64()
|
|
);
|
|
assert!(super::SCANNER_RUNTIME_CONFIG.read().expect("runtime config state").applied);
|
|
|
|
let newer = super::ScannerRuntimeConfig {
|
|
cycle_interval: Duration::from_secs(120),
|
|
..retained
|
|
};
|
|
super::apply_resolved_runtime_config(Some(newer.clone()));
|
|
metrics.record_scanner_cycle_timeout(true, Duration::from_secs(17));
|
|
assert!(matches!(
|
|
super::refresh_scanner_runtime_config_from_global(),
|
|
Err(super::ScannerRuntimeConfigError::InvalidValue {
|
|
key: ENV_SCANNER_WALKERS_PER_DISK,
|
|
..
|
|
})
|
|
));
|
|
assert_eq!(super::current_scanner_runtime_config(), newer);
|
|
let report = metrics.report().await;
|
|
assert_eq!(report.cycle_interval_seconds, 120.0);
|
|
assert_eq!(report.cycle_last_progress_age, 17);
|
|
assert!(report.leader_lease_without_progress);
|
|
})
|
|
.await;
|
|
super::refresh_scanner_runtime_config_for_tests();
|
|
}
|
|
|
|
#[tokio::test]
|
|
#[serial]
|
|
async fn first_runtime_config_application_publishes_default_equal_values() {
|
|
let config = lookup_scanner_runtime_config(None).expect("initial resolved configuration");
|
|
{
|
|
let mut state = super::SCANNER_RUNTIME_CONFIG.write().expect("runtime config state");
|
|
state.config = config.clone();
|
|
state.applied = false;
|
|
}
|
|
let metrics = rustfs_scanner_metrics::metrics::global_metrics();
|
|
metrics.record_scanner_cycle_config(Duration::ZERO, None, None, None, None);
|
|
assert!(super::apply_resolved_runtime_config(Some(config.clone())));
|
|
assert_eq!(metrics.report().await.cycle_interval_seconds, config.cycle_interval.as_secs_f64());
|
|
metrics.record_scanner_cycle_timeout(true, Duration::from_secs(17));
|
|
assert!(!super::apply_resolved_runtime_config(Some(config)));
|
|
let unchanged = metrics.report().await;
|
|
assert_eq!(unchanged.cycle_last_progress_age, 17);
|
|
assert!(unchanged.leader_lease_without_progress);
|
|
}
|
|
|
|
#[test]
|
|
#[serial]
|
|
fn repeated_runtime_config_application_does_not_reconfigure_or_notify() {
|
|
let config = server_config_with_scanner(&[]);
|
|
super::apply_scanner_runtime_config(&config).expect("initial configuration should apply");
|
|
let generation = super::scanner_runtime_config_generation();
|
|
let resolved = super::current_scanner_runtime_config();
|
|
for _ in 0..1000 {
|
|
assert!(!super::apply_resolved_runtime_config(Some(resolved.clone())));
|
|
}
|
|
super::apply_scanner_runtime_config(&config).expect("unchanged configuration should be accepted");
|
|
assert_eq!(super::scanner_runtime_config_generation(), generation);
|
|
}
|
|
|
|
#[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]
|
|
fn scanner_unset_budget_uses_safe_default_but_explicit_zero_is_unbounded() {
|
|
let config = server_config_with_scanner(&[]);
|
|
with_var_unset(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, || {
|
|
let resolved = lookup_scanner_runtime_config(Some(&config)).expect("scanner runtime config");
|
|
assert_eq!(resolved.cycle_budget.max_duration, Some(Duration::from_secs(1800)));
|
|
assert_eq!(resolved.cycle_max_duration_source, ScannerRuntimeConfigSource::Default);
|
|
});
|
|
|
|
let config = server_config_with_scanner(&[(SCANNER_CYCLE_MAX_DURATION, "0")]);
|
|
with_var_unset(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, || {
|
|
let resolved = lookup_scanner_runtime_config(Some(&config)).expect("scanner runtime config");
|
|
assert_eq!(resolved.cycle_budget.max_duration, None);
|
|
assert_eq!(resolved.cycle_max_duration_source, ScannerRuntimeConfigSource::Config);
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn cycle_budget_invalid_or_overflow_config_is_rejected() {
|
|
with_var(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, Some("invalid"), || {
|
|
let error = lookup_scanner_runtime_config(None).expect_err("invalid duration env must be rejected");
|
|
assert!(error.to_string().contains(ENV_SCANNER_CYCLE_MAX_DURATION_SECS));
|
|
assert!(error.to_string().contains("<invalid>"));
|
|
assert!(!error.to_string().contains(": invalid ("));
|
|
});
|
|
with_var(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, Some("18446744073709551616"), || {
|
|
assert!(lookup_scanner_runtime_config(None).is_err());
|
|
});
|
|
with_var(ENV_SCANNER_CYCLE_MAX_DURATION_SECS, Some("18446744073709551615"), || {
|
|
assert!(lookup_scanner_runtime_config(None).is_err());
|
|
});
|
|
let config = server_config_with_scanner(&[(SCANNER_CYCLE_MAX_DURATION, "not-a-duration")]);
|
|
assert!(lookup_scanner_runtime_config(Some(&config)).is_err());
|
|
}
|
|
|
|
#[test]
|
|
#[serial]
|
|
fn recovery_backoff_cap_defaults_and_persisted_values() {
|
|
with_var_unset(ENV_SCANNER_RECOVERY_BACKOFF_CAP_SECS, || {
|
|
let default = lookup_scanner_runtime_config(None).expect("default scanner runtime config");
|
|
assert_eq!(default.recovery_backoff_cap, Duration::from_secs(600));
|
|
assert_eq!(default.recovery_backoff_cap_source, ScannerRuntimeConfigSource::Default);
|
|
assert_eq!(super::ScannerRuntimeConfig::default().recovery_backoff_cap, default.recovery_backoff_cap);
|
|
|
|
for (value, secs, source) in [
|
|
("1", 1, ScannerRuntimeConfigSource::Config),
|
|
(" 17 ", 17, ScannerRuntimeConfigSource::Config),
|
|
("1200", 1200, ScannerRuntimeConfigSource::Config),
|
|
("600", DEFAULT_SCANNER_RECOVERY_BACKOFF_CAP_SECS, ScannerRuntimeConfigSource::Default),
|
|
] {
|
|
let config = server_config_with_scanner(&[(SCANNER_RECOVERY_BACKOFF_CAP, value)]);
|
|
validate_scanner_runtime_config(&config).expect("valid persisted cap");
|
|
let resolved = lookup_scanner_runtime_config(Some(&config)).expect("persisted scanner runtime config");
|
|
assert_eq!(resolved.recovery_backoff_cap, Duration::from_secs(secs), "persisted {value}");
|
|
assert_eq!(resolved.recovery_backoff_cap_source, source, "persisted {value}");
|
|
}
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
#[serial]
|
|
fn recovery_backoff_cap_env_precedes_config_and_reports_source() {
|
|
let config = server_config_with_scanner(&[(SCANNER_RECOVERY_BACKOFF_CAP, "1200")]);
|
|
with_var(ENV_SCANNER_RECOVERY_BACKOFF_CAP_SECS, Some("17"), || {
|
|
super::apply_scanner_runtime_config(&config).expect("valid env override");
|
|
assert_eq!(super::scanner_recovery_backoff_cap(), Duration::from_secs(17));
|
|
let status = super::scanner_runtime_config_status();
|
|
assert_eq!(status.recovery_backoff_cap_seconds.value, 17);
|
|
assert_eq!(status.recovery_backoff_cap_seconds.source, ScannerRuntimeConfigSource::Env);
|
|
});
|
|
super::refresh_scanner_runtime_config_for_tests();
|
|
}
|
|
|
|
#[test]
|
|
#[serial]
|
|
fn recovery_backoff_cap_invalid_environment_is_rejected() {
|
|
let config = server_config_with_scanner(&[(SCANNER_RECOVERY_BACKOFF_CAP, "17")]);
|
|
for value in ["0", "-1", "1.5", "not-a-cap", "18446744073709551616", "18446744073709551615"] {
|
|
with_var(ENV_SCANNER_RECOVERY_BACKOFF_CAP_SECS, Some(value), || {
|
|
let err = lookup_scanner_runtime_config(Some(&config)).expect_err("invalid env must not fall back to config");
|
|
assert!(matches!(
|
|
err,
|
|
super::ScannerRuntimeConfigError::InvalidValue { key, .. } if key == ENV_SCANNER_RECOVERY_BACKOFF_CAP_SECS
|
|
));
|
|
});
|
|
}
|
|
with_var(ENV_SCANNER_RECOVERY_BACKOFF_CAP_SECS, Some("sensitive-invalid-value"), || {
|
|
let err = lookup_scanner_runtime_config(None).expect_err("invalid env cap");
|
|
assert!(err.to_string().contains("<invalid>"));
|
|
assert!(!err.to_string().contains("sensitive-invalid-value"));
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
#[serial]
|
|
fn recovery_backoff_cap_invalid_persisted_value_cannot_be_hidden_by_environment() {
|
|
with_var(ENV_SCANNER_RECOVERY_BACKOFF_CAP_SECS, Some("17"), || {
|
|
let before = super::current_scanner_runtime_config();
|
|
let generation = super::scanner_runtime_config_generation();
|
|
for value in ["0", "-1", "1.5", "not-a-cap", "18446744073709551616", "18446744073709551615"] {
|
|
let config = server_config_with_scanner(&[(SCANNER_RECOVERY_BACKOFF_CAP, value)]);
|
|
let err = validate_scanner_runtime_config(&config).expect_err("invalid persisted cap must be validated");
|
|
assert!(matches!(
|
|
err,
|
|
super::ScannerRuntimeConfigError::InvalidValue { key, .. } if key == SCANNER_RECOVERY_BACKOFF_CAP
|
|
));
|
|
assert!(lookup_scanner_runtime_config(Some(&config)).is_err(), "persisted {value}");
|
|
assert!(super::apply_scanner_runtime_config(&config).is_err(), "persisted {value}");
|
|
assert_eq!(
|
|
super::current_scanner_runtime_config(),
|
|
before,
|
|
"a rejected update must not change runtime state"
|
|
);
|
|
assert_eq!(super::scanner_runtime_config_generation(), generation);
|
|
}
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
fn scanner_runtime_config_validation_rejects_overflow_persisted_duration() {
|
|
let config = server_config_with_scanner(&[(SCANNER_CYCLE_MAX_DURATION, "18446744073709551615")]);
|
|
|
|
let error = validate_scanner_runtime_config(&config)
|
|
.expect_err("persisted duration that exceeds the timer range must be rejected");
|
|
assert!(error.to_string().contains(SCANNER_CYCLE_MAX_DURATION));
|
|
}
|
|
|
|
#[test]
|
|
#[serial]
|
|
fn scanner_runtime_config_normalizes_persisted_default_speed() {
|
|
let config = server_config_with_scanner(&[(SCANNER_SPEED, "default")]);
|
|
|
|
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::Default);
|
|
assert_eq!(resolved.speed_source, ScannerRuntimeConfigSource::Default);
|
|
assert_eq!(resolved.cycle_interval_source, ScannerRuntimeConfigSource::Default);
|
|
});
|
|
});
|
|
}
|
|
|
|
#[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]
|
|
#[serial]
|
|
fn scanner_runtime_config_prefers_heal_bitrot_cycle_over_scanner_compat_config() {
|
|
let config = server_config_with_scanner_and_heal(&[(SCANNER_BITROT_CYCLE, "3600")], &[(HEAL_BITROT_CYCLE, "off")]);
|
|
|
|
with_var_unset(ENV_SCANNER_BITROT_CYCLE_SECS, || {
|
|
let resolved = lookup_scanner_runtime_config(Some(&config)).expect("scanner runtime config");
|
|
|
|
assert_eq!(resolved.bitrot_cycle, None);
|
|
assert_eq!(resolved.bitrot_cycle_source, ScannerRuntimeConfigSource::Config);
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
#[serial]
|
|
fn scanner_runtime_config_marks_scanner_bitrot_cycle_as_compat_source() {
|
|
let config = server_config_with_scanner(&[(SCANNER_BITROT_CYCLE, "3600")]);
|
|
|
|
with_var_unset(ENV_SCANNER_BITROT_CYCLE_SECS, || {
|
|
let resolved = lookup_scanner_runtime_config(Some(&config)).expect("scanner runtime config");
|
|
super::apply_resolved_runtime_config(Some(resolved));
|
|
|
|
let encoded = serde_json::to_value(super::scanner_runtime_config_status()).expect("scanner status should serialize");
|
|
|
|
assert_eq!(encoded["bitrot_cycle_seconds"]["value"], 3600);
|
|
assert_eq!(encoded["bitrot_cycle_seconds"]["source"], "scanner_compat_config");
|
|
});
|
|
super::refresh_scanner_runtime_config_for_tests();
|
|
}
|
|
|
|
#[test]
|
|
#[serial]
|
|
fn scanner_runtime_config_normalizes_persisted_default_bitrot_cycles() {
|
|
let default_cycle = DEFAULT_HEAL_BITROT_CYCLE_SECS.to_string();
|
|
for config in [
|
|
server_config_with_scanner_and_heal(&[], &[(HEAL_BITROT_CYCLE, default_cycle.as_str())]),
|
|
server_config_with_scanner(&[(SCANNER_BITROT_CYCLE, default_cycle.as_str())]),
|
|
] {
|
|
with_var_unset(ENV_SCANNER_BITROT_CYCLE_SECS, || {
|
|
let resolved = lookup_scanner_runtime_config(Some(&config)).expect("scanner runtime config");
|
|
|
|
assert_eq!(resolved.bitrot_cycle, Some(Duration::from_secs(DEFAULT_HEAL_BITROT_CYCLE_SECS)));
|
|
assert_eq!(resolved.bitrot_cycle_source, ScannerRuntimeConfigSource::Default);
|
|
});
|
|
}
|
|
}
|
|
|
|
#[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_invalid_persisted_delay() {
|
|
let config = server_config_with_scanner(&[(SCANNER_DELAY, "-1")]);
|
|
|
|
let err = validate_scanner_runtime_config(&config).expect_err("persisted scanner delay should be validated");
|
|
assert!(err.to_string().contains("delay"));
|
|
}
|
|
|
|
#[test]
|
|
fn scanner_runtime_config_validation_rejects_excessive_persisted_delay() {
|
|
let config = server_config_with_scanner(&[(SCANNER_DELAY, "10000.1")]);
|
|
|
|
let err = validate_scanner_runtime_config(&config).expect_err("persisted scanner delay should be bounded");
|
|
assert!(err.to_string().contains("delay"));
|
|
}
|
|
|
|
#[test]
|
|
fn scanner_runtime_config_validation_accepts_delay_upper_bound() {
|
|
let config = server_config_with_scanner(&[(SCANNER_DELAY, "10000")]);
|
|
|
|
validate_scanner_runtime_config(&config).expect("delay upper bound should be accepted");
|
|
}
|
|
|
|
#[test]
|
|
#[serial]
|
|
fn scanner_runtime_config_uses_derived_delay_for_excessive_env_override() {
|
|
let config = server_config_with_scanner(&[(SCANNER_SPEED, "slow")]);
|
|
|
|
with_var(ENV_SCANNER_DELAY, Some("1e308"), || {
|
|
let resolved = lookup_scanner_runtime_config(Some(&config)).expect("scanner runtime config");
|
|
|
|
assert_eq!(resolved.delay, ScannerSpeed::Slow.sleep_factor());
|
|
assert_eq!(resolved.delay_source, ScannerRuntimeConfigSource::Env);
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
#[serial]
|
|
fn scanner_runtime_config_walkers_per_disk_defaults_to_one_walker() {
|
|
let config = server_config_with_scanner(&[]);
|
|
|
|
with_var_unset(ENV_SCANNER_WALKERS_PER_DISK, || {
|
|
let resolved = lookup_scanner_runtime_config(Some(&config)).expect("scanner runtime config");
|
|
|
|
assert_eq!(resolved.walkers_per_disk, 1);
|
|
assert_eq!(resolved.walkers_per_disk_source, ScannerRuntimeConfigSource::Default);
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
#[serial]
|
|
fn scanner_runtime_config_disk_io_budget_defaults_to_auto_and_parses_overrides() {
|
|
let config = server_config_with_scanner(&[
|
|
(SCANNER_DISK_IO_BUDGET_IOPS, "250"),
|
|
(SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC, "off"),
|
|
]);
|
|
|
|
with_var_unset(ENV_SCANNER_DISK_IO_BUDGET_IOPS, || {
|
|
with_var_unset(ENV_SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC, || {
|
|
let resolved =
|
|
lookup_scanner_runtime_config(Some(&server_config_with_scanner(&[]))).expect("default budget config");
|
|
assert_eq!(resolved.disk_io_budget_iops, ScannerIoBudgetLimit::Auto);
|
|
assert_eq!(resolved.disk_io_budget_iops_source, ScannerRuntimeConfigSource::Default);
|
|
assert_eq!(resolved.disk_io_budget_bytes_per_sec, ScannerIoBudgetLimit::Auto);
|
|
assert_eq!(resolved.disk_io_budget_bytes_per_sec_source, ScannerRuntimeConfigSource::Default);
|
|
|
|
let resolved = lookup_scanner_runtime_config(Some(&config)).expect("persisted budget config");
|
|
assert_eq!(resolved.disk_io_budget_iops, ScannerIoBudgetLimit::Explicit(250));
|
|
assert_eq!(resolved.disk_io_budget_iops_source, ScannerRuntimeConfigSource::Config);
|
|
assert_eq!(resolved.disk_io_budget_bytes_per_sec, ScannerIoBudgetLimit::Unlimited);
|
|
assert_eq!(resolved.disk_io_budget_bytes_per_sec_source, ScannerRuntimeConfigSource::Config);
|
|
|
|
let invalid = server_config_with_scanner(&[(SCANNER_DISK_IO_BUDGET_IOPS, "fast")]);
|
|
let err = validate_scanner_runtime_config(&invalid).expect_err("a non-numeric budget must be rejected");
|
|
assert!(err.to_string().contains(SCANNER_DISK_IO_BUDGET_IOPS), "{err}");
|
|
assert!(lookup_scanner_runtime_config(Some(&invalid)).is_err());
|
|
});
|
|
});
|
|
|
|
with_var(ENV_SCANNER_DISK_IO_BUDGET_IOPS, Some("0"), || {
|
|
with_var(ENV_SCANNER_DISK_IO_BUDGET_BYTES_PER_SEC, Some("auto"), || {
|
|
let resolved = lookup_scanner_runtime_config(Some(&config)).expect("env budget config");
|
|
assert_eq!(resolved.disk_io_budget_iops, ScannerIoBudgetLimit::Explicit(0));
|
|
assert_eq!(resolved.disk_io_budget_iops_source, ScannerRuntimeConfigSource::Env);
|
|
assert_eq!(resolved.disk_io_budget_bytes_per_sec, ScannerIoBudgetLimit::Auto);
|
|
assert_eq!(resolved.disk_io_budget_bytes_per_sec_source, ScannerRuntimeConfigSource::Env);
|
|
|
|
super::apply_resolved_runtime_config(Some(resolved));
|
|
let status = super::scanner_runtime_config_status();
|
|
assert_eq!(status.disk_io_budget_iops.value, "0");
|
|
assert_eq!(status.disk_io_budget_iops.source, ScannerRuntimeConfigSource::Env);
|
|
assert_eq!(status.disk_io_budget_bytes_per_sec.value, "auto");
|
|
assert_eq!(status.disk_io_budget_bytes_per_sec.source, ScannerRuntimeConfigSource::Env);
|
|
});
|
|
});
|
|
with_var(ENV_SCANNER_DISK_IO_BUDGET_IOPS, Some("-5"), || {
|
|
assert!(lookup_scanner_runtime_config(Some(&config)).is_err());
|
|
});
|
|
super::refresh_scanner_runtime_config_for_tests();
|
|
}
|
|
|
|
#[test]
|
|
#[serial]
|
|
fn scanner_runtime_config_walkers_per_disk_accepts_persisted_range_and_rejects_zero_or_excess() {
|
|
with_var_unset(ENV_SCANNER_WALKERS_PER_DISK, || {
|
|
for (value, expected) in [("1", 1), ("4", 4), ("16", 16)] {
|
|
let config = server_config_with_scanner(&[(SCANNER_WALKERS_PER_DISK, value)]);
|
|
validate_scanner_runtime_config(&config).expect("walkers per disk inside the range should validate");
|
|
let resolved = lookup_scanner_runtime_config(Some(&config)).expect("scanner runtime config");
|
|
assert_eq!(resolved.walkers_per_disk, expected, "persisted {value}");
|
|
let expected_source = if expected == 1 {
|
|
ScannerRuntimeConfigSource::Default
|
|
} else {
|
|
ScannerRuntimeConfigSource::Config
|
|
};
|
|
assert_eq!(resolved.walkers_per_disk_source, expected_source, "persisted {value}");
|
|
}
|
|
for value in ["0", "17", "-1", "many"] {
|
|
let config = server_config_with_scanner(&[(SCANNER_WALKERS_PER_DISK, value)]);
|
|
let err = validate_scanner_runtime_config(&config).expect_err("walkers per disk outside the range must fail");
|
|
assert!(err.to_string().contains("walkers_per_disk"), "persisted {value}: {err}");
|
|
assert!(
|
|
lookup_scanner_runtime_config(Some(&config)).is_err(),
|
|
"persisted {value} must not resolve into a runtime config"
|
|
);
|
|
}
|
|
});
|
|
}
|
|
|
|
#[test]
|
|
#[serial]
|
|
fn scanner_runtime_config_walkers_per_disk_env_overrides_and_is_range_checked() {
|
|
let config = server_config_with_scanner(&[(SCANNER_WALKERS_PER_DISK, "2")]);
|
|
|
|
with_var(ENV_SCANNER_WALKERS_PER_DISK, Some("3"), || {
|
|
let resolved = lookup_scanner_runtime_config(Some(&config)).expect("scanner runtime config");
|
|
assert_eq!(resolved.walkers_per_disk, 3);
|
|
assert_eq!(resolved.walkers_per_disk_source, ScannerRuntimeConfigSource::Env);
|
|
});
|
|
for value in ["0", "17", "x"] {
|
|
with_var(ENV_SCANNER_WALKERS_PER_DISK, Some(value), || {
|
|
let err = lookup_scanner_runtime_config(Some(&config)).expect_err("env walkers per disk is range checked");
|
|
assert!(err.to_string().contains(ENV_SCANNER_WALKERS_PER_DISK), "env {value}: {err}");
|
|
});
|
|
}
|
|
}
|
|
|
|
#[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(Some(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();
|
|
}
|
|
|
|
#[test]
|
|
#[serial]
|
|
fn applied_runtime_config_is_the_authoritative_scheduler_state() {
|
|
let config = server_config_with_scanner(&[(SCANNER_CYCLE, "321")]);
|
|
|
|
with_var_unset(ENV_SCANNER_CYCLE, || {
|
|
let generation = super::scanner_runtime_config_generation();
|
|
super::apply_scanner_runtime_config(&config).expect("scanner runtime config should apply");
|
|
|
|
let applied = super::current_scanner_runtime_config();
|
|
assert_eq!(applied.cycle_interval, Duration::from_secs(321));
|
|
assert_eq!(applied.cycle_interval_source, ScannerRuntimeConfigSource::Config);
|
|
assert!(super::scanner_runtime_config_generation() > generation);
|
|
});
|
|
super::refresh_scanner_runtime_config_for_tests();
|
|
}
|
|
|
|
#[test]
|
|
#[serial]
|
|
fn scanner_runtime_config_status_reports_persisted_pacing_overrides() {
|
|
let config = server_config_with_scanner(&[("delay", "3.5"), ("max_wait", "7")]);
|
|
|
|
with_var_unset("RUSTFS_SCANNER_DELAY", || {
|
|
with_var_unset("RUSTFS_SCANNER_MAX_WAIT_SECS", || {
|
|
let resolved = lookup_scanner_runtime_config(Some(&config)).expect("scanner runtime config");
|
|
super::apply_resolved_runtime_config(Some(resolved));
|
|
|
|
let encoded =
|
|
serde_json::to_value(super::scanner_runtime_config_status()).expect("scanner status should serialize");
|
|
|
|
assert_eq!(encoded["delay"]["value"], 3.5);
|
|
assert_eq!(encoded["delay"]["source"], "config");
|
|
assert_eq!(encoded["max_wait_seconds"]["value"], 7.0);
|
|
assert_eq!(encoded["max_wait_seconds"]["source"], "config");
|
|
});
|
|
});
|
|
super::refresh_scanner_runtime_config_for_tests();
|
|
}
|
|
|
|
#[test]
|
|
#[serial]
|
|
fn scanner_runtime_config_status_prefers_env_pacing_overrides() {
|
|
let config = server_config_with_scanner(&[("delay", "3.5"), ("max_wait", "7")]);
|
|
|
|
with_var("RUSTFS_SCANNER_DELAY", Some("1.25"), || {
|
|
with_var("RUSTFS_SCANNER_MAX_WAIT_SECS", Some("2"), || {
|
|
let resolved = lookup_scanner_runtime_config(Some(&config)).expect("scanner runtime config");
|
|
super::apply_resolved_runtime_config(Some(resolved));
|
|
|
|
let encoded =
|
|
serde_json::to_value(super::scanner_runtime_config_status()).expect("scanner status should serialize");
|
|
|
|
assert_eq!(encoded["delay"]["value"], 1.25);
|
|
assert_eq!(encoded["delay"]["source"], "env");
|
|
assert_eq!(encoded["max_wait_seconds"]["value"], 2.0);
|
|
assert_eq!(encoded["max_wait_seconds"]["source"], "env");
|
|
});
|
|
});
|
|
super::refresh_scanner_runtime_config_for_tests();
|
|
}
|
|
|
|
#[test]
|
|
#[serial]
|
|
fn scanner_runtime_config_status_preserves_subsecond_max_wait() {
|
|
let config = server_config_with_scanner(&[(SCANNER_SPEED, "fast")]);
|
|
|
|
with_var_unset(ENV_SCANNER_SPEED, || {
|
|
with_var_unset(ENV_SCANNER_MAX_WAIT_SECS, || {
|
|
let resolved = lookup_scanner_runtime_config(Some(&config)).expect("scanner runtime config");
|
|
super::apply_resolved_runtime_config(Some(resolved));
|
|
|
|
let status = super::scanner_runtime_config_status();
|
|
|
|
assert!((status.max_wait_seconds.value - 0.1).abs() < f64::EPSILON);
|
|
assert_eq!(status.max_wait_seconds.source, ScannerRuntimeConfigSource::Config);
|
|
});
|
|
});
|
|
super::refresh_scanner_runtime_config_for_tests();
|
|
}
|
|
}
|