mirror of
https://github.com/rustfs/rustfs.git
synced 2026-08-21 11:56:38 +00:00
chore(ecstore): collapse the expiry worker knobs to one documented env var (#6034)
init_background_expiry resolved its worker count through three env vars, none documented, none set in any known deployment: RUSTFS_MAX_EXPIRY_WORKERS, silently overridden by the underscore-prefixed _RUSTFS_ILM_EXPIRATION_WORKERS (a MinIO fossil, comment included), with a zero value then falling through to RUSTFS_DEFAULT_EXPIRY_WORKERS. RUSTFS_MAX_EXPIRY_WORKERS stays as the canonical name per the rc constraint (no new env var names): the count is now resolved once — a set, parseable, non-zero value wins, anything else falls back to min(cpus, 16). The constant moves to rustfs-config's runtime constants alongside ENV_TRANSITION_WORKERS, the two dead names are gone repo-wide (rg-verified), and a serial four-state unit test (unset/zero/valid/garbage) pins the resolution, modeled on the transition-worker env harness. Ref rustfs/backlog#1832 (PR2).
This commit is contained in:
@@ -81,6 +81,9 @@ pub const ENV_TEST_IAM_FAIL_INIT_ATTEMPTS: &str = "RUSTFS_TEST_IAM_FAIL_INIT_ATT
|
|||||||
pub const ENV_TEST_IAM_RETRY_INTERVAL_MS: &str = "RUSTFS_TEST_IAM_RETRY_INTERVAL_MS";
|
pub const ENV_TEST_IAM_RETRY_INTERVAL_MS: &str = "RUSTFS_TEST_IAM_RETRY_INTERVAL_MS";
|
||||||
/// Runtime env var controlling the transition worker count.
|
/// Runtime env var controlling the transition worker count.
|
||||||
pub const ENV_TRANSITION_WORKERS: &str = "RUSTFS_MAX_TRANSITION_WORKERS";
|
pub const ENV_TRANSITION_WORKERS: &str = "RUSTFS_MAX_TRANSITION_WORKERS";
|
||||||
|
/// Runtime env var controlling the ILM expiry worker count. A set, parsable,
|
||||||
|
/// non-zero value wins; anything else falls back to `min(cpus, 16)`.
|
||||||
|
pub const ENV_MAX_EXPIRY_WORKERS: &str = "RUSTFS_MAX_EXPIRY_WORKERS";
|
||||||
/// Runtime env var controlling the absolute maximum transition workers.
|
/// Runtime env var controlling the absolute maximum transition workers.
|
||||||
pub const ENV_TRANSITION_WORKERS_ABSOLUTE_MAX: &str = "RUSTFS_ABSOLUTE_MAX_WORKERS";
|
pub const ENV_TRANSITION_WORKERS_ABSOLUTE_MAX: &str = "RUSTFS_ABSOLUTE_MAX_WORKERS";
|
||||||
/// Runtime env var controlling the transition queue capacity.
|
/// Runtime env var controlling the transition queue capacity.
|
||||||
|
|||||||
@@ -79,8 +79,8 @@ use rustfs_common::metrics::{
|
|||||||
};
|
};
|
||||||
use rustfs_config::{
|
use rustfs_config::{
|
||||||
DEFAULT_TRANSITION_QUEUE_CAPACITY, DEFAULT_TRANSITION_QUEUE_SEND_TIMEOUT_MS, DEFAULT_TRANSITION_WORKERS_ABSOLUTE_MAX,
|
DEFAULT_TRANSITION_QUEUE_CAPACITY, DEFAULT_TRANSITION_QUEUE_SEND_TIMEOUT_MS, DEFAULT_TRANSITION_WORKERS_ABSOLUTE_MAX,
|
||||||
DEFAULT_TRANSITION_WORKERS_CAP, ENV_TRANSITION_QUEUE_CAPACITY, ENV_TRANSITION_QUEUE_SEND_TIMEOUT_MS, ENV_TRANSITION_WORKERS,
|
DEFAULT_TRANSITION_WORKERS_CAP, ENV_MAX_EXPIRY_WORKERS, ENV_TRANSITION_QUEUE_CAPACITY, ENV_TRANSITION_QUEUE_SEND_TIMEOUT_MS,
|
||||||
ENV_TRANSITION_WORKERS_ABSOLUTE_MAX,
|
ENV_TRANSITION_WORKERS, ENV_TRANSITION_WORKERS_ABSOLUTE_MAX,
|
||||||
};
|
};
|
||||||
use rustfs_data_usage::TierStats;
|
use rustfs_data_usage::TierStats;
|
||||||
use rustfs_filemeta::{
|
use rustfs_filemeta::{
|
||||||
@@ -2017,18 +2017,25 @@ fn is_slow_down(err: &Error) -> bool {
|
|||||||
matches!(err, Error::SlowDown)
|
matches!(err, Error::SlowDown)
|
||||||
}
|
}
|
||||||
|
|
||||||
pub async fn init_background_expiry(api: Arc<ECStore>) {
|
/// Resolves the expiry worker count from the single documented knob,
|
||||||
let mut workers = get_env_usize("RUSTFS_MAX_EXPIRY_WORKERS", std::cmp::min(num_cpus::get(), 16));
|
/// `RUSTFS_MAX_EXPIRY_WORKERS`: a set, parsable, non-zero value wins;
|
||||||
//globalILMConfig.getExpirationWorkers()
|
/// anything else falls back to `min(cpus, 16)`. The historical
|
||||||
if let Ok(env_expiration_workers) = env::var("_RUSTFS_ILM_EXPIRATION_WORKERS")
|
/// `_RUSTFS_ILM_EXPIRATION_WORKERS` silent override and the
|
||||||
&& let Ok(num_expirations) = env_expiration_workers.parse::<usize>()
|
/// `RUSTFS_DEFAULT_EXPIRY_WORKERS` zero-fallback were undocumented, unset in
|
||||||
{
|
/// every known deployment, and are removed (backlog#1832).
|
||||||
workers = num_expirations;
|
fn expiry_worker_count() -> usize {
|
||||||
|
let default = std::cmp::min(num_cpus::get(), 16);
|
||||||
|
match env::var(ENV_MAX_EXPIRY_WORKERS) {
|
||||||
|
Ok(value) => match value.parse::<usize>() {
|
||||||
|
Ok(workers) if workers > 0 => workers,
|
||||||
|
_ => default,
|
||||||
|
},
|
||||||
|
Err(_) => default,
|
||||||
}
|
}
|
||||||
|
}
|
||||||
|
|
||||||
if workers == 0 {
|
pub async fn init_background_expiry(api: Arc<ECStore>) {
|
||||||
workers = get_env_usize("RUSTFS_DEFAULT_EXPIRY_WORKERS", 8);
|
let workers = expiry_worker_count();
|
||||||
}
|
|
||||||
|
|
||||||
ExpiryState::resize_workers(workers, api.clone()).await;
|
ExpiryState::resize_workers(workers, api.clone()).await;
|
||||||
let _ = spawn_tier_free_version_recovery_once(api.clone(), &TIER_FREE_VERSION_RECOVERY_STARTED);
|
let _ = spawn_tier_free_version_recovery_once(api.clone(), &TIER_FREE_VERSION_RECOVERY_STARTED);
|
||||||
@@ -5086,6 +5093,7 @@ pub async fn apply_lifecycle_action(event: &lifecycle::Event, src: &LcEventSrc,
|
|||||||
|
|
||||||
#[cfg(test)]
|
#[cfg(test)]
|
||||||
mod tests {
|
mod tests {
|
||||||
|
use super::expiry_worker_count;
|
||||||
use super::{
|
use super::{
|
||||||
DATE_EXPIRY_EXISTING_OBJECTS_GRACE_SECS, DEFAULT_TRANSITION_QUEUE_CAPACITY, DEFAULT_TRANSITION_WORKERS_ABSOLUTE_MAX,
|
DATE_EXPIRY_EXISTING_OBJECTS_GRACE_SECS, DEFAULT_TRANSITION_QUEUE_CAPACITY, DEFAULT_TRANSITION_WORKERS_ABSOLUTE_MAX,
|
||||||
DEFAULT_TRANSITION_WORKERS_CAP, EVENT_LIFECYCLE_EVALUATION_FAILED, EVENT_LIFECYCLE_EXPIRED_DETECTED,
|
DEFAULT_TRANSITION_WORKERS_CAP, EVENT_LIFECYCLE_EVALUATION_FAILED, EVENT_LIFECYCLE_EXPIRED_DETECTED,
|
||||||
@@ -5169,6 +5177,7 @@ mod tests {
|
|||||||
#[cfg(feature = "test-util")]
|
#[cfg(feature = "test-util")]
|
||||||
use http::HeaderMap;
|
use http::HeaderMap;
|
||||||
use rustfs_common::metrics::{IlmAction, global_metrics};
|
use rustfs_common::metrics::{IlmAction, global_metrics};
|
||||||
|
use rustfs_config::ENV_MAX_EXPIRY_WORKERS;
|
||||||
use rustfs_config::ENV_TRANSITION_WORKERS_ABSOLUTE_MAX;
|
use rustfs_config::ENV_TRANSITION_WORKERS_ABSOLUTE_MAX;
|
||||||
use rustfs_data_usage::TierStats;
|
use rustfs_data_usage::TierStats;
|
||||||
use rustfs_filemeta::{FileInfo, FileMeta};
|
use rustfs_filemeta::{FileInfo, FileMeta};
|
||||||
@@ -7167,6 +7176,63 @@ mod tests {
|
|||||||
}
|
}
|
||||||
}
|
}
|
||||||
|
|
||||||
|
// SAFETY: same contract as with_transition_worker_env — only used from
|
||||||
|
// `#[serial]` tests, so no concurrent reader/writer can access the process
|
||||||
|
// environment while `env::set_var`/`env::remove_var` is active.
|
||||||
|
#[allow(unsafe_code)]
|
||||||
|
fn with_expiry_worker_env<F>(value: Option<&str>, test_fn: F)
|
||||||
|
where
|
||||||
|
F: FnOnce(),
|
||||||
|
{
|
||||||
|
let original = env::var_os(ENV_MAX_EXPIRY_WORKERS);
|
||||||
|
|
||||||
|
match value {
|
||||||
|
Some(v) => unsafe {
|
||||||
|
env::set_var(ENV_MAX_EXPIRY_WORKERS, v);
|
||||||
|
},
|
||||||
|
None => unsafe {
|
||||||
|
env::remove_var(ENV_MAX_EXPIRY_WORKERS);
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
let result = std::panic::catch_unwind(std::panic::AssertUnwindSafe(test_fn));
|
||||||
|
|
||||||
|
match original {
|
||||||
|
Some(v) => unsafe {
|
||||||
|
env::set_var(ENV_MAX_EXPIRY_WORKERS, v);
|
||||||
|
},
|
||||||
|
None => unsafe {
|
||||||
|
env::remove_var(ENV_MAX_EXPIRY_WORKERS);
|
||||||
|
},
|
||||||
|
}
|
||||||
|
|
||||||
|
if let Err(e) = result {
|
||||||
|
std::panic::resume_unwind(e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// backlog#1832: the single expiry knob must resolve all four env states
|
||||||
|
/// (unset / zero / valid / garbage); the removed `_RUSTFS_ILM_EXPIRATION_WORKERS`
|
||||||
|
/// override and `RUSTFS_DEFAULT_EXPIRY_WORKERS` fallback must stay gone.
|
||||||
|
#[test]
|
||||||
|
#[serial]
|
||||||
|
fn expiry_worker_count_resolves_all_env_states() {
|
||||||
|
let default = std::cmp::min(num_cpus::get(), 16);
|
||||||
|
|
||||||
|
with_expiry_worker_env(None, || {
|
||||||
|
assert_eq!(expiry_worker_count(), default, "unset env must fall back to min(cpus, 16)");
|
||||||
|
});
|
||||||
|
with_expiry_worker_env(Some("0"), || {
|
||||||
|
assert_eq!(expiry_worker_count(), default, "zero must fall back instead of spawning zero workers");
|
||||||
|
});
|
||||||
|
with_expiry_worker_env(Some("4"), || {
|
||||||
|
assert_eq!(expiry_worker_count(), 4, "a valid positive value must win");
|
||||||
|
});
|
||||||
|
with_expiry_worker_env(Some("not-a-number"), || {
|
||||||
|
assert_eq!(expiry_worker_count(), default, "garbage must fall back to the default");
|
||||||
|
});
|
||||||
|
}
|
||||||
|
|
||||||
// SAFETY: this helper is only used from `#[serial]` tests and those tests run under a
|
// SAFETY: this helper is only used from `#[serial]` tests and those tests run under a
|
||||||
// single-thread runtime (`worker_threads = 1`), so no concurrent reader/writer can access
|
// single-thread runtime (`worker_threads = 1`), so no concurrent reader/writer can access
|
||||||
// process environment while `env::set_var`/`env::remove_var` is active.
|
// process environment while `env::set_var`/`env::remove_var` is active.
|
||||||
|
|||||||
Reference in New Issue
Block a user