diff --git a/README.md b/README.md index e4ece764..d0da2f55 100644 --- a/README.md +++ b/README.md @@ -84,12 +84,151 @@ 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 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, 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 Backends may select fallback `/proc` CPU samples with regular expressions and @@ -131,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 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. -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`; -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 -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 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(()) }