From 599df58301a51417f653ac326b123323159a55b1 Mon Sep 17 00:00:00 2001 From: Thanos Makatos Date: Fri, 23 Oct 2026 06:57:57 +0000 Subject: [PATCH 1/2] Express configured durations in seconds, not polls Durations are now integer seconds in settings ending in _secs: scale_poll_secs, cooldown_secs (now integers, were floats) scale_down_sustain_polls -> scale_down_sustain_secs scale_validation_sample_polls -> scale_validation_delay_secs scale_validation_sample_polls_per_thread -> scale_validation_delay_per_thread_secs post_revert_up_grace_polls -> post_revert_up_grace_secs post_revert_up_recovery_polls -> post_revert_up_recovery_secs post_revert_up_backoff_max_polls -> post_revert_up_backoff_max_secs scale_down_backoff_max_polls -> scale_down_backoff_max_secs Every duration must be a multiple of scale_poll_secs. The engine receives the poll period when it is built and converts its durations to polls once, so engine logic is unchanged. The daemon refuses to start if any duration, controller or engine, is not a multiple, listing all of them with the nearest valid values. The round_up_durations config option and --round-up-durations flag (both off by default) instead round such durations up with a warning, so that a new poll period can be tried without rewriting the config. The daemon now logs the effective controller, engine, and backend configuration at startup, which shows rounded values. The daemon also prints errors with Display rather than Debug. AI-generated: everything. Signed-off-by: Thanos Makatos --- README.md | 30 +- io-thread-controller.d/engines/threshold.json | 14 +- io-thread-controller.json | 5 +- src/config.rs | 253 +++++++++++-- src/controller.rs | 47 +-- src/daemon.rs | 39 +- src/engines/mod.rs | 28 +- src/engines/threshold.rs | 356 ++++++++++++------ src/main.rs | 31 +- 9 files changed, 591 insertions(+), 212 deletions(-) diff --git a/README.md b/README.md index e4ece764..e35b01c8 100644 --- a/README.md +++ b/README.md @@ -90,6 +90,20 @@ Controller, engine, and backend JSON files may contain an `include` field. Its path is resolved relative to the including file, loaded recursively, and merged before the including file's keys are applied. +## Durations + +Durations are integer seconds in settings ending in `_secs`. Every duration, +including engine ones, must be a whole multiple of `scale_poll_secs`, and zero +is allowed. Otherwise the daemon refuses to start and lists each offending +setting with the nearest valid values. Changing only `scale_poll_secs` can +therefore invalidate other settings. + +To try a new poll period without rewriting the other settings, set +`round_up_durations` or pass `--round-up-durations`. Each offending duration is +then rounded up to the next multiple with a warning. At startup the daemon logs +the effective controller, engine, and backend configuration, which shows the +rounded values. + ## Thread-name filtering Backends may select fallback `/proc` CPU samples with regular expressions and @@ -131,21 +145,21 @@ queries and thread-count changes need an independently stateful handle for each VM. This keeps the controller independent of libvirt, QMP, control sockets, and other transport details. -After a successful action, the engine waits -`scale_validation_sample_polls` complete samples. The optional -`scale_validation_sample_polls_per_thread` setting adds that many samples for -each thread in the resulting pool, allowing larger pools more time to settle. +After a successful action, the engine samples for +`scale_validation_delay_secs`. The optional +`scale_validation_delay_per_thread_secs` setting adds that much time for each +thread in the resulting pool, allowing larger pools more time to settle. A scale-up is reverted unless IOPS gained at least `scale_up_min_gain_percent`; a scale-down is reverted when IOPS lost more than `scale_down_revert_drop_percent`. Setting either percentage to zero disables that direction's validation. Repeated failures back off per VM. After a reverted scale-up, scale-up is -blocked for `post_revert_up_grace_polls` plus `post_revert_up_recovery_polls`; +blocked for `post_revert_up_grace_secs` plus `post_revert_up_recovery_secs`; that hold doubles with each consecutive failed scale-up, up to -`post_revert_up_backoff_max_polls`. After a reverted scale-down, the next -scale-down needs `scale_down_sustain_polls` low-utilisation polls doubled once -per consecutive failure, up to `scale_down_backoff_max_polls`. A kept action +`post_revert_up_backoff_max_secs`. After a reverted scale-down, the next +scale-down needs `scale_down_sustain_secs` of low utilisation doubled once per +consecutive failure, up to `scale_down_backoff_max_secs`. A kept action resets its direction's backoff, as does an applied action in the opposite direction. Setting a maximum to zero disables that direction's backoff. diff --git a/io-thread-controller.d/engines/threshold.json b/io-thread-controller.d/engines/threshold.json index 7bbbde42..110084e9 100644 --- a/io-thread-controller.d/engines/threshold.json +++ b/io-thread-controller.d/engines/threshold.json @@ -1,14 +1,14 @@ { "scale_up_threshold_percent": 95, - "scale_down_sustain_polls": 12, + "scale_down_sustain_secs": 120, "max_scale_down_step": 0, "scale_up_min_gain_percent": 5, "scale_down_revert_drop_percent": 3, - "scale_validation_sample_polls": 3, - "scale_validation_sample_polls_per_thread": 0, + "scale_validation_delay_secs": 30, + "scale_validation_delay_per_thread_secs": 0, "block_scale_up_when_cgroup_throttled": true, - "post_revert_up_grace_polls": 1, - "post_revert_up_recovery_polls": 2, - "post_revert_up_backoff_max_polls": 30, - "scale_down_backoff_max_polls": 120 + "post_revert_up_grace_secs": 10, + "post_revert_up_recovery_secs": 20, + "post_revert_up_backoff_max_secs": 300, + "scale_down_backoff_max_secs": 1200 } diff --git a/io-thread-controller.json b/io-thread-controller.json index 9bab418e..9efee460 100644 --- a/io-thread-controller.json +++ b/io-thread-controller.json @@ -1,11 +1,12 @@ { - "scale_poll_secs": 10.0, + "scale_poll_secs": 10, + "round_up_durations": false, "enable_per_vm_status_line": true, "enable_aggregate_status_line": true, "min_thread_count": 1, "max_thread_count": 8, "host_cpu_scale_up_ceiling_percent": 90, - "cooldown_secs": 10.0, + "cooldown_secs": 10, "engine": "threshold", "engine_config_dir": "/etc/io-thread-controller.d/engines", "backend_config_dir": "/etc/io-thread-controller.d/backends", diff --git a/src/config.rs b/src/config.rs index 8c7b106e..9e4b4156 100644 --- a/src/config.rs +++ b/src/config.rs @@ -5,7 +5,7 @@ //! Configuration for the io-thread-controller daemon. -use std::io; +use std::{io, time::Duration}; use serde::{Deserialize, Deserializer, Serialize, Serializer}; use thiserror::Error; @@ -17,6 +17,13 @@ pub enum ConfigError { #[error("invalid configuration value: {0}")] InvalidValue(String), + #[error( + "durations must be multiples of scale_poll_secs (set round_up_durations or pass \ + --round-up-durations to round them up):\n {}", + .0.join("\n ") + )] + NotPollMultiple(Vec), + #[error(transparent)] Io(#[from] io::Error), @@ -59,11 +66,15 @@ pub struct Config { pub host_cpu_scale_up_ceiling: f64, /// Per-VM delay after ordinary successful scale actions. #[serde(default = "default_cooldown_secs")] - pub cooldown_secs: f64, - /// Poll cadence in seconds. The controller refreshes - /// per-instance state on each tick and calls the active - /// scaling engine for a decision. - pub scale_poll_secs: f64, + pub cooldown_secs: u64, + /// Poll cadence in seconds. The controller refreshes per-instance state on + /// each tick and calls the active scaling engine for a decision. Every + /// other duration, including engine ones, must be a multiple of it. + pub scale_poll_secs: u64, + /// Round durations that are not multiples of `scale_poll_secs` up to the + /// next multiple, with a warning, instead of rejecting them. + #[serde(default)] + pub round_up_durations: bool, /// Emit a `vm=` status line per tracked instance on /// every tick (default `true`). #[serde(default = "default_true")] @@ -111,8 +122,92 @@ fn default_host_cpu_scale_up_ceiling() -> f64 { 0.9 } -fn default_cooldown_secs() -> f64 { - 30.0 +fn default_cooldown_secs() -> u64 { + 30 +} + +/// Controller poll period that durations are expressed in multiples of. +#[derive(Debug, Clone, Copy)] +pub struct PollPeriod { + secs: u64, + round_up: bool, +} + +impl PollPeriod { + /// `secs` must be non-zero. + pub fn new(secs: u64, round_up: bool) -> Self { + assert!(secs != 0, "poll period must be non-zero"); + Self { secs, round_up } + } + + /// Start converting a group of durations to poll counts. + pub fn converter(self) -> PollConverter { + PollConverter { + poll: self, + errors: Vec::new(), + } + } +} + +/// Converts durations to poll counts, collecting every invalid duration so +/// that they can be reported together. +pub struct PollConverter { + poll: PollPeriod, + errors: Vec, +} + +impl PollConverter { + /// Return the poll period. + pub fn period(&self) -> Duration { + Duration::from_secs(self.poll.secs) + } + + /// Return how many polls `secs` spans. + /// + /// A `secs` that is not a multiple of the poll period is either rounded + /// up in place, so the effective configuration shows the value in use, or + /// recorded as an error for [`Self::finish`]. + pub fn polls(&mut self, name: &str, secs: &mut u64) -> u32 { + let period = self.poll.secs; + let mut polls = *secs / period; + if !secs.is_multiple_of(period) { + let below = polls * period; + let above = below.saturating_add(period); + if !self.poll.round_up { + self.errors.push(format!( + "{name} ({secs}) is not a multiple of scale_poll_secs ({period}); \ + nearest valid values: {below} or {above}" + )); + return 0; + } + tracing::warn!( + target: "controller", + setting = name, + configured = *secs, + effective = above, + "rounded duration up to a multiple of scale_poll_secs" + ); + *secs = above; + polls += 1; + } + match u32::try_from(polls) { + Ok(polls) => polls, + Err(_) => { + self.errors + .push(format!("{name} ({secs}) spans too many polls")); + 0 + } + } + } + + /// Fail with every duration rejected by [`Self::polls`]. + pub fn finish(self) -> Result<(), ConfigError> { + if self.errors.is_empty() { + Ok(()) + } else { + Err(ConfigError::NotPollMultiple(self.errors)) + } + } } /// Deserialise a human-readable percent to a floating point value. @@ -140,11 +235,12 @@ impl Default for Config { engine: "threshold".to_string(), engine_config_dir: Path::new("/etc/io-thread-controller.d/engines"), backend_config_dir: Path::new("/etc/io-thread-controller.d/backends"), - scale_poll_secs: 10.0, + scale_poll_secs: 10, + round_up_durations: false, min_thread_count: 2, max_thread_count: 8, host_cpu_scale_up_ceiling: 0.9, - cooldown_secs: 30.0, + cooldown_secs: default_cooldown_secs(), vm_state_path: Path::new("/run/io-thread-controller/vm-ownership.json"), enable_per_vm_status_line: true, enable_per_vm_main_log: false, @@ -200,11 +296,26 @@ fn load_config_value(path: &Path) -> Result { Ok(merged) } +impl Config { + /// Return the poll period that every other duration is a multiple of. + /// + /// Panics if `scale_poll_secs` is zero, which [`validate_config`] + /// rejects. + pub fn poll_period(&self) -> PollPeriod { + PollPeriod::new(self.scale_poll_secs, self.round_up_durations) + } +} + // TODO add a RawConfig or ValidatedConfig to ensure an unvalidated config // cannot be constructed outside this module. -/// Ensures configuration is internally coherent. -pub fn validate_config(cfg: &Config) -> Result<(), ConfigError> { - if !cfg.scale_poll_secs.is_finite() || cfg.scale_poll_secs <= 0.0 { +/// Ensures configuration is internally coherent, rounding durations up to +/// poll multiples when `round_up_durations` is set. +/// +/// Invalid durations are not reported yet: the returned converter holds them +/// so that the engine can convert its own durations with it, and +/// [`PollConverter::finish`] then reports every invalid duration at once. +pub fn validate_config(cfg: &mut Config) -> Result { + if cfg.scale_poll_secs == 0 { return Err(ConfigError::InvalidValue( "scale_poll_secs must be > 0".to_string(), )); @@ -226,12 +337,9 @@ pub fn validate_config(cfg: &Config) -> Result<(), ConfigError> { "host_cpu_scale_up_ceiling_percent must be in [0, 100]".to_string(), )); } - if !(cfg.cooldown_secs.is_finite() && cfg.cooldown_secs >= 0.0) { - return Err(ConfigError::InvalidValue( - "cooldown_secs must be finite and non-negative".to_string(), - )); - } - Ok(()) + let mut converter = cfg.poll_period().converter(); + converter.polls("cooldown_secs", &mut cfg.cooldown_secs); + Ok(converter) } /// Serialise the default config as indented JSON for CLI @@ -248,10 +356,10 @@ mod tests { /// engine with a 10s poll. #[test] fn default_config_is_valid() { - let cfg = Config::default(); + let mut cfg = Config::default(); assert_eq!(cfg.engine, "threshold"); - assert!((cfg.scale_poll_secs - 10.0).abs() < f64::EPSILON); - validate_config(&cfg).unwrap(); + assert_eq!(cfg.scale_poll_secs, 10); + validate_config(&mut cfg).unwrap().finish().unwrap(); } /// Test that `validate_config` rejects an empty engine name. @@ -259,24 +367,96 @@ mod tests { fn validate_rejects_empty_engine() { let mut cfg = Config::default(); cfg.engine.clear(); - let err = validate_config(&cfg).unwrap_err().to_string(); + let err = validate_config(&mut cfg).err().unwrap().to_string(); assert!(err.contains("engine must be non-empty")); } - /// Test that `validate_config` rejects non-positive - /// `scale_poll_secs`. + /// Test that `validate_config` rejects a zero `scale_poll_secs`. #[test] - fn validate_rejects_non_positive_poll() { + fn validate_rejects_zero_poll() { let mut cfg = Config { - scale_poll_secs: 0.0, + scale_poll_secs: 0, ..Default::default() }; - let err = validate_config(&cfg).unwrap_err().to_string(); + let err = validate_config(&mut cfg).err().unwrap().to_string(); assert!(err.contains("scale_poll_secs must be > 0")); + } - cfg.scale_poll_secs = f64::NAN; - let err = validate_config(&cfg).unwrap_err().to_string(); - assert!(err.contains("scale_poll_secs must be > 0")); + /// Tests that a cooldown that is not a multiple of the poll period is + /// rejected with the nearest valid values, and left unchanged. + #[test] + fn test_validate_rejects_cooldown_not_poll_multiple() { + let mut cfg = Config { + cooldown_secs: 25, + ..Default::default() + }; + let err = validate_config(&mut cfg) + .unwrap() + .finish() + .unwrap_err() + .to_string(); + assert!( + err.contains( + "cooldown_secs (25) is not a multiple of scale_poll_secs (10); nearest valid \ + values: 20 or 30" + ), + "{err}" + ); + assert!(err.contains("--round-up-durations"), "{err}"); + assert_eq!(cfg.cooldown_secs, 25); + } + + /// Tests that `round_up_durations` rounds a non-multiple cooldown up to + /// the next poll multiple instead of rejecting it. + #[test] + fn test_validate_rounds_up_cooldown() { + let mut cfg = Config { + cooldown_secs: 25, + round_up_durations: true, + ..Default::default() + }; + validate_config(&mut cfg).unwrap().finish().unwrap(); + assert_eq!(cfg.cooldown_secs, 30); + } + + /// Tests that the poll converter counts polls, accepts zero, reports + /// every invalid duration at once, and rounds up when enabled. + #[test] + fn test_poll_converter() { + let mut converter = PollPeriod::new(4, false).converter(); + assert_eq!(converter.polls("zero", &mut 0), 0); + assert_eq!(converter.polls("exact", &mut 12), 3); + converter.polls("short", &mut 1); + converter.polls("long", &mut 30); + let err = converter.finish().unwrap_err().to_string(); + assert!( + err.contains( + "short (1) is not a multiple of scale_poll_secs (4); nearest valid values: 0 or 4" + ), + "{err}" + ); + assert!( + err.contains( + "long (30) is not a multiple of scale_poll_secs (4); nearest valid values: 28 or 32" + ), + "{err}" + ); + + let mut converter = PollPeriod::new(4, true).converter(); + let mut secs = 30; + assert_eq!(converter.polls("long", &mut secs), 8); + assert_eq!(secs, 32); + converter.finish().unwrap(); + } + + /// Tests that durations must be integer seconds. + #[test] + fn test_durations_must_be_integers() { + let err = serde_json::from_str::( + r#"{"engine": "foo", "engine_config_dir": "/engines", "backend_config_dir": "/backends", "scale_poll_secs": 7.5, "vm_state_path": "/path/to/vm-state"}"#, + ) + .unwrap_err(); + assert!(err.to_string().contains("expected u64"), "{err}"); } /// Test that minimal valid JSON loads into `Config` with the @@ -291,7 +471,7 @@ mod tests { "engine": "threshold", "engine_config_dir": "/etc/io-thread-controller/engines", "backend_config_dir": "/etc/io-thread-controller/backends", - "scale_poll_secs": 7.5, + "scale_poll_secs": 7, "vm_state_path": "/var/lib/io-thread-controller/vm-state.json" }"#, ) @@ -299,7 +479,7 @@ mod tests { let cfg: Config = load_json_with_includes(Path::new(path.to_str().unwrap())).unwrap(); assert_eq!(cfg.engine, "threshold"); - assert!((cfg.scale_poll_secs - 7.5).abs() < f64::EPSILON); + assert_eq!(cfg.scale_poll_secs, 7); assert_eq!( cfg.engine_config_dir.as_os_str(), std::ffi::OsStr::new("/etc/io-thread-controller/engines") @@ -318,7 +498,7 @@ mod tests { "engine": "threshold", "engine_config_dir": "/engines", "backend_config_dir": "/backends", - "scale_poll_secs": 1.0, + "scale_poll_secs": 1, "not_a_real_field": true }"#, ) @@ -334,7 +514,7 @@ mod tests { fn dump_default_config_contains_engine() { let dumped = dump_default_config(); assert!(dumped.contains(r#""engine": "threshold""#)); - assert!(dumped.contains(r#""scale_poll_secs": 10.0"#)); + assert!(dumped.contains(r#""scale_poll_secs": 10"#)); assert!(dumped.contains(r#""/etc/io-thread-controller.d/engines""#)); } @@ -346,7 +526,8 @@ mod tests { assert_eq!(cfg.min_thread_count, 1); assert_eq!(cfg.max_thread_count, 8); assert!((cfg.host_cpu_scale_up_ceiling - 0.9).abs() < f64::EPSILON); - assert_eq!(cfg.cooldown_secs, 30.0); + assert_eq!(cfg.cooldown_secs, 30); + assert!(!cfg.round_up_durations); let serialized = serde_json::to_string(&cfg).unwrap(); assert!(serialized.contains(r#""host_cpu_scale_up_ceiling_percent":90.0"#)); diff --git a/src/controller.rs b/src/controller.rs index 1485f28c..65bb2714 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -550,7 +550,6 @@ impl Controller { host_cpu_util: self.host_cpu_util, psi: self.psi, tick_index: self.tick_index, - scale_poll_secs: self.cfg.scale_poll_secs, }; let fleet: Vec<_> = self.instances.values().cloned().collect(); let refresh_cgroup = self.cfg.refresh_cgroup_on_each_read; @@ -808,7 +807,7 @@ impl Controller { ScaleAction::Up(_) | ScaleAction::Down(_) | ScaleAction::Revert(_) ) { instance.status.write().await.cooldown_until = - Some(Instant::now() + Duration::from_secs_f64(self.cfg.cooldown_secs)); + Some(Instant::now() + Duration::from_secs(self.cfg.cooldown_secs)); } let applied_at = Instant::now(); self.engine @@ -835,7 +834,7 @@ impl Controller { ScaleAction::Up(_) | ScaleAction::Down(_) | ScaleAction::Revert(_) ) { status.cooldown_until = - Some(applied_at + Duration::from_secs_f64(self.cfg.cooldown_secs)); + Some(applied_at + Duration::from_secs(self.cfg.cooldown_secs)); } drop(status); self.engine @@ -1088,6 +1087,14 @@ mod tests { #[cfg(feature = "threshold-engine")] use crate::engines::threshold::{ThresholdConfig, ThresholdEngine}; + #[cfg(feature = "threshold-engine")] + fn default_threshold_engine() -> ThresholdEngine { + let mut durations = Config::default().poll_period().converter(); + let engine = ThresholdEngine::new(ThresholdConfig::default(), &mut durations).unwrap(); + durations.finish().unwrap(); + engine + } + struct VcpuLimitedClient { target: Arc, } @@ -1118,7 +1125,7 @@ mod tests { let state_dir = tempfile::tempdir().unwrap(); let cfg = Config { vm_state_path: Path::new(&state_dir.path().join("ownership.json")), - cooldown_secs: 0.0, + cooldown_secs: 0, ..Default::default() }; let target = Arc::new(AtomicU32::new(2)); @@ -1137,11 +1144,7 @@ mod tests { status.vcpu_count = 4; status.ownership_classification = Some(true); } - let mut controller = Controller::new( - cfg, - Box::new(ThresholdEngine::new(ThresholdConfig::default())), - ) - .unwrap(); + let mut controller = Controller::new(cfg, Box::new(default_threshold_engine())).unwrap(); controller .instances .insert(instance.id.clone(), Arc::clone(&instance)); @@ -1200,7 +1203,7 @@ mod tests { min_thread_count: 2, max_thread_count: 3, host_cpu_scale_up_ceiling: 0.5, - cooldown_secs: 30.0, + cooldown_secs: 30, ..Default::default() }; let target = Arc::new(AtomicU32::new(2)); @@ -1218,11 +1221,7 @@ mod tests { status.thread_count = 2; status.vcpu_count = 4; } - let mut controller = Controller::new( - cfg, - Box::new(ThresholdEngine::new(ThresholdConfig::default())), - ) - .unwrap(); + let mut controller = Controller::new(cfg, Box::new(default_threshold_engine())).unwrap(); controller.host_cpu_util = 0.5; controller .instances @@ -1272,7 +1271,7 @@ mod tests { let state_dir = tempfile::tempdir().unwrap(); let cfg = Config { vm_state_path: Path::new(&state_dir.path().join("ownership.json")), - cooldown_secs: 30.0, + cooldown_secs: 30, min_thread_count: 1, max_thread_count: 8, ..Default::default() @@ -1293,11 +1292,7 @@ mod tests { status.vcpu_count = 8; status.ownership_classification = Some(true); } - let mut controller = Controller::new( - cfg, - Box::new(ThresholdEngine::new(ThresholdConfig::default())), - ) - .unwrap(); + let mut controller = Controller::new(cfg, Box::new(default_threshold_engine())).unwrap(); controller .instances .insert(instance.id.clone(), Arc::clone(&instance)); @@ -1334,7 +1329,7 @@ mod tests { let state_dir = tempfile::tempdir().unwrap(); let cfg = Config { vm_state_path: Path::new(&state_dir.path().join("ownership.json")), - cooldown_secs: 60.0, + cooldown_secs: 60, min_thread_count: 1, max_thread_count: 8, ..Default::default() @@ -1355,11 +1350,7 @@ mod tests { status.vcpu_count = 8; status.ownership_classification = Some(true); } - let mut controller = Controller::new( - cfg, - Box::new(ThresholdEngine::new(ThresholdConfig::default())), - ) - .unwrap(); + let mut controller = Controller::new(cfg, Box::new(default_threshold_engine())).unwrap(); controller .instances .insert(instance.id.clone(), Arc::clone(&instance)); @@ -1419,7 +1410,7 @@ mod tests { vm_state_path: Path::new(&state_dir.path().join("ownership.json")), ..Default::default() }; - let engine = Box::new(ThresholdEngine::new(ThresholdConfig::default())); + let engine = Box::new(default_threshold_engine()); let mut controller = Controller::new(cfg, engine).unwrap(); let good_target = Arc::new(AtomicU32::new(1)); let good = BulkClient { diff --git a/src/daemon.rs b/src/daemon.rs index ed3ce058..a733e724 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -22,7 +22,7 @@ use crate::{ config::Config, controller::{Controller, ControllerError}, dbus, - engines::{EngineError, load_registered_engine}, + engines::ScalingEngine, instance::Instance, util::Path, }; @@ -32,9 +32,6 @@ pub enum DaemonError { #[error(transparent)] Controller(#[from] ControllerError), - #[error(transparent)] - Engine(#[from] EngineError), - #[error("backend watch path `{0}` is not a directory")] InvalidWatchPath(String), @@ -52,7 +49,11 @@ pub enum DaemonError { const INOTIFY_EVENT_BUF_SIZE: usize = 32 * 1024; /// Run until SIGTERM or SIGINT. -pub async fn run(cfg: Config, backends: Vec>) -> Result<(), DaemonError> { +pub async fn run( + cfg: Config, + engine: Box, + backends: Vec>, +) -> Result<(), DaemonError> { let per_vm_destination = if cfg.enable_per_vm_status_line { if cfg.enable_per_vm_main_log { "main log".to_string() @@ -85,7 +86,7 @@ pub async fn run(cfg: Config, backends: Vec>) -> Result<(), Dae "# aggregate line: " ); } - let engine = load_registered_engine(&cfg.engine_config_dir, &cfg.engine)?; + log_effective_config(&cfg, engine.as_ref(), &backends); let mut controller = Controller::new(cfg.clone(), engine)?; let discovered = discover_all(&backends).await; @@ -95,7 +96,7 @@ pub async fn run(cfg: Config, backends: Vec>) -> Result<(), Dae let (dbus_tx, mut dbus_rx) = tokio::sync::mpsc::channel(1); let _dbus_connection = dbus::serve(dbus::Service::new(dbus_tx)).await?; let mut inotify_events = setup_inotify(&backends)?; - let mut poll = interval(Duration::from_secs_f64(cfg.scale_poll_secs)); + let mut poll = interval(Duration::from_secs(cfg.scale_poll_secs)); poll.set_missed_tick_behavior(MissedTickBehavior::Skip); let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?; let mut sigint = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt())?; @@ -126,6 +127,30 @@ pub async fn run(cfg: Config, backends: Vec>) -> Result<(), Dae Ok(()) } +/// Log the configuration in use after defaults, command-line overrides, and +/// duration rounding were applied. +fn log_effective_config(cfg: &Config, engine: &dyn ScalingEngine, backends: &[Box]) { + tracing::info!( + target: "controller", + config = %serde_json::to_value(cfg).unwrap_or_default(), + "effective controller configuration" + ); + tracing::info!( + target: "controller", + engine = engine.name(), + config = %engine.dump_config(), + "effective engine configuration" + ); + for backend in backends { + tracing::info!( + target: "controller", + backend = backend.name(), + config = %backend.dump_config(), + "effective backend configuration" + ); + } +} + /// Discover all backend inventories concurrently. async fn discover_all(backends: &[Box]) -> Vec> { join_all(backends.iter().map(|backend| backend.discover())) diff --git a/src/engines/mod.rs b/src/engines/mod.rs index 15b8c4c9..357e94f1 100644 --- a/src/engines/mod.rs +++ b/src/engines/mod.rs @@ -12,17 +12,28 @@ use futures_util::future::join_all; use linkme::distributed_slice; use thiserror::Error; -use crate::{backends::BackendClientError, config::ConfigError, instance::Instance, util::Path}; +use crate::{ + backends::BackendClientError, + config::{ConfigError, PollConverter}, + instance::Instance, + util::Path, +}; #[cfg(feature = "threshold-engine")] pub mod threshold; +/// Build an engine from ``. Engine durations must be +/// converted with the supplied converter, which reports invalid durations +/// together with the controller's after the engine is built. +pub type EngineBuilder = + fn(&Path, &mut PollConverter) -> Result, EngineError>; + /// One in-tree (or out-of-tree) engine factory registered on [`ENGINES`]. pub struct EngineRegistration { /// Stable name matched against `Config::engine`. pub name: &'static str, - /// Build the engine from ``. - pub build: fn(&Path) -> Result, EngineError>, + /// Build the engine. + pub build: EngineBuilder, } /// Linked-in engine factories. Feature-gated modules append themselves here. @@ -103,8 +114,6 @@ pub struct EngineTickContext { pub psi: PsiSample, /// Monotonically increasing tick sequence number. pub tick_index: u64, - /// Controller poll interval in seconds (for operator-facing durations). - pub scale_poll_secs: f64, } impl Default for EngineTickContext { @@ -116,7 +125,6 @@ impl Default for EngineTickContext { host_cpu_util: 0.0, psi: PsiSample::default(), tick_index: 0, - scale_poll_secs: 10.0, } } } @@ -273,10 +281,11 @@ pub trait ScalingEngine: Send + Sync { pub fn load_registered_engine( engine_config_dir: &Path, name: &str, + durations: &mut PollConverter, ) -> Result, EngineError> { for registration in ENGINES { if registration.name == name { - return (registration.build)(engine_config_dir); + return (registration.build)(engine_config_dir, durations); } } Err(EngineError::NoSuchEngine(name.to_string())) @@ -315,10 +324,11 @@ mod tests { #[test] fn load_registered_engine_finds_threshold_and_rejects_unknown() { let dir = Path::new("/tmp/missing-engines"); - let engine = load_registered_engine(&dir, "threshold").unwrap(); + let mut durations = crate::config::Config::default().poll_period().converter(); + let engine = load_registered_engine(&dir, "threshold", &mut durations).unwrap(); assert_eq!(engine.name(), "threshold"); - match load_registered_engine(&dir, "no-such-engine") { + match load_registered_engine(&dir, "no-such-engine", &mut durations) { Err(EngineError::NoSuchEngine(name)) => assert_eq!(name, "no-such-engine"), Ok(_) => panic!("expected NoSuchEngine"), Err(other) => panic!("expected NoSuchEngine, got {other}"), diff --git a/src/engines/threshold.rs b/src/engines/threshold.rs index 3102812b..0a932356 100644 --- a/src/engines/threshold.rs +++ b/src/engines/threshold.rs @@ -5,14 +5,18 @@ //! Threshold engine: size a worker pool to carry the observed CPU load. -use std::{collections::HashMap, sync::Arc, time::Instant}; +use std::{ + collections::HashMap, + sync::Arc, + time::{Duration, Instant}, +}; use async_trait::async_trait; use serde::{Deserialize, Serialize}; use tokio::sync::Mutex; use crate::{ - config::{ConfigError, deserialize_percent, serialize_percent}, + config::{ConfigError, PollConverter, deserialize_percent, serialize_percent}, engines::{ AppliedOutcome, EngineError, EngineRegistration, EngineTickContext, ScaleAction, ScalingEngine, @@ -24,8 +28,11 @@ use crate::{ /// Registry and configuration name of the threshold engine. pub const ENGINE_NAME: &str = "threshold"; -fn build_threshold_engine(dir: &Path) -> Result, EngineError> { - Ok(Box::new(ThresholdEngine::from_config_dir(dir)?)) +fn build_threshold_engine( + dir: &Path, + durations: &mut PollConverter, +) -> Result, EngineError> { + Ok(Box::new(ThresholdEngine::from_config_dir(dir, durations)?)) } #[linkme::distributed_slice(super::ENGINES)] @@ -35,6 +42,8 @@ static THRESHOLD_ENGINE: EngineRegistration = EngineRegistration { }; /// Scale, sustain, and post-action validation settings. +/// +/// Every duration must be a multiple of the controller poll period. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct ThresholdConfig { @@ -46,9 +55,9 @@ pub struct ThresholdConfig { serialize_with = "serialize_percent" )] pub scale_up_threshold: f64, - /// Consecutive eligible polls required before scale-down. - #[serde(default = "default_scale_down_sustain_polls")] - pub scale_down_sustain_polls: u32, + /// Seconds scale-down must stay eligible before it is proposed. + #[serde(default = "default_scale_down_sustain_secs")] + pub scale_down_sustain_secs: u64, /// Maximum number of threads than can be reduced in a scale down operation. /// 0 means unlimited. #[serde(default = "default_max_scale_down_step")] @@ -69,39 +78,53 @@ pub struct ThresholdConfig { serialize_with = "serialize_percent" )] pub scale_down_revert_drop: f64, - /// Complete samples to wait before validating a successful action. - #[serde(default = "default_scale_validation_sample_polls")] - pub scale_validation_sample_polls: u32, - /// Additional validation polls per target thread after a scale action. + /// Seconds to sample after a successful action before validating it. + #[serde(default = "default_scale_validation_delay_secs")] + pub scale_validation_delay_secs: u64, + /// Additional validation seconds per target thread after a scale action. #[serde(default)] - pub scale_validation_sample_polls_per_thread: u32, + pub scale_validation_delay_per_thread_secs: u64, /// Suppress scale-up after the VM's cgroup reports new throttled CPU time. #[serde(default = "default_true")] pub block_scale_up_when_cgroup_throttled: bool, - /// Polls after a failed scale-up revert during which metrics are ignored. - #[serde(default = "default_post_revert_up_grace_polls")] - pub post_revert_up_grace_polls: u32, - /// Polls after grace spent measuring a settled baseline while scale-up is - /// still blocked. - #[serde(default = "default_post_revert_up_recovery_polls")] - pub post_revert_up_recovery_polls: u32, - /// Cap on the post-revert hold (grace plus recovery), which doubles with - /// each consecutive failed scale-up. Zero disables the backoff. - #[serde(default = "default_post_revert_up_backoff_max_polls")] - pub post_revert_up_backoff_max_polls: u32, - /// Cap on the low-utilisation polls required before scale-down, which - /// doubles with each consecutive failed scale-down. Zero disables the + /// Seconds after a failed scale-up revert during which metrics are + /// ignored. + #[serde(default = "default_post_revert_up_grace_secs")] + pub post_revert_up_grace_secs: u64, + /// Seconds after grace spent measuring a settled baseline while scale-up + /// is still blocked. + #[serde(default = "default_post_revert_up_recovery_secs")] + pub post_revert_up_recovery_secs: u64, + /// Cap in seconds on the post-revert hold (grace plus recovery), which + /// doubles with each consecutive failed scale-up. Zero disables the /// backoff. - #[serde(default = "default_scale_down_backoff_max_polls")] - pub scale_down_backoff_max_polls: u32, + #[serde(default = "default_post_revert_up_backoff_max_secs")] + pub post_revert_up_backoff_max_secs: u64, + /// Cap in seconds on the low-utilisation time required before scale-down, + /// which doubles with each consecutive failed scale-down. Zero disables + /// the backoff. + #[serde(default = "default_scale_down_backoff_max_secs")] + pub scale_down_backoff_max_secs: u64, +} + +/// [`ThresholdConfig`] durations converted to controller polls. +#[derive(Debug, Clone, Copy)] +struct ThresholdPolls { + scale_down_sustain: u32, + scale_validation_delay: u32, + scale_validation_delay_per_thread: u32, + post_revert_up_grace: u32, + post_revert_up_recovery: u32, + post_revert_up_backoff_max: u32, + scale_down_backoff_max: u32, } fn default_scale_up_threshold() -> f64 { 0.95 } -fn default_scale_down_sustain_polls() -> u32 { - 3 +fn default_scale_down_sustain_secs() -> u64 { + 30 } fn default_max_scale_down_step() -> u32 { @@ -114,23 +137,23 @@ fn default_scale_up_min_gain() -> f64 { fn default_scale_down_revert_drop() -> f64 { 0.05 } -fn default_scale_validation_sample_polls() -> u32 { - 2 +fn default_scale_validation_delay_secs() -> u64 { + 20 } fn default_true() -> bool { true } -fn default_post_revert_up_grace_polls() -> u32 { - 1 +fn default_post_revert_up_grace_secs() -> u64 { + 10 } -fn default_post_revert_up_recovery_polls() -> u32 { - 2 +fn default_post_revert_up_recovery_secs() -> u64 { + 20 } -fn default_post_revert_up_backoff_max_polls() -> u32 { - 30 +fn default_post_revert_up_backoff_max_secs() -> u64 { + 300 } -fn default_scale_down_backoff_max_polls() -> u32 { - 120 +fn default_scale_down_backoff_max_secs() -> u64 { + 1200 } /// `base` doubled `doublings` times, capped at `max` (never below `base`). @@ -147,17 +170,17 @@ impl Default for ThresholdConfig { fn default() -> Self { Self { scale_up_threshold: default_scale_up_threshold(), - scale_down_sustain_polls: default_scale_down_sustain_polls(), + scale_down_sustain_secs: default_scale_down_sustain_secs(), max_scale_down_step: default_max_scale_down_step(), scale_up_min_gain: default_scale_up_min_gain(), scale_down_revert_drop: default_scale_down_revert_drop(), - scale_validation_sample_polls: default_scale_validation_sample_polls(), - scale_validation_sample_polls_per_thread: 0, + scale_validation_delay_secs: default_scale_validation_delay_secs(), + scale_validation_delay_per_thread_secs: 0, block_scale_up_when_cgroup_throttled: true, - post_revert_up_grace_polls: default_post_revert_up_grace_polls(), - post_revert_up_recovery_polls: default_post_revert_up_recovery_polls(), - post_revert_up_backoff_max_polls: default_post_revert_up_backoff_max_polls(), - scale_down_backoff_max_polls: default_scale_down_backoff_max_polls(), + post_revert_up_grace_secs: default_post_revert_up_grace_secs(), + post_revert_up_recovery_secs: default_post_revert_up_recovery_secs(), + post_revert_up_backoff_max_secs: default_post_revert_up_backoff_max_secs(), + scale_down_backoff_max_secs: default_scale_down_backoff_max_secs(), } } } @@ -187,6 +210,39 @@ impl ThresholdConfig { } Ok(()) } + + /// Convert every duration to polls, rounding durations up in place when + /// `converter` allows it. + fn resolve_polls(&mut self, converter: &mut PollConverter) -> ThresholdPolls { + ThresholdPolls { + scale_down_sustain: converter + .polls("scale_down_sustain_secs", &mut self.scale_down_sustain_secs), + scale_validation_delay: converter.polls( + "scale_validation_delay_secs", + &mut self.scale_validation_delay_secs, + ), + scale_validation_delay_per_thread: converter.polls( + "scale_validation_delay_per_thread_secs", + &mut self.scale_validation_delay_per_thread_secs, + ), + post_revert_up_grace: converter.polls( + "post_revert_up_grace_secs", + &mut self.post_revert_up_grace_secs, + ), + post_revert_up_recovery: converter.polls( + "post_revert_up_recovery_secs", + &mut self.post_revert_up_recovery_secs, + ), + post_revert_up_backoff_max: converter.polls( + "post_revert_up_backoff_max_secs", + &mut self.post_revert_up_backoff_max_secs, + ), + scale_down_backoff_max: converter.polls( + "scale_down_backoff_max_secs", + &mut self.scale_down_backoff_max_secs, + ), + } + } } /// One successful action waiting for post-action IOPS validation. @@ -390,6 +446,9 @@ fn iops_rate(start: IoSample, end: IoSample) -> u64 { /// Scaling engine driven by CPU utilisation thresholds. pub struct ThresholdEngine { cfg: ThresholdConfig, + /// `cfg` durations in polls. + polls: ThresholdPolls, + poll_period: Duration, // TODO the controller already maintains this mapping, an alternative would // be for this engine to define a private type and keep a reference to @@ -398,16 +457,25 @@ pub struct ThresholdEngine { } impl ThresholdEngine { - /// Construct an engine from validated settings. - pub fn new(cfg: ThresholdConfig) -> Self { - Self { + /// Validate `cfg` and convert its durations to polls. + /// + /// The engine must not be used before `durations` finished without error. + pub fn new( + mut cfg: ThresholdConfig, + durations: &mut PollConverter, + ) -> Result { + cfg.validate()?; + let polls = cfg.resolve_polls(durations); + Ok(Self { cfg, + polls, + poll_period: durations.period(), state: Mutex::new(HashMap::new()), - } + }) } /// Load `threshold.json`, falling back to built-in defaults when absent. - pub fn from_config_dir(dir: &Path) -> Result { + pub fn from_config_dir(dir: &Path, durations: &mut PollConverter) -> Result { let path = Path::new(&dir.join(format!("{ENGINE_NAME}.json"))); // TODO TOCTOU, blindly load and return default if ENOENT let cfg: ThresholdConfig = if path.exists() { @@ -415,20 +483,20 @@ impl ThresholdEngine { } else { ThresholdConfig::default() }; - cfg.validate()?; - Ok(Self::new(cfg)) + Ok(Self::new(cfg, durations)?) } - /// Return the engine's effective configuration. + /// Return the engine's effective configuration, with any rounded-up + /// durations. pub fn config(&self) -> &ThresholdConfig { &self.cfg } /// Return the validation polls for an action's resulting thread count. fn validation_sample_polls(&self, target_thread_count: u32) -> u32 { - self.cfg.scale_validation_sample_polls.saturating_add( - self.cfg - .scale_validation_sample_polls_per_thread + self.polls.scale_validation_delay.saturating_add( + self.polls + .scale_validation_delay_per_thread .saturating_mul(target_thread_count), ) } @@ -514,11 +582,11 @@ impl ThresholdEngine { /// failed scale-ups. The backoff extends grace so the recovery baseline /// window keeps its configured length. fn post_revert_up_grace_polls(&self, failures: u32) -> u32 { - let recovery = self.cfg.post_revert_up_recovery_polls; + let recovery = self.polls.post_revert_up_recovery; let hold = backoff_polls( - self.cfg.post_revert_up_grace_polls.saturating_add(recovery), + self.polls.post_revert_up_grace.saturating_add(recovery), failures.saturating_sub(1), - self.cfg.post_revert_up_backoff_max_polls, + self.polls.post_revert_up_backoff_max, ); hold.saturating_sub(recovery) } @@ -527,18 +595,20 @@ impl ThresholdEngine { /// consecutive failed scale-downs. fn scale_down_sustain_polls(&self, failures: u32) -> u32 { backoff_polls( - self.cfg.scale_down_sustain_polls, + self.polls.scale_down_sustain, failures, - self.cfg.scale_down_backoff_max_polls, + self.polls.scale_down_backoff_max, ) } - /// Seconds until a post-scale IOPS sample is evaluated. - fn validation_after_secs(&self, scale_poll_secs: f64, target_thread_count: u32) -> f64 { + /// Time until a post-scale IOPS sample is evaluated. + fn validation_after(&self, target_thread_count: u32) -> Duration { // Countdown consumes the effective validation polls, then the next // tick performs the comparison. - // TODO use Duration - (f64::from(self.validation_sample_polls(target_thread_count)) + 1.0) * scale_poll_secs + self.poll_period.saturating_mul( + self.validation_sample_polls(target_thread_count) + .saturating_add(1), + ) } } @@ -580,7 +650,7 @@ fn log_scale_up_decision( target: u32, current_iops: u64, min_target_iops: u64, - validation_after_secs: f64, + validation_after: Duration, ) { tracing::info!( target: "engine", @@ -590,7 +660,7 @@ fn log_scale_up_decision( thr = %format!("{thread_count}->{target}"), current_iops, min_target_iops, - validation_after_secs, + validation_after_secs = validation_after.as_secs(), "threshold scaling decision" ); } @@ -715,7 +785,7 @@ impl ScalingEngine for ThresholdEngine { &instance.id, failures, grace_polls, - self.cfg.post_revert_up_recovery_polls, + self.polls.post_revert_up_recovery, ); } else { instance_state.consecutive_failed_downs = @@ -730,8 +800,8 @@ impl ScalingEngine for ThresholdEngine { return ScaleAction::Revert(target); } - let block_scale_up = instance_state - .advance_post_revert_up_hold(sample, self.cfg.post_revert_up_recovery_polls); + let block_scale_up = + instance_state.advance_post_revert_up_hold(sample, self.polls.post_revert_up_recovery); // Grace ignores metrics; recovery updates the rate inside advance. // Once the hold ends (or if there was none), refresh the normal baseline. if !block_scale_up { @@ -760,13 +830,13 @@ impl ScalingEngine for ThresholdEngine { // or equal to the max_thread_count here. let target = (thread_count + 1).min(context.max_thread_count); let current_iops = instance_state.current_baseline_iops; - let (min_target_iops, validation_after_secs) = if self.cfg.scale_up_min_gain > 0.0 { + let (min_target_iops, validation_after) = if self.cfg.scale_up_min_gain > 0.0 { ( ((current_iops as f64) * (1.0 + self.cfg.scale_up_min_gain)).ceil() as u64, - self.validation_after_secs(context.scale_poll_secs, target), + self.validation_after(target), ) } else { - (current_iops, 0.0) + (current_iops, Duration::ZERO) }; log_scale_up_decision( instance, @@ -775,7 +845,7 @@ impl ScalingEngine for ThresholdEngine { target, current_iops, min_target_iops, - validation_after_secs, + validation_after, ); return ScaleAction::Up(target); } @@ -865,6 +935,7 @@ mod tests { }; use crate::{ backends::BackendClientError, + config::PollPeriod, engines::{AppliedOutcome, EngineTickContext, ScaleAction, ScalingEngine}, instance::{ Instance, InstanceClient, InstancePerfSample, InstanceStatus, ThreadPoolSnapshot, @@ -921,10 +992,28 @@ mod tests { instance } + const POLL_SECS: u64 = 10; + + /// Return the seconds `count` polls span. + fn polls(count: u64) -> u64 { + POLL_SECS * count + } + + /// Build an engine whose durations are converted with `round_up`. + fn try_new_engine(cfg: ThresholdConfig, round_up: bool) -> Result { + let mut durations = PollPeriod::new(POLL_SECS, round_up).converter(); + let engine = ThresholdEngine::new(cfg, &mut durations).map_err(|e| e.to_string())?; + durations.finish().map_err(|e| e.to_string())?; + Ok(engine) + } + + fn new_engine(cfg: ThresholdConfig) -> ThresholdEngine { + try_new_engine(cfg, false).unwrap() + } + fn context_at(now: Instant) -> EngineTickContext { EngineTickContext { now, - scale_poll_secs: 10.0, ..EngineTickContext::default() } } @@ -994,10 +1083,10 @@ mod tests { #[fixture] fn engine() -> ThresholdEngine { - ThresholdEngine::new(ThresholdConfig { + new_engine(ThresholdConfig { scale_up_threshold: 0.8, scale_up_min_gain: 0.05, - scale_validation_sample_polls: 0, + scale_validation_delay_secs: 0, ..Default::default() }) } @@ -1112,12 +1201,12 @@ mod tests { #[tokio::test] async fn test_failed_scale_up_grace_then_recovery_baseline(#[future] instance: Arc) { let instance = instance.await; - let engine = ThresholdEngine::new(ThresholdConfig { + let engine = new_engine(ThresholdConfig { scale_up_threshold: 0.8, scale_up_min_gain: 0.05, - scale_validation_sample_polls: 0, - post_revert_up_grace_polls: 1, - post_revert_up_recovery_polls: 2, + scale_validation_delay_secs: 0, + post_revert_up_grace_secs: polls(1), + post_revert_up_recovery_secs: polls(2), ..Default::default() }); let t0 = Instant::now(); @@ -1193,13 +1282,13 @@ mod tests { #[tokio::test] async fn test_scale_down_allowed_during_post_revert_up_hold(#[future] instance: Arc) { let instance = instance.await; - let engine = ThresholdEngine::new(ThresholdConfig { + let engine = new_engine(ThresholdConfig { scale_up_threshold: 0.8, scale_up_min_gain: 0.05, - scale_validation_sample_polls: 0, - scale_down_sustain_polls: 1, - post_revert_up_grace_polls: 2, - post_revert_up_recovery_polls: 2, + scale_validation_delay_secs: 0, + scale_down_sustain_secs: polls(1), + post_revert_up_grace_secs: polls(2), + post_revert_up_recovery_secs: polls(2), ..Default::default() }); let t0 = Instant::now(); @@ -1240,11 +1329,11 @@ mod tests { #[tokio::test] async fn scale_down_keeps_shrink_when_util_stays_low(#[future] instance: Arc) { let instance = instance.await; - let engine = ThresholdEngine::new(ThresholdConfig { + let engine = new_engine(ThresholdConfig { scale_up_threshold: 0.8, scale_down_revert_drop: 0.05, - scale_validation_sample_polls: 0, - scale_down_sustain_polls: 1, + scale_validation_delay_secs: 0, + scale_down_sustain_secs: polls(1), ..Default::default() }); let t0 = Instant::now(); @@ -1280,10 +1369,10 @@ mod tests { #[future] instance: Arc, ) { let instance = instance.await; - let engine = ThresholdEngine::new(ThresholdConfig { + let engine = new_engine(ThresholdConfig { scale_up_threshold: 0.8, scale_down_revert_drop: 0.05, - scale_validation_sample_polls: 0, + scale_validation_delay_secs: 0, ..Default::default() }); let t0 = Instant::now(); @@ -1364,13 +1453,13 @@ mod tests { #[tokio::test] async fn test_failed_scale_ups_back_off_until_kept(#[future] instance: Arc) { let instance = instance.await; - let engine = ThresholdEngine::new(ThresholdConfig { + let engine = new_engine(ThresholdConfig { scale_up_threshold: 0.8, scale_up_min_gain: 0.05, - scale_validation_sample_polls: 0, - post_revert_up_grace_polls: 1, - post_revert_up_recovery_polls: 2, - post_revert_up_backoff_max_polls: 12, + scale_validation_delay_secs: 0, + post_revert_up_grace_secs: polls(1), + post_revert_up_recovery_secs: polls(2), + post_revert_up_backoff_max_secs: polls(12), ..Default::default() }); engine.on_instance_added(&instance).await; @@ -1409,12 +1498,12 @@ mod tests { #[tokio::test] async fn test_failed_scale_downs_back_off_until_scale_up(#[future] instance: Arc) { let instance = instance.await; - let engine = ThresholdEngine::new(ThresholdConfig { + let engine = new_engine(ThresholdConfig { scale_up_threshold: 0.8, scale_down_revert_drop: 0.05, - scale_validation_sample_polls: 0, - scale_down_sustain_polls: 3, - scale_down_backoff_max_polls: 20, + scale_validation_delay_secs: 0, + scale_down_sustain_secs: polls(3), + scale_down_backoff_max_secs: polls(20), ..Default::default() }); engine.on_instance_added(&instance).await; @@ -1509,9 +1598,9 @@ mod tests { #[test] fn test_validation_polls_increase_with_target_thread_count() { - let engine = ThresholdEngine::new(ThresholdConfig { - scale_validation_sample_polls: 2, - scale_validation_sample_polls_per_thread: 3, + let engine = new_engine(ThresholdConfig { + scale_validation_delay_secs: polls(2), + scale_validation_delay_per_thread_secs: polls(3), ..Default::default() }); @@ -1547,7 +1636,7 @@ mod tests { set_observation(&instance, 5, 0.88, 220_000).await; - for i in 0..engine.cfg.scale_validation_sample_polls { + for i in 0..engine.polls.scale_validation_delay { assert_eq!( engine .evaluate( @@ -1585,10 +1674,10 @@ mod tests { #[future] instance: Arc, ) { let instance = instance.await; - let engine = ThresholdEngine::new(ThresholdConfig { + let engine = new_engine(ThresholdConfig { scale_up_threshold: 0.8, scale_up_min_gain: 0.05, - scale_validation_sample_polls: 2, + scale_validation_delay_secs: polls(2), ..Default::default() }); let t0 = Instant::now(); @@ -1609,7 +1698,7 @@ mod tests { .await; set_observation(&instance, 5, 0.88, 140_000).await; - for i in 0..engine.cfg.scale_validation_sample_polls { + for i in 0..engine.polls.scale_validation_delay { assert_eq!( engine .evaluate( @@ -1670,19 +1759,64 @@ mod tests { #[test] fn missing_config_file_uses_defaults() { let dir = tempfile::tempdir().unwrap(); - let engine = - ThresholdEngine::from_config_dir(&Path::new(dir.path().to_str().unwrap())).unwrap(); + let mut durations = PollPeriod::new(POLL_SECS, false).converter(); + let engine = ThresholdEngine::from_config_dir( + &Path::new(dir.path().to_str().unwrap()), + &mut durations, + ) + .unwrap(); + durations.finish().unwrap(); assert_eq!(engine.name(), super::ENGINE_NAME); let _ = engine.config(); } + /// Tests that the default durations are multiples of the default + /// controller poll period. + #[test] + fn test_defaults_are_multiples_of_default_poll_period() { + let mut durations = crate::config::Config::default().poll_period().converter(); + ThresholdEngine::new(ThresholdConfig::default(), &mut durations).unwrap(); + durations.finish().unwrap(); + } + + /// Tests that every duration that is not a multiple of the poll period is + /// reported in a single error. + #[test] + fn test_durations_not_poll_multiples_are_all_rejected() { + let cfg = ThresholdConfig { + scale_down_sustain_secs: 25, + post_revert_up_grace_secs: 5, + ..Default::default() + }; + let err = try_new_engine(cfg, false).err().unwrap(); + assert!(err.contains("scale_down_sustain_secs (25)"), "{err}"); + assert!(err.contains("post_revert_up_grace_secs (5)"), "{err}"); + } + + /// Tests that rounding converts durations to the next whole poll and + /// that the effective configuration shows the rounded values. + #[test] + fn test_rounded_durations_are_used_and_dumped() { + let cfg = ThresholdConfig { + scale_down_sustain_secs: 25, + scale_validation_delay_secs: 1, + ..Default::default() + }; + let engine = try_new_engine(cfg, true).unwrap(); + assert_eq!(engine.polls.scale_down_sustain, 3); + assert_eq!(engine.polls.scale_validation_delay, 1); + let dumped = engine.dump_config(); + assert_eq!(dumped["scale_down_sustain_secs"], 30); + assert_eq!(dumped["scale_validation_delay_secs"], 10); + } + /// Test that CPU utilisation below scale-up/down thresholds yields /// Hold/`None`. #[rstest] #[tokio::test] async fn evaluate_holds_when_util_below_thresholds(#[future] instance: Arc) { let instance = instance.await; - let engine = ThresholdEngine::new(ThresholdConfig::default()); + let engine = new_engine(ThresholdConfig::default()); let action = engine .evaluate(&instance, &context_at(Instant::now())) .await; diff --git a/src/main.rs b/src/main.rs index feaff3f0..05b42bd5 100644 --- a/src/main.rs +++ b/src/main.rs @@ -5,7 +5,7 @@ //! `io-thread-controller` daemon entry point. -use std::io::ErrorKind; +use std::{io::ErrorKind, process::ExitCode}; use clap::{self, CommandFactory, FromArgMatches, Parser}; use io_thread_controller::{ @@ -13,6 +13,7 @@ use io_thread_controller::{ backends::registered_backends, config::{Config, ConfigError, dump_default_config, load_json_with_includes, validate_config}, daemon::{DaemonError, VERSION, run}, + engines::{EngineError, load_registered_engine}, per_vm_log::{self, PER_VM_LOG_DIR, PerVmLog}, util::Path, }; @@ -34,6 +35,9 @@ enum IoThreadControllerError { #[error(transparent)] Daemon(#[from] DaemonError), + #[error(transparent)] + Engine(#[from] EngineError), + #[error("no backend `{0}`")] NoSuchBackend(String), } @@ -83,6 +87,10 @@ struct Cli { /// Log scaling decisions without calling set_thread_count. #[arg(long)] dry_run: bool, + /// Round durations that are not multiples of `scale_poll_secs` up to the + /// next multiple, with a warning, instead of refusing to start. + #[arg(long)] + round_up_durations: bool, } /// See `Cli::log_style`. @@ -95,7 +103,17 @@ enum LogStyle { } #[tokio::main] -async fn main() -> Result<(), IoThreadControllerError> { +async fn main() -> ExitCode { + match run_daemon().await { + Ok(()) => ExitCode::SUCCESS, + Err(error) => { + eprintln!("error: {error}"); + ExitCode::FAILURE + } + } +} + +async fn run_daemon() -> Result<(), IoThreadControllerError> { let bootstrap_cfg = Config::default(); let bootstrap_backends = registered_backends(&bootstrap_cfg)?; let mut command = Cli::command(); @@ -140,13 +158,18 @@ async fn main() -> Result<(), IoThreadControllerError> { if cli.dry_run { cfg.dry_run = true; } - validate_config(&cfg)?; + if cli.round_up_durations { + cfg.round_up_durations = true; + } + let mut durations = validate_config(&mut cfg)?; + let engine = load_registered_engine(&cfg.engine_config_dir, &cfg.engine, &mut durations)?; + durations.finish()?; per_vm_log.set_dir( (cfg.enable_per_vm_status_line && !cfg.enable_per_vm_main_log) .then(|| Path::from(PER_VM_LOG_DIR)), ); let backends = registered_backends(&cfg)?; - run(cfg, backends).await?; + run(cfg, engine, backends).await?; Ok(()) } From 180df311e2f9975610c02d2a8c1c5303d66f6bf6 Mon Sep 17 00:00:00 2001 From: Thanos Makatos Date: Fri, 23 Oct 2026 08:11:11 +0000 Subject: [PATCH 2/2] Document every controller and threshold engine setting Add a Configuration section that explains where configuration files are read from, how defaults apply, and every controller and threshold engine setting, and move the threshold validation and backoff description into it. AI-generated: everything. --- README.md | 153 ++++++++++++++++++++++++++++++++++++++++++++++-------- 1 file changed, 130 insertions(+), 23 deletions(-) diff --git a/README.md b/README.md index e35b01c8..d0da2f55 100644 --- a/README.md +++ b/README.md @@ -84,13 +84,32 @@ managed and unmanaged classifications across daemon restarts. +------------------------------------------+ ``` -## Configuration includes +## Configuration + +The daemon reads its controller configuration from +`/etc/io-thread-controller.json`, or from the file given with `--config`. The +controller configuration names the scaling engine and the directories holding +the engine and backend files: the active engine reads +`/.json`, and each backend reads its own file from +`backend_config_dir`. Every file is JSON, and unknown keys are rejected so that +misspelt settings fail loudly. + +If the controller file does not exist, the daemon runs with built-in defaults, +which `--dump-config` prints. A missing engine file likewise selects the +engine's defaults. Settings omitted from a file that does exist take the +defaults listed below. + +At startup the daemon logs the effective controller, engine, and backend +configuration, after defaults, command-line overrides, and duration rounding +have been applied. + +### Includes Controller, engine, and backend JSON files may contain an `include` field. Its path is resolved relative to the including file, loaded recursively, and merged before the including file's keys are applied. -## Durations +### Durations Durations are integer seconds in settings ending in `_secs`. Every duration, including engine ones, must be a whole multiple of `scale_poll_secs`, and zero @@ -100,9 +119,115 @@ therefore invalidate other settings. To try a new poll period without rewriting the other settings, set `round_up_durations` or pass `--round-up-durations`. Each offending duration is -then rounded up to the next multiple with a warning. At startup the daemon logs -the effective controller, engine, and backend configuration, which shows the -rounded values. +then rounded up to the next multiple with a warning, and the effective +configuration logged at startup shows the rounded values. + +Percentages are written as numbers from 0 to 100 in settings ending in +`_percent`. + +### Controller + +The first five settings are required when the controller file exists. + +| Setting | Default | Meaning | +|---|---|---| +| `engine` | required | Scaling engine name, e.g. `threshold`. | +| `engine_config_dir` | required | Directory holding `.json`. | +| `backend_config_dir` | required | Directory holding backend files, e.g. `qemu.json`. | +| `vm_state_path` | required | VM ownership file. | +| `scale_poll_secs` | required | Poll period in seconds; must be at least 1. | +| `round_up_durations` | `false` | Round durations up to poll multiples instead of rejecting them. | +| `min_thread_count` | `1` | Fewest worker threads the controller will set. | +| `max_thread_count` | `8` | Most worker threads the controller will set. | +| `host_cpu_scale_up_ceiling_percent` | `90` | Host CPU use that blocks scale-up; `0` disables. | +| `cooldown_secs` | `30` | Per-VM pause after a successful scale action. | +| `max_instances_adjusted_per_poll` | `0` | Most VMs resized per poll; `0` means no limit. | +| `dry_run` | `false` | Log decisions without resizing pools. | +| `enable_per_vm_status_line` | `true` | Write a status line per VM every poll. | +| `enable_per_vm_main_log` | `false` | Send per-VM status lines to the main log instead of per-VM files. | +| `enable_aggregate_status_line` | `true` | Write one fleet-wide status line every poll. | +| `print_status_header` | `false` | Log a legend of the status-line fields at startup. | +| `refresh_cgroup_on_each_read` | `false` | Re-read the VM's cgroup CPU limits on every refresh. | +| `experimental_psi_monitoring` | `false` | Sample host pressure-stall information every poll. | + +Every `scale_poll_secs` the controller refreshes each VM, asks the engine for +decisions, and applies them. The thread-count bounds are hard limits: an +engine proposal outside `min_thread_count` and `max_thread_count`, or above the +VM's vCPU count, is not applied. `min_thread_count` must be at least 1 and no +greater than `max_thread_count`. When the controller file does not exist, +`min_thread_count` defaults to 2 rather than 1. + +`host_cpu_scale_up_ceiling_percent` protects the host: while overall host CPU +use is at or above it, scale-ups are blocked, but scale-downs and reverts still +apply. `cooldown_secs` starts after every successful scale-up, scale-down, or +revert and blocks further automatic scale-ups and scale-downs of that VM until +it expires; reverts are never blocked by it. + +`max_instances_adjusted_per_poll` limits how disruptive a single poll can be. +When more VMs need resizing, scale-downs are applied before scale-ups, and the +remaining VMs are reconsidered on the next poll. + +`vm_state_path` records which VMs the controller manages and which it leaves +to the operator, as classified by the backend when it first sees each VM. The +default, `/run/io-thread-controller/vm-ownership.json`, survives daemon +restarts but not a host reboot. + +By default each VM's status line goes to +`/var/log/io-thread-controller/vms/.log`, together with that VM's log +events; see [Status line](#status-line) for the format. +`enable_per_vm_main_log` sends the status lines to the main log instead. + +`dry_run` and `print_status_header` can also be enabled with `--dry-run` and +`--print-status-header`. `refresh_cgroup_on_each_read` and +`experimental_psi_monitoring` are described under +[Cgroup throttling](#cgroup-throttling) and +[Experimental PSI monitoring](#experimental-psi-monitoring). + +### Threshold engine + +The threshold engine reads `threshold.json` and sizes each VM's worker pool to +the CPU load it observes. + +| Setting | Default | Meaning | +|---|---|---| +| `scale_up_threshold_percent` | `95` | Average per-thread CPU above which a thread is added. | +| `scale_down_sustain_secs` | `30` | How long scale-down must stay possible before it happens. | +| `max_scale_down_step` | `0` | Most threads removed at once; `0` means no limit. | +| `scale_up_min_gain_percent` | `5` | IOPS gain a scale-up must deliver; `0` disables the check. | +| `scale_down_revert_drop_percent` | `5` | IOPS loss that reverts a scale-down; `0` disables the check. | +| `scale_validation_delay_secs` | `20` | How long to measure after an action before validating it. | +| `scale_validation_delay_per_thread_secs` | `0` | Extra validation time per thread in the new pool. | +| `block_scale_up_when_cgroup_throttled` | `true` | Skip scale-up while the VM's cgroup is being CPU-throttled. | +| `post_revert_up_grace_secs` | `10` | Time after a reverted scale-up during which metrics are ignored. | +| `post_revert_up_recovery_secs` | `20` | Time after the grace period spent re-measuring IOPS. | +| `post_revert_up_backoff_max_secs` | `300` | Cap on the post-revert hold; `0` disables backoff. | +| `scale_down_backoff_max_secs` | `1200` | Cap on the scale-down sustain time; `0` disables backoff. | + +Each poll, the engine compares the pool's average per-thread CPU use with +`scale_up_threshold_percent`. Above it, the engine proposes one more thread. +Otherwise it works out the smallest pool that would keep the average at or +below the threshold. If that is smaller than the current pool for +`scale_down_sustain_secs`, the engine shrinks the pool to it, by at most +`max_scale_down_step` threads at a time. + +After a successful action, the engine measures IOPS for +`scale_validation_delay_secs`, plus `scale_validation_delay_per_thread_secs` +for each thread in the resulting pool, which gives larger pools more time to +settle. A scale-up is reverted unless IOPS gained at least +`scale_up_min_gain_percent`. A scale-down is reverted when IOPS fell by more +than `scale_down_revert_drop_percent` while CPU use is above the scale-up +threshold; if CPU use stayed low, demand simply left, and the smaller pool is +kept. Setting either percentage to zero disables that direction's validation. + +Repeated failures back off per VM. After a reverted scale-up, scale-up is +blocked for `post_revert_up_grace_secs`, while the pool settles, plus +`post_revert_up_recovery_secs`, while the engine measures a fresh IOPS baseline +for the next attempt. That hold doubles with each consecutive failed scale-up, +up to `post_revert_up_backoff_max_secs`, and scale-down remains allowed +throughout. After a reverted scale-down, the next scale-down needs +`scale_down_sustain_secs` of low utilisation doubled once per consecutive +failure, up to `scale_down_backoff_max_secs`. A kept action resets its +direction's backoff, as does an applied action in the opposite direction. ## Thread-name filtering @@ -145,24 +270,6 @@ queries and thread-count changes need an independently stateful handle for each VM. This keeps the controller independent of libvirt, QMP, control sockets, and other transport details. -After a successful action, the engine samples for -`scale_validation_delay_secs`. The optional -`scale_validation_delay_per_thread_secs` setting adds that much time for each -thread in the resulting pool, allowing larger pools more time to settle. -A scale-up is reverted unless IOPS gained at least -`scale_up_min_gain_percent`; a scale-down is reverted when IOPS lost more than -`scale_down_revert_drop_percent`. Setting either percentage to zero disables -that direction's validation. - -Repeated failures back off per VM. After a reverted scale-up, scale-up is -blocked for `post_revert_up_grace_secs` plus `post_revert_up_recovery_secs`; -that hold doubles with each consecutive failed scale-up, up to -`post_revert_up_backoff_max_secs`. After a reverted scale-down, the next -scale-down needs `scale_down_sustain_secs` of low utilisation doubled once per -consecutive failure, up to `scale_down_backoff_max_secs`. A kept action -resets its direction's backoff, as does an applied action in the opposite -direction. Setting a maximum to zero disables that direction's backoff. - ## QEMU backend The QEMU backend discovers active VMs through libvirt and sends QMP commands