From ee4289b15a6ff045de7b1fc7f01a1d8ea881615d Mon Sep 17 00:00:00 2001 From: Thanos Makatos Date: Fri, 23 Oct 2026 06:57:57 +0000 Subject: [PATCH] Express configured durations in seconds, not polls Expressing parameters in poll cycles makes it trivial for the implementation but harder for the operator to understand since they have to multiply the parameters by the poll cycle duration to get to the actual value. Express everything in seconds instead, even if that complicates the implementation. 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, manual fixup of the commit message 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 8eeaf0c0..284eabb6 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(()) }