From 67ba3d243279685ace436cc5bd4d21d32f8ca6cb Mon Sep 17 00:00:00 2001 From: Thanos Makatos Date: Thu, 22 Oct 2026 10:29:12 +0000 Subject: [PATCH 01/17] Space-separate every field in per-VM log Grouping related values with '/' makes the per-VM log awkward to process with standard tools. Separate every field with a space instead. Missing latency renders one '-' per cell so the number of fields stays fixed. The main log keeps the '/' grouping, since its values are keyed. Fixes #100 Signed-off-by: Thanos Makatos --- README.md | 29 ++++++++++---------- src/controller.rs | 70 ++++++++++++++++++++++++++++++++--------------- src/daemon.rs | 11 ++++---- src/rolling.rs | 7 +++-- 4 files changed, 73 insertions(+), 44 deletions(-) diff --git a/README.md b/README.md index 09107c48..bb679cc7 100644 --- a/README.md +++ b/README.md @@ -179,28 +179,29 @@ on the `status` tracing target. With `--print-status-header`, it also emits a By default each VM's line is appended to `/var/log/io-thread-controller/vms/.log`, so the line itself does not repeat -the VM identifier. Fields are positional, without keys. The examples are -wrapped with backslashes for readability; each emitted record occupies one -line. +the VM identifier. Fields are positional, without keys, and separated by single +spaces. The examples are wrapped with backslashes for readability; each emitted +record occupies one line. ``` -2026-01-01T00:00:00.000Z status=4 155593/0/0 154995/153840/151220 \ - 20394/0 90/87/358 40/80/120/45 - 23/16/19 +2026-01-01T00:00:00.000Z status=4 155593 0 0 154995 153840 151220 \ + 20394 0 90 87 358 40 80 120 45 - - - - 23 16 19 ``` After the leading timestamp, the `status=` prefix distinguishes status records from `event=` records, and the fields are: 1. matched worker threads. -2. current read / write / other operations per second. -3. average total IOPS over rolling 1m / 5m / 15m windows. -4. current read / write bandwidth in MB/s. -5. average / median / total worker CPU percentage. -6. read latency p50 / p95 / p99 / average in microseconds. -7. write latency p50 / p95 / p99 / average in microseconds. -8. CPU microseconds per completed I/O over rolling 1m / 5m / 15m windows. - -A dash (`-`) means no complete sample is available for that window. +2. current read, write, and other operations per second (3 fields). +3. average total IOPS over rolling 1m, 5m, and 15m windows (3 fields). +4. current read and write bandwidth in MB/s (2 fields). +5. average, median, and total worker CPU percentage (3 fields). +6. read latency p50, p95, p99, and average in microseconds (4 fields). +7. write latency p50, p95, p99, and average in microseconds (4 fields). +8. CPU microseconds per completed I/O over rolling 1m, 5m, and 15m windows (3 + fields). + +A dash (`-`) means no sample is available for that field. ### Aggregate line diff --git a/src/controller.rs b/src/controller.rs index e5026061..4113c52f 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -964,7 +964,7 @@ impl Controller { "{} {} {}", self.instances.len(), total_threads, - format_optional_cells(aggregate) + format_optional_cells(aggregate, "/") ); } } @@ -975,21 +975,24 @@ impl Controller { ) -> String { let cpu = compute_cpu_stats(status); format!( - "status={} {}/{}/{} {} {}/{} {}/{}/{} {} {} {}", + "status={} {} {} {} {} {} {} {} {} {} {} {} {}", status.thread_count, status.read_iops, status.write_iops, status.other_iops, - format_optional_cells(iops_windows), + format_optional_cells(iops_windows, " "), status.read_bytes_per_second / 1_000_000, status.write_bytes_per_second / 1_000_000, cpu.avg_pct, cpu.median_pct, cpu.total_pct, - format_latency_cell(status.read_latency_us), - format_latency_cell(status.write_latency_us), - format_1_5_15(&status.rolling, |rolling, window| rolling - .cpu_us_per_io_over(window)), + format_latency_cell(status.read_latency_us, " "), + format_latency_cell(status.write_latency_us, " "), + format_1_5_15( + &status.rolling, + |rolling, window| rolling.cpu_us_per_io_over(window), + " " + ), ) } @@ -1019,11 +1022,11 @@ impl Controller { read_iops = status.read_iops, write_iops = status.write_iops, other_iops = status.other_iops, - iops_1_5_15m = %format_optional_cells(iops_windows), + iops_1_5_15m = %format_optional_cells(iops_windows, "/"), read_bw_mb_s = status.read_bytes_per_second / 1_000_000, write_bw_mb_s = status.write_bytes_per_second / 1_000_000, - read_latency_us = %format_latency_cell(status.read_latency_us), - write_latency_us = %format_latency_cell(status.write_latency_us) + read_latency_us = %format_latency_cell(status.read_latency_us, "/"), + write_latency_us = %format_latency_cell(status.write_latency_us, "/") ); } @@ -1157,8 +1160,9 @@ fn host_cpu_utilisation(previous: HostCpuSample, current: HostCpuSample) -> f64 } } -/// Render the 1m/5m/15m cells, using `-` before a window has data. -fn format_optional_cells(values: [Option; 3]) -> String { +/// Render the 1m/5m/15m cells joined by `sep`, using `-` before a window has +/// data. +fn format_optional_cells(values: [Option; 3], sep: &str) -> String { values .into_iter() .map(|value| { @@ -1167,18 +1171,20 @@ fn format_optional_cells(values: [Option; 3]) -> String { .unwrap_or_else(|| "-".to_string()) }) .collect::>() - .join("/") + .join(sep) } -fn format_latency_cell(latency: Option) -> String { - latency - .map(|latency| { - format!( - "{}/{}/{}/{}", - latency.p50, latency.p95, latency.p99, latency.avg - ) - }) - .unwrap_or_else(|| "-".to_string()) +/// Render p50/p95/p99/avg latency cells joined by `sep`. +/// +/// Missing latency still renders one `-` per cell so positional consumers see +/// a fixed number of fields. +fn format_latency_cell(latency: Option, sep: &str) -> String { + match latency { + Some(latency) => [latency.p50, latency.p95, latency.p99, latency.avg] + .map(|value| value.to_string()) + .join(sep), + None => ["-"; 4].join(sep), + } } #[cfg(test)] @@ -1640,6 +1646,26 @@ mod tests { )) } + /// Test that the per-VM status line has a fixed number of + /// space-separated fields whether or not samples are available. + #[test] + fn test_per_vm_status_line_has_fixed_space_separated_fields() { + let mut status = InstanceStatus::default(); + let empty = Controller::format_instance_status_line(&status, [None; 3]); + status.read_latency_us = Some(crate::instance::SnapshotLatency { + p50: 1, + p95: 2, + p99: 3, + avg: 4, + }); + let full = Controller::format_instance_status_line(&status, [Some(1), Some(2), Some(3)]); + for line in [&empty, &full] { + assert!(!line.contains('/'), "{line}"); + assert_eq!(line.split(' ').count(), 23, "{line}"); + } + assert!(full.contains(" 1 2 3 4 - - - - "), "{full}"); + } + /// Test that discovery sync adds new VMs, retains existing ones, /// and removes disappeared ones. #[tokio::test] diff --git a/src/daemon.rs b/src/daemon.rs index dd46083e..80a7c5a4 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -73,11 +73,12 @@ pub async fn run(cfg: Config, backends: Vec>) -> Result<(), Dae if cfg.print_status_header { tracing::info!( target: "status", - "# per-VM line: status= // \ - // / \ - // \ - /// /// \ - /<5m>/<15m>" + "# per-VM line: status= \ + \ + \ + \ + \ + " ); tracing::info!( target: "status", diff --git a/src/rolling.rs b/src/rolling.rs index 915281f4..b39d5cf3 100644 --- a/src/rolling.rs +++ b/src/rolling.rs @@ -217,8 +217,9 @@ impl RollingMetrics { } } -/// Render 1m/5m/15m cells, using `-` until a window has a sample. -pub fn format_1_5_15(metrics: &RollingMetrics, getter: F) -> String +/// Render 1m/5m/15m cells joined by `sep`, using `-` until a window has a +/// sample. +pub fn format_1_5_15(metrics: &RollingMetrics, getter: F, sep: &str) -> String where F: Fn(&RollingMetrics, Duration) -> Option, { @@ -228,7 +229,7 @@ where .map(|value| value.to_string()) .unwrap_or_else(|| "-".to_string()) }) - .join("/") + .join(sep) } #[cfg(test)] From 5220ef160be997276f0752ec0710b765a737704a Mon Sep 17 00:00:00 2001 From: Thanos Makatos Date: Thu, 22 Oct 2026 13:34:22 +0000 Subject: [PATCH 02/17] fixup! Space-separate every field in per-VM log Signed-off-by: Thanos Makatos --- src/controller.rs | 10 ++++------ src/rolling.rs | 28 ++++++++++++++-------------- 2 files changed, 18 insertions(+), 20 deletions(-) diff --git a/src/controller.rs b/src/controller.rs index 4113c52f..759859e2 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -27,7 +27,7 @@ use crate::{ AppliedOutcome, BlockedReason, EngineTickContext, PsiSample, ScaleAction, ScalingEngine, }, instance::{CgroupError, Instance, InstanceStatus, SnapshotLatency}, - rolling::format_1_5_15, + per_vm_log, state::{StateError, VmOwnership, VmStateStore}, util::Path, }; @@ -988,11 +988,9 @@ impl Controller { cpu.total_pct, format_latency_cell(status.read_latency_us, " "), format_latency_cell(status.write_latency_us, " "), - format_1_5_15( - &status.rolling, - |rolling, window| rolling.cpu_us_per_io_over(window), - " " - ), + &status + .rolling + .format_1_5_15(|rolling, window| rolling.cpu_us_per_io_over(window), " "), ) } diff --git a/src/rolling.rs b/src/rolling.rs index b39d5cf3..eaa350fe 100644 --- a/src/rolling.rs +++ b/src/rolling.rs @@ -215,21 +215,21 @@ impl RollingMetrics { } Some((io_ops, cpu_ns, wall_ns)) } -} -/// Render 1m/5m/15m cells joined by `sep`, using `-` until a window has a -/// sample. -pub fn format_1_5_15(metrics: &RollingMetrics, getter: F, sep: &str) -> String -where - F: Fn(&RollingMetrics, Duration) -> Option, -{ - [1, 5, 15] - .map(|minutes| { - getter(metrics, Duration::from_secs(minutes * 60)) - .map(|value| value.to_string()) - .unwrap_or_else(|| "-".to_string()) - }) - .join(sep) + /// Render 1m/5m/15m cells joined by `sep`, using `-` until a window has a + /// sample. + pub fn format_1_5_15(&self, getter: F, sep: &str) -> String + where + F: Fn(&Self, Duration) -> Option, + { + [1, 5, 15] + .map(|minutes| { + getter(self, Duration::from_secs(minutes * 60)) + .map(|value| value.to_string()) + .unwrap_or_else(|| "-".to_string()) + }) + .join(sep) + } } #[cfg(test)] From c195e1d0d91557f9da159da53eec1468add522ad Mon Sep 17 00:00:00 2001 From: Thanos Makatos Date: Thu, 22 Oct 2026 13:46:36 +0000 Subject: [PATCH 03/17] fixup! Space-separate every field in per-VM log Signed-off-by: Thanos Makatos --- src/controller.rs | 1 - 1 file changed, 1 deletion(-) diff --git a/src/controller.rs b/src/controller.rs index 759859e2..4d740cba 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -27,7 +27,6 @@ use crate::{ AppliedOutcome, BlockedReason, EngineTickContext, PsiSample, ScaleAction, ScalingEngine, }, instance::{CgroupError, Instance, InstanceStatus, SnapshotLatency}, - per_vm_log, state::{StateError, VmOwnership, VmStateStore}, util::Path, }; From 6664078c13fcb0910c86df880c8e17eaeb44bd20 Mon Sep 17 00:00:00 2001 From: Thanos Makatos Date: Thu, 22 Oct 2026 13:55:01 +0000 Subject: [PATCH 04/17] fixup! Space-separate every field in per-VM log Signed-off-by: Thanos Makatos --- src/controller.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/controller.rs b/src/controller.rs index 4d740cba..2727d478 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -987,7 +987,7 @@ impl Controller { cpu.total_pct, format_latency_cell(status.read_latency_us, " "), format_latency_cell(status.write_latency_us, " "), - &status + status .rolling .format_1_5_15(|rolling, window| rolling.cpu_us_per_io_over(window), " "), ) From 12e53be87b498110defb62ea84678dcd67ab6793 Mon Sep 17 00:00:00 2001 From: Thanos Makatos Date: Thu, 22 Oct 2026 09:58:37 +0000 Subject: [PATCH 05/17] Add STYLE.md Signed-off-by: Thanos Makatos --- STYLE.md | 11 +++++++++++ 1 file changed, 11 insertions(+) create mode 100644 STYLE.md diff --git a/STYLE.md b/STYLE.md new file mode 100644 index 00000000..826f81b2 --- /dev/null +++ b/STYLE.md @@ -0,0 +1,11 @@ +# Commits messages + +- Commit title should start with a capital letter. +- If you used AI, state so in the commit message, e.g.: + "AI-generated: everything except commit message". + +# Unit Tests + +- Unit test name should have the `test_` prefix. +- Unit test documentation should explain what this does test tests, e.g.: + "Tests that foo does such and such." From eb940d10646ce66429b8eb63039aca4fdaef20e7 Mon Sep 17 00:00:00 2001 From: Thanos Makatos Date: Thu, 22 Oct 2026 13:37:57 +0000 Subject: [PATCH 06/17] fixup! Add STYLE.md Signed-off-by: Thanos Makatos --- STYLE.md | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/STYLE.md b/STYLE.md index 826f81b2..aaa7e8f0 100644 --- a/STYLE.md +++ b/STYLE.md @@ -1,4 +1,4 @@ -# Commits messages +# Commit messages - Commit title should start with a capital letter. - If you used AI, state so in the commit message, e.g.: From 957f65d3035386a50194e0517342d1b0c0f34aa1 Mon Sep 17 00:00:00 2001 From: Thanos Makatos Date: Thu, 22 Oct 2026 10:29:21 +0000 Subject: [PATCH 07/17] Space-separate every field in aggregate status line The aggregate status line still grouped its 1m, 5m and 15m IOPS with '/'. Separate them with spaces, like the per-VM log. AI-generated: everything Signed-off-by: Thanos Makatos --- README.md | 2 +- src/controller.rs | 2 +- src/daemon.rs | 2 +- 3 files changed, 3 insertions(+), 3 deletions(-) diff --git a/README.md b/README.md index bb679cc7..6db2c05f 100644 --- a/README.md +++ b/README.md @@ -206,7 +206,7 @@ A dash (`-`) means no sample is available for that field. ### Aggregate line ``` -INFO status: 1 4 154995/153840/151220 +INFO status: 1 4 154995 153840 151220 ``` `status:` is the tracing target, which separates status lines from the diff --git a/src/controller.rs b/src/controller.rs index 2727d478..1f2fb8a0 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -963,7 +963,7 @@ impl Controller { "{} {} {}", self.instances.len(), total_threads, - format_optional_cells(aggregate, "/") + format_optional_cells(aggregate, " ") ); } } diff --git a/src/daemon.rs b/src/daemon.rs index 80a7c5a4..ed3ce058 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -82,7 +82,7 @@ pub async fn run(cfg: Config, backends: Vec>) -> Result<(), Dae ); tracing::info!( target: "status", - "# aggregate line: //" + "# aggregate line: " ); } let engine = load_registered_engine(&cfg.engine_config_dir, &cfg.engine)?; From 3c62f8fc5dac6fea30353b6f0c9757ed7b639d9d Mon Sep 17 00:00:00 2001 From: Thanos Makatos Date: Wed, 21 Oct 2026 21:41:18 +0000 Subject: [PATCH 08/17] Log VM appearing and leaving in per-VM log Otherwise there's no way to tell from the per-VM log when the daemon started or stopped tracking a VM, or why it stopped. AI-generated: everything Signed-off-by: Thanos Makatos --- README.md | 4 ++++ src/controller.rs | 28 ++++++++++++++++++++++++++++ 2 files changed, 32 insertions(+) diff --git a/README.md b/README.md index 6db2c05f..b7bd1de1 100644 --- a/README.md +++ b/README.md @@ -203,6 +203,10 @@ from `event=` records, and the fields are: A dash (`-`) means no sample is available for that field. +The per-VM log also records `event=vm_added` when the daemon starts tracking +the VM and `event=vm_removed reason=` when it +stops. + ### Aggregate line ``` diff --git a/src/controller.rs b/src/controller.rs index 1f2fb8a0..327404b1 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -267,6 +267,7 @@ impl Controller { status.ownership_classification = classification; drop(status); self.engine.on_instance_added(&instance).await; + self.log_per_vm_line(&instance.id, "event=vm_added"); self.instances.insert(instance.id.clone(), instance); added += 1; } @@ -282,6 +283,7 @@ impl Controller { if let Some(instance) = self.instances.remove(&id) { instance.client.close().await; self.engine.on_instance_removed(&id).await; + self.log_per_vm_line(&id, "event=vm_removed reason=not_discovered"); } } Ok((added, removed)) @@ -571,6 +573,7 @@ impl Controller { { self.instances.remove(&id); self.engine.on_instance_removed(&id).await; + self.log_per_vm_line(&id, "event=vm_removed reason=refresh_failed"); // FIXME IIUC a failed instance gets dropped but we don't call // close() } @@ -1684,6 +1687,8 @@ mod tests { }), ) .unwrap(); + let log_dir = tempfile::tempdir().unwrap(); + controller.per_vm_log_dir = Path::new(log_dir.path()); let closed = Arc::new(AtomicUsize::new(0)); let first = instance("vm-a", 1, Arc::clone(&closed)); @@ -1706,6 +1711,22 @@ mod tests { assert!(controller.instances.is_empty()); assert_eq!(removed.load(Ordering::Relaxed), 1); assert_eq!(evaluated.load(Ordering::Relaxed), 0); + assert_eq!( + per_vm_log_events(&controller, "vm-a"), + ["event=vm_added", "event=vm_removed reason=not_discovered"] + ); + } + + /// Return the `event=` records of one VM's per-VM log, without + /// timestamps. + fn per_vm_log_events(controller: &Controller, vm: &str) -> Vec { + std::fs::read_to_string(controller.per_vm_log_path(vm)) + .unwrap() + .lines() + .filter_map(|line| line.split_once(' ').map(|(_, record)| record)) + .filter(|record| record.starts_with("event=")) + .map(str::to_string) + .collect() } /// Test that one tick refreshes state, runs the engine, and drops @@ -1729,6 +1750,8 @@ mod tests { }), ) .unwrap(); + let log_dir = tempfile::tempdir().unwrap(); + controller.per_vm_log_dir = Path::new(log_dir.path()); let closed = Arc::new(AtomicUsize::new(0)); let ok = instance("ok", 4, Arc::clone(&closed)); @@ -1751,5 +1774,10 @@ mod tests { ); assert_eq!(evaluated.load(Ordering::Relaxed), 1); assert_eq!(removed.load(Ordering::Relaxed), 1); + assert_eq!(per_vm_log_events(&controller, "ok"), ["event=vm_added"]); + assert_eq!( + per_vm_log_events(&controller, "bad"), + ["event=vm_added", "event=vm_removed reason=refresh_failed"] + ); } } From 1523c2a88d658d31a0144d7a65642f534ac86ffe Mon Sep 17 00:00:00 2001 From: Thanos Makatos Date: Wed, 21 Oct 2026 21:41:58 +0000 Subject: [PATCH 09/17] Don't log per-tick VM refresh spans at INFO refresh_state and read_cpu_sample run for every VM on every tick, and as INFO spans they print a 'new' line each time. Demote them to DEBUG and attach the VM ID and PID directly to the warnings they emit, which previously relied on the span for that context. AI-generated: everything Signed-off-by: Thanos Makatos --- src/instance.rs | 8 ++++++-- 1 file changed, 6 insertions(+), 2 deletions(-) diff --git a/src/instance.rs b/src/instance.rs index 8ce74b2f..bfcee026 100644 --- a/src/instance.rs +++ b/src/instance.rs @@ -203,7 +203,7 @@ impl Instance { } /// Refresh one VM and mark its client broken on failure. - #[tracing::instrument(skip(self), fields(id = %self.id))] + #[tracing::instrument(level = "debug", skip(self), fields(id = %self.id))] pub async fn refresh_state(&self, refresh_cgroup: bool) -> bool { match self.client.get_thread_pool_snapshot().await { Ok(snapshot) => { @@ -213,6 +213,7 @@ impl Instance { Err(error) => { tracing::warn!( target: "controller", + vm = %self.id, %error, "refresh failed; removing instance" ); @@ -235,6 +236,7 @@ impl Instance { Err(error) => { tracing::warn!( target: "controller", + vm = %self.id, %error, "failed to sample backend task CPU" ); @@ -640,7 +642,7 @@ fn compute_per_worker_util( } /// Read cumulative CPU time across matching `/proc//task/*/stat` files. -#[tracing::instrument(skip(filter), fields(pid))] +#[tracing::instrument(level = "debug", skip(filter))] fn read_cpu_sample( pid: i32, filter: &crate::instance::ThreadNameFilter, @@ -656,6 +658,7 @@ fn read_cpu_sample( Err(error) => { tracing::warn!( target: "controller", + pid, %error, "failed to enumerate backend task" ); @@ -667,6 +670,7 @@ fn read_cpu_sample( Err(error) => { tracing::warn!( target: "controller", + pid, %error, "failed to read backend task stat" ); From 1ba0f130f09c9742d39efc601e97bacb70bb56f6 Mon Sep 17 00:00:00 2001 From: Thanos Makatos Date: Thu, 22 Oct 2026 13:42:49 +0000 Subject: [PATCH 10/17] fixup! Don't log per-tick VM refresh spans at INFO Signed-off-by: Thanos Makatos --- src/instance.rs | 2 +- 1 file changed, 1 insertion(+), 1 deletion(-) diff --git a/src/instance.rs b/src/instance.rs index bfcee026..9eed3dc0 100644 --- a/src/instance.rs +++ b/src/instance.rs @@ -203,7 +203,7 @@ impl Instance { } /// Refresh one VM and mark its client broken on failure. - #[tracing::instrument(level = "debug", skip(self), fields(id = %self.id))] + #[tracing::instrument(level = "trace", skip(self), fields(id = %self.id))] pub async fn refresh_state(&self, refresh_cgroup: bool) -> bool { match self.client.get_thread_pool_snapshot().await { Ok(snapshot) => { From 90a90ce5b9278149b2eff4aad63f66afc97d8180 Mon Sep 17 00:00:00 2001 From: Thanos Makatos Date: Wed, 21 Oct 2026 21:42:33 +0000 Subject: [PATCH 11/17] Don't log span creation in systemd log style The systemd log style printed a line for every INFO span created, including zbus's internal spans, which flooded the log with D-Bus handshake details on start up. AI-generated: everything Signed-off-by: Thanos Makatos --- src/main.rs | 2 -- 1 file changed, 2 deletions(-) diff --git a/src/main.rs b/src/main.rs index d276d99e..06aaccb9 100644 --- a/src/main.rs +++ b/src/main.rs @@ -17,7 +17,6 @@ use io_thread_controller::{ }; use thiserror::Error; use tracing::{debug, info, instrument}; -use tracing_subscriber::fmt::format::FmtSpan; #[derive(Error, Debug)] enum IoThreadControllerError { @@ -189,7 +188,6 @@ fn init_logging(filter: &str, style: LogStyle) { .with_target(true) .without_time() .with_env_filter(env_filter) - .with_span_events(FmtSpan::NEW) // FIXME elsewhere .finish(); let _ = tracing::subscriber::set_global_default(subscriber); } From 4a2a75a507195e2cd554f659648b56635f79e2f7 Mon Sep 17 00:00:00 2001 From: Thanos Makatos Date: Wed, 21 Oct 2026 21:44:47 +0000 Subject: [PATCH 12/17] Don't reconnect to known control sockets on every discovery Discovery connected to every VM's control socket on every pass, only to read the peer PID, even for VMs already tracked. Backends that listen with a minimal backlog reject such a connection with EAGAIN while an earlier one is still pending, which logged a warning and dropped the VM from that discovery pass. Cache the peer PID per socket path and reconnect only when the socket file is replaced or the cached process is gone. EAGAIN on a socket that isn't cached yet is transient, so log it at debug and retry on the next pass. AI-generated: everything Signed-off-by: Thanos Makatos --- src/backends/mod.rs | 162 +++++++++++++++++++++++++++++++++++++++++++- 1 file changed, 159 insertions(+), 3 deletions(-) diff --git a/src/backends/mod.rs b/src/backends/mod.rs index c63f6fce..14547124 100644 --- a/src/backends/mod.rs +++ b/src/backends/mod.rs @@ -140,7 +140,13 @@ fn is_peer_disconnect_kind(kind: ErrorKind) -> bool { ) } -use std::{io, sync::Arc}; +use std::{ + collections::{HashMap, HashSet}, + io, + os::unix::fs::MetadataExt, + path::PathBuf, + sync::{Arc, LazyLock, Mutex}, +}; #[cfg(feature = "qemu-backend")] pub mod qemu; @@ -240,6 +246,7 @@ where } }; let mut instances = Vec::new(); + let mut seen = HashSet::new(); while let Some(entry) = entries.next_entry().await? { match entry.metadata().await { Ok(metadata) => { @@ -256,9 +263,16 @@ where continue; }; let sock_path = Path::new(&socket_dir.join(entry.path().join(socket_name))); - // FIXME add unit test for this function - let pid = match control_socket_peer_pid(&sock_path).await { + let pid = match cached_control_socket_peer_pid(&sock_path).await { Ok(pid) => pid, + Err(error) if error.kind() == io::ErrorKind::WouldBlock => { + tracing::debug!( + target: "controller", + vm = %id, + "control socket busy; retrying on next discovery" + ); + continue; + } Err(error) => { if error.kind() != io::ErrorKind::NotFound { // FIXME ignore ENOENT @@ -272,12 +286,74 @@ where continue; } }; + seen.insert(sock_path.to_path_buf()); let client = make_client(sock_path.clone()); instances.push(Arc::new(Instance::new(id, sock_path, pid, client))); } + PEER_CACHE + .lock() + .expect("lock poisoned") + .retain(|path, _| !path.starts_with(socket_dir) || seen.contains(path)); Ok(instances) } +/// Peer process last identified behind one control socket. +#[derive(Debug, Clone, Copy)] +struct CachedPeer { + /// Socket file device; a restarted backend binds a new socket file. + dev: u64, + /// Socket file inode. + ino: u64, + /// Peer process ID. + pid: i32, + /// Peer process start time, guarding against PID reuse. + start_time: u64, +} + +/// Control-socket peers identified by earlier discovery passes, keyed by +/// socket path. +/// +/// Backends may listen with a minimal backlog, in which case connecting on +/// every pass just to read the peer credentials fails with `EAGAIN` whenever +/// an earlier connection has not been accepted yet. +static PEER_CACHE: LazyLock>> = LazyLock::new(Default::default); + +/// Return the control-socket peer PID, connecting only when the socket file +/// or the previously identified peer process changed. +async fn cached_control_socket_peer_pid(sock_path: &Path) -> io::Result { + let key: &std::path::Path = sock_path; + let metadata = tokio::fs::metadata(sock_path).await?; + let cached = PEER_CACHE.lock().expect("lock poisoned").get(key).copied(); + if let Some(peer) = cached + && peer.dev == metadata.dev() + && peer.ino == metadata.ino() + && process_start_time(peer.pid) == Some(peer.start_time) + { + return Ok(peer.pid); + } + let pid = control_socket_peer_pid(sock_path).await?; + if let Some(start_time) = process_start_time(pid) { + PEER_CACHE.lock().expect("lock poisoned").insert( + key.to_path_buf(), + CachedPeer { + dev: metadata.dev(), + ino: metadata.ino(), + pid, + start_time, + }, + ); + } + Ok(pid) +} + +/// Return the start time of `pid`, or `None` if it no longer exists. +fn process_start_time(pid: i32) -> Option { + procfs::process::Process::new(pid) + .and_then(|process| process.stat()) + .map(|stat| stat.starttime) + .ok() +} + /// Return the process ID at the far end of a UNIX control socket. async fn control_socket_peer_pid(sock_path: &Path) -> io::Result { let stream = tokio::net::UnixStream::connect(sock_path).await?; @@ -288,3 +364,83 @@ async fn control_socket_peer_pid(sock_path: &Path) -> io::Result { ) }) } + +#[cfg(test)] +mod tests { + use std::os::unix::net::UnixListener; + + use async_trait::async_trait; + + use super::*; + use crate::instance::ThreadPoolSnapshot; + + struct NullClient; + + #[async_trait] + impl InstanceClient for NullClient { + async fn set_thread_count(&self, _count: u32) -> Result<(), BackendClientError> { + Ok(()) + } + + async fn get_thread_pool_snapshot(&self) -> Result { + Err(BackendClientError::NotSupported("snapshot".to_string())) + } + + async fn close(&self) {} + } + + async fn discover(socket_dir: &std::path::Path) -> Vec> { + discover_via_control_socket(Path::new(socket_dir), "ctl", |_| { + Box::new(NullClient) as Box + }) + .await + .unwrap() + } + + fn bind(path: &std::path::Path) -> UnixListener { + let listener = UnixListener::bind(path).unwrap(); + listener.set_nonblocking(true).unwrap(); + listener + } + + /// Return how many connections are pending on `listener`. + fn pending(listener: &UnixListener) -> usize { + std::iter::from_fn(|| listener.accept().ok()).count() + } + + /// Test that discovery identifies a control-socket peer once and does + /// not reconnect on later passes until the socket is rebound. + #[tokio::test] + async fn test_discovery_reuses_identified_control_socket_peer() { + let socket_dir = tempfile::tempdir().unwrap(); + std::fs::create_dir(socket_dir.path().join("vm-a")).unwrap(); + let sock_path = socket_dir.path().join("vm-a").join("ctl"); + let listener = bind(&sock_path); + + let instances = discover(socket_dir.path()).await; + assert_eq!(instances.len(), 1); + assert_eq!(instances[0].id, "vm-a"); + assert_eq!(instances[0].pid, std::process::id() as i32); + assert_eq!(pending(&listener), 1); + + assert_eq!(discover(socket_dir.path()).await.len(), 1); + assert_eq!(pending(&listener), 0); + + drop(listener); + std::fs::remove_file(&sock_path).unwrap(); + let listener = bind(&sock_path); + assert_eq!(discover(socket_dir.path()).await.len(), 1); + assert_eq!(pending(&listener), 1); + + drop(listener); + std::fs::remove_file(&sock_path).unwrap(); + assert!(discover(socket_dir.path()).await.is_empty()); + assert!( + !PEER_CACHE + .lock() + .unwrap() + .keys() + .any(|path| path.starts_with(socket_dir.path())) + ); + } +} From 758c8f5be933057eea68f78ff2888b7c08bee747 Mon Sep 17 00:00:00 2001 From: Thanos Makatos Date: Thu, 22 Oct 2026 16:42:09 +0000 Subject: [PATCH 13/17] fixup! Don't reconnect to known control sockets on every discovery Signed-off-by: Thanos Makatos --- src/backends/mod.rs | 8 +++++--- 1 file changed, 5 insertions(+), 3 deletions(-) diff --git a/src/backends/mod.rs b/src/backends/mod.rs index 14547124..34c3f9de 100644 --- a/src/backends/mod.rs +++ b/src/backends/mod.rs @@ -415,7 +415,7 @@ mod tests { let socket_dir = tempfile::tempdir().unwrap(); std::fs::create_dir(socket_dir.path().join("vm-a")).unwrap(); let sock_path = socket_dir.path().join("vm-a").join("ctl"); - let listener = bind(&sock_path); + let mut listener = bind(&sock_path); let instances = discover(socket_dir.path()).await; assert_eq!(instances.len(), 1); @@ -426,9 +426,11 @@ mod tests { assert_eq!(discover(socket_dir.path()).await.len(), 1); assert_eq!(pending(&listener), 0); - drop(listener); + // Keep the old socket inode alive while rebinding so the filesystem + // cannot hand the same inode number to the new socket. std::fs::remove_file(&sock_path).unwrap(); - let listener = bind(&sock_path); + let old_listener = std::mem::replace(&mut listener, bind(&sock_path)); + drop(old_listener); assert_eq!(discover(socket_dir.path()).await.len(), 1); assert_eq!(pending(&listener), 1); From 387cdae423683ff963d2ebd83d1d56e345420498 Mon Sep 17 00:00:00 2001 From: Thanos Makatos Date: Wed, 21 Oct 2026 22:05:43 +0000 Subject: [PATCH 14/17] Log engine scaling utilisation as a percentage The threshold engine logged per-thread utilisation as a raw fraction with full f64 precision (e.g. util=0.9932802709036076). Log it as a percentage with two decimals instead (e.g. util=99.33%). AI-generated: everything Signed-off-by: Thanos Makatos --- README.md | 4 ++-- src/engines/threshold.rs | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/README.md b/README.md index b7bd1de1..0329432a 100644 --- a/README.md +++ b/README.md @@ -231,11 +231,11 @@ you get one INFO line on the `engine` target using the same `vm=` convention as the status lines: ``` -INFO vm=vm-1615a59c-... util=0.8955 action="up" thr=4->5 +INFO vm=vm-1615a59c-... util=89.55% action="up" thr=4->5 ``` `vm` is the same identifier used by status lines, `util` is -the per-thread utilisation that triggered the decision, `action` +the per-thread utilisation percentage that triggered the decision, `action` is `up`, `down`, or `revert`, and `thr` is the requested `->` transition. diff --git a/src/engines/threshold.rs b/src/engines/threshold.rs index 65d358ae..045bc2f8 100644 --- a/src/engines/threshold.rs +++ b/src/engines/threshold.rs @@ -586,7 +586,7 @@ fn log_scale_up_decision( target: "engine", event = "scale_up", vm = instance.to_string(), - util = per_thread_util, + util = %format!("{:.2}%", per_thread_util * 100.0), thr = %format!("{thread_count}->{target}"), current_iops, min_target_iops, @@ -606,7 +606,7 @@ fn log_scale_down_decision( target: "engine", event = "scale_down", vm = instance.to_string(), - util = per_thread_util, + util = %format!("{:.2}%", per_thread_util * 100.0), thr = %format!("{thread_count}->{target}"), "threshold scaling decision" ); From fc6f4426c38ac990a321fac1b3ee95cfccdc2085 Mon Sep 17 00:00:00 2001 From: Thanos Makatos Date: Thu, 22 Oct 2026 11:03:33 +0000 Subject: [PATCH 15/17] Write per-VM log through a tracing layer Every per-VM event was logged twice, once through tracing and once as a hand-formatted line for the per-VM file, and events logged elsewhere, such as the threshold engine's scaling decisions, never reached the per-VM file at all. Add a tracing layer that copies every event carrying a vm field to that VM's file, without the vm field, and log per-VM events only once. Status and lifecycle records use a dedicated per_vm target that the main log filters out, so they still only go to the per-VM files. The layer has its own INFO filter, so RUST_LOG no longer affects the per-VM files. Use vm consistently for the VM ID field, instead of id or uuid, so that those events are routed too. AI-generated: everything Signed-off-by: Thanos Makatos --- README.md | 4 +- src/backends/qemu/client.rs | 12 +- src/backends/qemu/libvirt.rs | 2 +- src/backends/qemu/mod.rs | 4 +- src/controller.rs | 255 +++++++++++----------------------- src/engines/threshold.rs | 2 +- src/lib.rs | 1 + src/main.rs | 58 +++----- src/per_vm_log.rs | 262 +++++++++++++++++++++++++++++++++++ 9 files changed, 377 insertions(+), 223 deletions(-) create mode 100644 src/per_vm_log.rs diff --git a/README.md b/README.md index 0329432a..e4ece764 100644 --- a/README.md +++ b/README.md @@ -205,7 +205,9 @@ A dash (`-`) means no sample is available for that field. The per-VM log also records `event=vm_added` when the daemon starts tracking the VM and `event=vm_removed reason=` when it -stops. +stops. Every other log event that names the VM in a `vm` field, such as scaling +decisions, blocked or failed scale operations, and backend errors, is copied to +the VM's file as its message followed by its remaining `key=value` fields. ### Aggregate line diff --git a/src/backends/qemu/client.rs b/src/backends/qemu/client.rs index cd72439a..5dfb862b 100644 --- a/src/backends/qemu/client.rs +++ b/src/backends/qemu/client.rs @@ -103,7 +103,7 @@ impl QemuInstanceClient { if let Err(del_err) = self.client.del_io_thread(&new_id).await { tracing::debug!( target: "qemu", - uuid = %self.uuid, + vm = %self.uuid, iothread = %new_id, "rollback object-del after set-mapping failure: {del_err}" ); @@ -133,7 +133,7 @@ impl QemuInstanceClient { if let Err(e) = self.client.del_io_thread(&doomed).await { tracing::warn!( target: "qemu", - uuid = %self.uuid, + vm = %self.uuid, iothread = %doomed, error = ?e, "object-del failed; will retry next tick" @@ -159,7 +159,7 @@ impl QemuInstanceClient { if thread_count == 1 { tracing::debug!( target: "qemu", - uuid = %self.uuid, + vm = %self.uuid, "using minimal fallback snapshot without cached topology" ); } @@ -170,7 +170,7 @@ impl QemuInstanceClient { Err(e) => { tracing::debug!( target: "qemu", - uuid = %self.uuid, + vm = %self.uuid, error = ?e, "query-blockstats failed" ); @@ -239,7 +239,7 @@ impl InstanceClient for QemuInstanceClient { if should_warn { tracing::warn!( target: "qemu", - uuid = %self.uuid, + vm = %self.uuid, "QMP command x-query-prometheus-metrics unsupported; using fallback snapshot" ); } @@ -305,7 +305,7 @@ impl InstanceClient for QemuInstanceClient { Err(e) => { tracing::debug!( target: "qemu", - uuid = %self.uuid, + vm = %self.uuid, error = ?e, "query-blockstats failed" ); diff --git a/src/backends/qemu/libvirt.rs b/src/backends/qemu/libvirt.rs index 4f054cf7..99d8c292 100644 --- a/src/backends/qemu/libvirt.rs +++ b/src/backends/qemu/libvirt.rs @@ -108,7 +108,7 @@ impl LibvirtConn { Err(e) => { tracing::debug!( target: "qemu", - uuid = %uuid, + vm = %uuid, error = ?e, "libvirt: skip vm (xml fetch)" ); diff --git a/src/backends/qemu/mod.rs b/src/backends/qemu/mod.rs index 51265dc0..a840953a 100644 --- a/src/backends/qemu/mod.rs +++ b/src/backends/qemu/mod.rs @@ -186,7 +186,7 @@ impl Backend for QemuBackend { Ok(_) => { tracing::warn!( target: "qemu", - %uuid, + vm = %uuid, "libvirt reported zero maximum vCPUs; skipping VM" ); continue; @@ -194,7 +194,7 @@ impl Backend for QemuBackend { Err(error) => { tracing::warn!( target: "qemu", - %uuid, + vm = %uuid, %error, "failed to read maximum vCPU count; skipping VM" ); diff --git a/src/controller.rs b/src/controller.rs index 327404b1..920a1771 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -7,10 +7,8 @@ use std::{ collections::{HashMap, HashSet}, - fs::{OpenOptions, create_dir_all}, - io::Write, sync::Arc, - time::{Duration, Instant, SystemTime}, + time::{Duration, Instant}, }; use futures_util::future::join_all; @@ -28,7 +26,6 @@ use crate::{ }, instance::{CgroupError, Instance, InstanceStatus, SnapshotLatency}, state::{StateError, VmOwnership, VmStateStore}, - util::Path, }; /// Wire-facing container for the D-Bus `GetSnapshot` reply. #[derive(Debug, Clone, Default, Serialize, Deserialize)] @@ -182,7 +179,6 @@ const fn from_mins(minutes: u64) -> Duration { /// Rolling windows displayed in uptime-style status fields. const STATUS_WINDOWS: [Duration; 3] = [from_mins(1), from_mins(5), from_mins(15)]; -const PER_VM_LOG_DIR: &str = "/var/log/io-thread-controller/vms"; /// Host-wide cumulative CPU counters from `/proc/stat`. #[derive(Debug, Default, Clone, Copy)] @@ -211,10 +207,6 @@ pub struct Controller { /// and decide. // FIXME Instance probably doesn't need to be an Arc. pub instances: HashMap>, - /// Per-VM log file destination when per-VM main log is disabled. - per_vm_log_dir: Path, - /// VM IDs for which per-VM log I/O failures were already reported. - per_vm_log_io_error_warned: std::sync::Mutex>, /// Monotonically increasing tick sequence. pub tick_index: u64, /// Previous host CPU sample. @@ -236,8 +228,6 @@ impl Controller { vm_state_store, vm_ownership, instances: HashMap::new(), - per_vm_log_dir: Path::from(PER_VM_LOG_DIR), - per_vm_log_io_error_warned: std::sync::Mutex::new(HashSet::new()), tick_index: 0, last_host_cpu: None, host_cpu_util: 0.0, @@ -267,7 +257,7 @@ impl Controller { status.ownership_classification = classification; drop(status); self.engine.on_instance_added(&instance).await; - self.log_per_vm_line(&instance.id, "event=vm_added"); + tracing::info!(target: per_vm_log::TARGET, vm = %instance.id, event = "vm_added"); self.instances.insert(instance.id.clone(), instance); added += 1; } @@ -283,7 +273,12 @@ impl Controller { if let Some(instance) = self.instances.remove(&id) { instance.client.close().await; self.engine.on_instance_removed(&id).await; - self.log_per_vm_line(&id, "event=vm_removed reason=not_discovered"); + tracing::info!( + target: per_vm_log::TARGET, + vm = %id, + event = "vm_removed", + reason = "not_discovered" + ); } } Ok((added, removed)) @@ -573,7 +568,12 @@ impl Controller { { self.instances.remove(&id); self.engine.on_instance_removed(&id).await; - self.log_per_vm_line(&id, "event=vm_removed reason=refresh_failed"); + tracing::info!( + target: per_vm_log::TARGET, + vm = %id, + event = "vm_removed", + reason = "refresh_failed" + ); // FIXME IIUC a failed instance gets dropped but we don't call // close() } @@ -688,10 +688,6 @@ impl Controller { if sticky { self.report_blocked_scale(&instance.id, action, BlockedReason::ManualOverride) .await; - self.log_per_vm_line( - &instance.id, - &format!("event=scale_blocked reason=manual_override action={action}"), - ); tracing::info!( target: "controller", event = "scale_blocked", @@ -705,10 +701,6 @@ impl Controller { // FIXME why would an engine even consider an unmanaged VM? self.report_blocked_scale(&instance.id, action, BlockedReason::UnmanagedVm) .await; - self.log_per_vm_line( - &instance.id, - &format!("event=scale_blocked reason=unmanaged_vm action={action}"), - ); tracing::info!( target: "controller", event = "scale_blocked", @@ -721,16 +713,12 @@ impl Controller { if target < self.cfg.min_thread_count { self.report_blocked_scale(instance_id, action, BlockedReason::TargetBelowMinimum) .await; - self.log_per_vm_line( - &instance.id, - &format!( - "event=scale_blocked reason=target_below_minimum action={action} target={} minimum={}", - target, self.cfg.min_thread_count - ), - ); tracing::info!( target: "controller", - id = %instance.id, + event = "scale_blocked", + reason = "target_below_minimum", + vm = %instance.id, + action = %action, target, minimum = self.cfg.min_thread_count, "automatic scaling suppressed by controller minimum" @@ -740,16 +728,12 @@ impl Controller { if target > self.cfg.max_thread_count { self.report_blocked_scale(instance_id, action, BlockedReason::TargetExceedsMaximum) .await; - self.log_per_vm_line( - &instance.id, - &format!( - "event=scale_blocked reason=target_exceeds_maximum action={action} target={} maximum={}", - target, self.cfg.max_thread_count - ), - ); tracing::info!( target: "controller", - id = %instance.id, + event = "scale_blocked", + reason = "target_exceeds_maximum", + vm = %instance.id, + action = %action, target, maximum = self.cfg.max_thread_count, "automatic scaling suppressed by controller maximum" @@ -759,16 +743,12 @@ impl Controller { if target > vcpu_count { self.report_blocked_scale(instance_id, action, BlockedReason::TargetExceedsVcpuCount) .await; - self.log_per_vm_line( - &instance.id, - &format!( - "event=scale_blocked reason=target_exceeds_vcpu_count action={action} target={} vcpus={}", - target, vcpu_count - ), - ); tracing::info!( target: "controller", - id = %instance.id, + event = "scale_blocked", + reason = "target_exceeds_vcpu_count", + vm = %instance.id, + action = %action, target, vcpus = vcpu_count, "automatic scaling suppressed by vCPU cap" @@ -780,13 +760,12 @@ impl Controller { { self.report_blocked_scale(instance_id, action, BlockedReason::Cooldown) .await; - self.log_per_vm_line( - &instance.id, - &format!("event=scale_blocked reason=cooldown action={action}"), - ); tracing::info!( target: "controller", - id = %instance.id, + event = "scale_blocked", + reason = "cooldown", + vm = %instance.id, + action = %action, "automatic scaling suppressed by cooldown" ); return Ok(false); @@ -801,16 +780,12 @@ impl Controller { { self.report_blocked_scale(instance_id, action, BlockedReason::HostCpuCeiling) .await; - self.log_per_vm_line( - &instance.id, - &format!( - "event=scale_blocked reason=host_cpu_ceiling action={action} host_cpu={} ceiling={}", - self.host_cpu_util, self.cfg.host_cpu_scale_up_ceiling - ), - ); tracing::info!( target: "controller", - id = %instance.id, + event = "scale_blocked", + reason = "host_cpu_ceiling", + vm = %instance.id, + action = %action, host_cpu = self.host_cpu_util, ceiling = self.cfg.host_cpu_scale_up_ceiling, "automatic scaling suppressed by host CPU guard" @@ -819,12 +794,9 @@ impl Controller { } if self.cfg.dry_run { - self.log_per_vm_line( - &instance.id, - &format!("event=scale_dry_run action={action} target={target}"), - ); tracing::info!( target: "controller", + event = "scale_dry_run", vm = %instance.id, action = %action, target, @@ -885,13 +857,6 @@ impl Controller { prev_thread_count = previous_count, prev_io_count_total = previous_io_count ); - self.log_per_vm_line( - &instance.id, - &format!( - "event=scale_applied action={action} target={} prev_thread_count={} prev_io_count_total={}", - target, previous_count, previous_io_count - ), - ); Ok(true) } Err(error) => { @@ -907,13 +872,6 @@ impl Controller { target, error = %error_text ); - self.log_per_vm_line( - &instance.id, - &format!( - "event=scale_failed action={action} target={} error={}", - target, error_text - ), - ); Ok(false) } } @@ -950,7 +908,7 @@ impl Controller { if self.cfg.enable_per_vm_status_line && self.cfg.enable_per_vm_main_log { Self::emit_instance_status_main(instance, &status, iops_windows); } else if self.cfg.enable_per_vm_status_line { - self.emit_instance_status_file(instance, &status, iops_windows); + Self::emit_instance_status_file(instance, &status, iops_windows); } if self.cfg.enable_per_vm_main_log { Self::emit_instance_status_main(instance, &status, iops_windows); @@ -997,14 +955,15 @@ impl Controller { } fn emit_instance_status_file( - &self, instance: &Instance, status: &InstanceStatus, iops_windows: [Option; 3], ) { - self.log_per_vm_line( - &instance.id, - &Self::format_instance_status_line(status, iops_windows), + tracing::info!( + target: per_vm_log::TARGET, + vm = %instance.id, + "{}", + Self::format_instance_status_line(status, iops_windows) ); } @@ -1029,88 +988,6 @@ impl Controller { write_latency_us = %format_latency_cell(status.write_latency_us, "/") ); } - - fn should_log_per_vm_file(&self) -> bool { - self.cfg.enable_per_vm_status_line && !self.cfg.enable_per_vm_main_log - } - - fn per_vm_log_path(&self, vm: &str) -> Path { - let name: String = vm - .chars() - .map(|ch| { - if ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.') { - ch - } else { - '_' - } - }) - .collect(); - Path::new(&self.per_vm_log_dir.join(format!("{name}.log"))) - } - - fn log_per_vm_line(&self, vm: &str, line: &str) { - if !self.should_log_per_vm_file() { - return; - } - let path = self.per_vm_log_path(vm); - let path_text = path.display().to_string(); - if let Some(parent) = path.parent() - && let Err(error) = create_dir_all(parent) - { - let mut warned = self - .per_vm_log_io_error_warned - .lock() - .expect("lock poisoned"); - if warned.insert(vm.to_string()) { - tracing::warn!( - target: "controller", - vm, - path = %path_text, - %error, - "failed to create per-VM log directory" - ); - } - return; - } - match OpenOptions::new().create(true).append(true).open(&path) { - Ok(mut file) => { - if let Err(error) = writeln!( - file, - "{} {line}", - humantime::format_rfc3339_millis(SystemTime::now()) - ) { - let mut warned = self - .per_vm_log_io_error_warned - .lock() - .expect("lock poisoned"); - if warned.insert(vm.to_string()) { - tracing::warn!( - target: "controller", - vm, - path = %path_text, - %error, - "failed to write per-VM log" - ); - } - } - } - Err(error) => { - let mut warned = self - .per_vm_log_io_error_warned - .lock() - .expect("lock poisoned"); - if warned.insert(vm.to_string()) { - tracing::warn!( - target: "controller", - vm, - path = %path_text, - %error, - "failed to open per-VM log" - ); - } - } - } - } } /// Read CPU, I/O, and memory pressure; unavailable values become zero. @@ -1202,8 +1079,10 @@ mod tests { backends::BackendClientError, engines::{EngineTickContext, ScalingEngine}, instance::{InstanceClient, ThreadPoolSnapshot}, + per_vm_log::PerVmLog, util::Path, }; + use tracing_subscriber::layer::SubscriberExt; #[cfg(feature = "threshold-engine")] use crate::engines::threshold::{ThresholdConfig, ThresholdEngine}; @@ -1265,6 +1144,7 @@ mod tests { controller .instances .insert(instance.id.clone(), Arc::clone(&instance)); + let (per_vm_log, _log_dir, _guard) = capture_per_vm_log(); controller .apply_engine_decision(&instance.id, ScaleAction::Up(4)) @@ -1294,6 +1174,16 @@ mod tests { .await .unwrap(); assert_eq!(target.load(Ordering::Relaxed), 3); + + let events = per_vm_log_events(&per_vm_log, &instance.id); + assert!(events[0].starts_with("event=scale_applied "), "{events:?}"); + assert!( + events[1].starts_with( + "automatic scaling suppressed by vCPU cap event=scale_blocked \ + reason=target_exceeds_vcpu_count " + ) && events[1].ends_with(" target=5 vcpus=4"), + "{events:?}" + ); } // FIXME This test seems like it mixes basic ScaleAction unit tests and more @@ -1687,8 +1577,7 @@ mod tests { }), ) .unwrap(); - let log_dir = tempfile::tempdir().unwrap(); - controller.per_vm_log_dir = Path::new(log_dir.path()); + let (per_vm_log, _log_dir, _guard) = capture_per_vm_log(); let closed = Arc::new(AtomicUsize::new(0)); let first = instance("vm-a", 1, Arc::clone(&closed)); @@ -1712,19 +1601,34 @@ mod tests { assert_eq!(removed.load(Ordering::Relaxed), 1); assert_eq!(evaluated.load(Ordering::Relaxed), 0); assert_eq!( - per_vm_log_events(&controller, "vm-a"), + per_vm_log_events(&per_vm_log, "vm-a"), ["event=vm_added", "event=vm_removed reason=not_discovered"] ); } - /// Return the `event=` records of one VM's per-VM log, without - /// timestamps. - fn per_vm_log_events(controller: &Controller, vm: &str) -> Vec { - std::fs::read_to_string(controller.per_vm_log_path(vm)) + /// Route this thread's per-VM events to files in a temporary directory. + fn capture_per_vm_log() -> ( + PerVmLog, + tempfile::TempDir, + tracing::subscriber::DefaultGuard, + ) { + let dir = tempfile::tempdir().unwrap(); + let per_vm_log = PerVmLog::default(); + per_vm_log.set_dir(Some(Path::new(dir.path()))); + let guard = tracing::subscriber::set_default( + tracing_subscriber::registry().with(per_vm_log.clone()), + ); + (per_vm_log, dir, guard) + } + + /// Return the records of one VM's per-VM log that carry an `event` + /// field, without timestamps. + fn per_vm_log_events(per_vm_log: &PerVmLog, vm: &str) -> Vec { + std::fs::read_to_string(per_vm_log.path(vm).unwrap()) .unwrap() .lines() .filter_map(|line| line.split_once(' ').map(|(_, record)| record)) - .filter(|record| record.starts_with("event=")) + .filter(|record| record.contains("event=")) .map(str::to_string) .collect() } @@ -1750,8 +1654,7 @@ mod tests { }), ) .unwrap(); - let log_dir = tempfile::tempdir().unwrap(); - controller.per_vm_log_dir = Path::new(log_dir.path()); + let (per_vm_log, _log_dir, _guard) = capture_per_vm_log(); let closed = Arc::new(AtomicUsize::new(0)); let ok = instance("ok", 4, Arc::clone(&closed)); @@ -1774,9 +1677,9 @@ mod tests { ); assert_eq!(evaluated.load(Ordering::Relaxed), 1); assert_eq!(removed.load(Ordering::Relaxed), 1); - assert_eq!(per_vm_log_events(&controller, "ok"), ["event=vm_added"]); + assert_eq!(per_vm_log_events(&per_vm_log, "ok"), ["event=vm_added"]); assert_eq!( - per_vm_log_events(&controller, "bad"), + per_vm_log_events(&per_vm_log, "bad"), ["event=vm_added", "event=vm_removed reason=refresh_failed"] ); } diff --git a/src/engines/threshold.rs b/src/engines/threshold.rs index 045bc2f8..3102812b 100644 --- a/src/engines/threshold.rs +++ b/src/engines/threshold.rs @@ -751,7 +751,7 @@ impl ScalingEngine for ThresholdEngine { if self.cfg.block_scale_up_when_cgroup_throttled && throttled_recently { tracing::info!( target: "controller", - id = %instance.id, + vm = %instance.id, "scale-up suppressed by cgroup CPU throttling" ); return ScaleAction::None; diff --git a/src/lib.rs b/src/lib.rs index ebc08606..252818b3 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -12,6 +12,7 @@ pub mod daemon; pub mod dbus; pub mod engines; pub mod instance; +pub mod per_vm_log; pub mod rolling; pub mod state; pub mod util; diff --git a/src/main.rs b/src/main.rs index 06aaccb9..feaff3f0 100644 --- a/src/main.rs +++ b/src/main.rs @@ -13,10 +13,12 @@ use io_thread_controller::{ backends::registered_backends, config::{Config, ConfigError, dump_default_config, load_json_with_includes, validate_config}, daemon::{DaemonError, VERSION, run}, + per_vm_log::{self, PER_VM_LOG_DIR, PerVmLog}, util::Path, }; use thiserror::Error; use tracing::{debug, info, instrument}; +use tracing_subscriber::{Layer, filter::LevelFilter, fmt, layer::SubscriberExt}; #[derive(Error, Debug)] enum IoThreadControllerError { @@ -104,7 +106,8 @@ async fn main() -> Result<(), IoThreadControllerError> { } let matches = command.get_matches(); let cli = Cli::from_arg_matches(&matches)?; - init_logging(&cli.log_level, cli.log_style); + let per_vm_log = PerVmLog::default(); + init_logging(&cli.log_level, cli.log_style, per_vm_log.clone()); if cli.dump_config { println!("{}", dump_default_config()); @@ -138,6 +141,10 @@ async fn main() -> Result<(), IoThreadControllerError> { cfg.dry_run = true; } validate_config(&cfg)?; + 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?; Ok(()) @@ -154,7 +161,7 @@ fn load_daemon_config(path: &Path) -> Result { } /// Install the process-wide tracing subscriber. -fn init_logging(filter: &str, style: LogStyle) { +fn init_logging(filter: &str, style: LogStyle, per_vm_log: PerVmLog) { // Auto-detection: systemd sets `JOURNAL_STREAM=:` // on every service invocation whose stderr is journald. // Presence is enough; the value itself is only useful when @@ -173,38 +180,17 @@ fn init_logging(filter: &str, style: LogStyle) { } } }; - let env_filter = tracing_subscriber::EnvFilter::try_new(filter) - .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")); - // `tracing_subscriber::fmt` returns two different concrete - // subscriber types depending on whether `.without_time()` - // is chained, so we build and install each branch in its - // own scope rather than fighting the type system to - // unify them. - match effective_style { - LogStyle::Systemd => { - let subscriber = tracing_subscriber::fmt() - // Keep target so downstream routing can separate - // per-VM status (`status`) from controller logs. - .with_target(true) - .without_time() - .with_env_filter(env_filter) - .finish(); - let _ = tracing::subscriber::set_global_default(subscriber); - } - LogStyle::Json => { - let subscriber = tracing_subscriber::fmt() - .json() - .with_target(true) - .with_env_filter(env_filter) - .finish(); - let _ = tracing::subscriber::set_global_default(subscriber); - } - LogStyle::Human | LogStyle::Auto => { - let subscriber = tracing_subscriber::fmt() - .with_target(true) - .with_env_filter(env_filter) - .finish(); - let _ = tracing::subscriber::set_global_default(subscriber); - } - } + let main_log = match effective_style { + // Keep target so downstream routing can separate per-VM status + // (`status`) from controller logs. + LogStyle::Systemd => fmt::layer().with_target(true).without_time().boxed(), + LogStyle::Json => fmt::layer().json().with_target(true).boxed(), + LogStyle::Human | LogStyle::Auto => fmt::layer().with_target(true).boxed(), + }; + // Per-VM files have their own filter so that RUST_LOG, which configures + // the main log, does not drop their status records. + let subscriber = tracing_subscriber::registry() + .with(main_log.with_filter(per_vm_log::main_log_filter(filter))) + .with(per_vm_log.with_filter(LevelFilter::INFO)); + let _ = tracing::subscriber::set_global_default(subscriber); } diff --git a/src/per_vm_log.rs b/src/per_vm_log.rs new file mode 100644 index 00000000..7ea41822 --- /dev/null +++ b/src/per_vm_log.rs @@ -0,0 +1,262 @@ +// SPDX-License-Identifier: BSD-3-Clause +// Copyright (c) 2026 Nutanix, Inc. +// +// Author: Thanos Makatos + +//! Per-VM log files. +//! +//! [`PerVmLog`] is a [`tracing_subscriber::Layer`] that appends every event +//! carrying a `vm` field to that VM's own log file, without repeating the `vm` +//! field. Events on [`TARGET`] belong only in these files and are filtered out +//! of the main log. + +use std::{ + collections::HashSet, + fmt::{self, Write as _}, + fs::{OpenOptions, create_dir_all}, + io::Write as _, + path::PathBuf, + sync::{Arc, Mutex, RwLock}, + time::SystemTime, +}; + +use tracing::{ + Event, Subscriber, + field::{Field, Visit}, +}; +use tracing_subscriber::{EnvFilter, Layer, layer::Context}; + +use crate::util::Path; + +/// Default per-VM log directory. +pub const PER_VM_LOG_DIR: &str = "/var/log/io-thread-controller/vms"; + +/// Target for events that belong only in per-VM log files. +pub const TARGET: &str = "per_vm"; + +/// Build the main log's filter from `directives`, excluding [`TARGET`]. +/// +/// Invalid directives fall back to `info`. +pub fn main_log_filter(directives: &str) -> EnvFilter { + EnvFilter::try_new(directives) + .unwrap_or_else(|_| EnvFilter::new("info")) + .add_directive(format!("{TARGET}=off").parse().expect("valid directive")) +} + +/// Layer writing `vm`-tagged events to per-VM files; clones share state. +#[derive(Clone, Default)] +pub struct PerVmLog { + inner: Arc, +} + +#[derive(Default)] +struct Inner { + /// Destination directory; `None` disables per-VM files. + dir: RwLock>, + /// VM IDs for which per-VM log I/O failures were already reported. + warned: Mutex>, +} + +impl PerVmLog { + /// Write per-VM files under `dir`, or stop writing them with `None`. + pub fn set_dir(&self, dir: Option) { + *self.inner.dir.write().expect("lock poisoned") = dir; + } + + /// Return the log file of `vm`, or `None` while per-VM files are disabled. + pub fn path(&self, vm: &str) -> Option { + let name: String = vm + .chars() + .map(|ch| { + if ch.is_ascii_alphanumeric() || matches!(ch, '-' | '_' | '.') { + ch + } else { + '_' + } + }) + .collect(); + self.inner + .dir + .read() + .expect("lock poisoned") + .as_ref() + .map(|dir| dir.join(format!("{name}.log"))) + } + + fn append(&self, vm: &str, line: &str) { + let Some(path) = self.path(vm) else { + return; + }; + let result = path.parent().map_or(Ok(()), create_dir_all).and_then(|()| { + let mut file = OpenOptions::new().create(true).append(true).open(&path)?; + writeln!( + file, + "{} {line}", + humantime::format_rfc3339_millis(SystemTime::now()) + ) + }); + if let Err(error) = result + && self + .inner + .warned + .lock() + .expect("lock poisoned") + .insert(vm.to_string()) + { + // Reporting through tracing from inside a layer would re-enter the + // subscriber while it is dispatching this event. + eprintln!( + "WARN failed to write per-VM log path={} error={error}", + path.display() + ); + } + } +} + +impl Layer for PerVmLog { + fn on_event(&self, event: &Event<'_>, _ctx: Context<'_, S>) { + if self.inner.dir.read().expect("lock poisoned").is_none() { + return; + } + let mut record = Record::default(); + event.record(&mut record); + if let Some(vm) = record.vm.take() { + self.append(&vm, &record.line()); + } + } +} + +/// One event split into its VM, message, and remaining `key=value` fields. +#[derive(Default)] +struct Record { + vm: Option, + message: Option, + fields: String, +} + +impl Record { + fn record(&mut self, field: &Field, value: String) { + match field.name() { + "vm" => self.vm = Some(value), + "message" => self.message = Some(value), + name => { + if !self.fields.is_empty() { + self.fields.push(' '); + } + let _ = write!(self.fields, "{name}={value}"); + } + } + } + + /// Render the message, if any, followed by the remaining fields. + fn line(self) -> String { + match self.message { + Some(message) if self.fields.is_empty() => message, + Some(message) => format!("{message} {}", self.fields), + None => self.fields, + } + } +} + +impl Visit for Record { + fn record_str(&mut self, field: &Field, value: &str) { + self.record(field, value.to_string()); + } + + fn record_debug(&mut self, field: &Field, value: &dyn fmt::Debug) { + self.record(field, format!("{value:?}")); + } +} + +#[cfg(test)] +mod tests { + use tracing_subscriber::{filter::LevelFilter, layer::SubscriberExt}; + + use super::*; + + /// Test that only `vm`-tagged events reach the per-VM file, rendered + /// as message then fields and without the `vm` field. + #[test] + fn test_writes_vm_tagged_events_without_vm_field() { + let dir = tempfile::tempdir().unwrap(); + let log = PerVmLog::default(); + log.set_dir(Some(Path::new(dir.path()))); + let subscriber = tracing_subscriber::registry().with(log.clone()); + tracing::subscriber::with_default(subscriber, || { + tracing::info!(target: TARGET, vm = "vm-a", "status=1 2 3"); + tracing::warn!(target: "controller", event = "scale_failed", vm = %"vm-a", target = 4, "oops"); + tracing::info!(target: "controller", event = "vm_added", vm = "vm-a"); + tracing::info!(target: "controller", "no VM here"); + }); + + let contents = std::fs::read_to_string(log.path("vm-a").unwrap()).unwrap(); + let records: Vec<_> = contents + .lines() + .map(|line| line.split_once(' ').unwrap().1) + .collect(); + assert_eq!( + records, + [ + "status=1 2 3", + "oops event=scale_failed target=4", + "event=vm_added" + ] + ); + assert_eq!(std::fs::read_dir(dir.path()).unwrap().count(), 1); + } + + #[derive(Clone, Default)] + struct Buffer(Arc>>); + + impl std::io::Write for Buffer { + fn write(&mut self, bytes: &[u8]) -> std::io::Result { + self.0.lock().unwrap().extend_from_slice(bytes); + Ok(bytes.len()) + } + + fn flush(&mut self) -> std::io::Result<()> { + Ok(()) + } + } + + /// Test that per-VM-only events stay out of the main log, and that the + /// main log's level does not drop them from per-VM files. + #[test] + fn test_per_vm_target_bypasses_main_log() { + let dir = tempfile::tempdir().unwrap(); + let log = PerVmLog::default(); + log.set_dir(Some(Path::new(dir.path()))); + let main_log = Buffer::default(); + let writer = main_log.clone(); + let subscriber = tracing_subscriber::registry() + .with( + tracing_subscriber::fmt::layer() + .with_writer(move || writer.clone()) + .with_filter(main_log_filter("warn")), + ) + .with(log.clone().with_filter(LevelFilter::INFO)); + tracing::subscriber::with_default(subscriber, || { + tracing::info!(target: TARGET, vm = "vm-a", "status=1"); + tracing::warn!(target: "controller", vm = "vm-a", "refresh failed"); + }); + + let main_log = String::from_utf8(main_log.0.lock().unwrap().clone()).unwrap(); + assert!(!main_log.contains("status=1"), "{main_log}"); + assert!(main_log.contains("refresh failed"), "{main_log}"); + let contents = std::fs::read_to_string(log.path("vm-a").unwrap()).unwrap(); + assert_eq!(contents.lines().count(), 2, "{contents}"); + } + + /// Test that nothing is written while per-VM files are disabled. + #[test] + fn test_disabled_writes_nothing() { + let dir = tempfile::tempdir().unwrap(); + let log = PerVmLog::default(); + let subscriber = tracing_subscriber::registry().with(log.clone()); + tracing::subscriber::with_default(subscriber, || { + tracing::info!(target: TARGET, vm = "vm-a", "status=1"); + }); + assert!(log.path("vm-a").is_none()); + assert_eq!(std::fs::read_dir(dir.path()).unwrap().count(), 0); + } +} From e39bfdb2883b12ac229d6db53df455d243eba8d3 Mon Sep 17 00:00:00 2001 From: Thanos Makatos Date: Thu, 22 Oct 2026 13:59:34 +0000 Subject: [PATCH 16/17] fixup! Write per-VM log through a tracing layer Signed-off-by: Thanos Makatos --- src/controller.rs | 1 + 1 file changed, 1 insertion(+) diff --git a/src/controller.rs b/src/controller.rs index 920a1771..1485f28c 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -25,6 +25,7 @@ use crate::{ AppliedOutcome, BlockedReason, EngineTickContext, PsiSample, ScaleAction, ScalingEngine, }, instance::{CgroupError, Instance, InstanceStatus, SnapshotLatency}, + per_vm_log, state::{StateError, VmOwnership, VmStateStore}, }; /// Wire-facing container for the D-Bus `GetSnapshot` reply. From 599df58301a51417f653ac326b123323159a55b1 Mon Sep 17 00:00:00 2001 From: Thanos Makatos Date: Fri, 23 Oct 2026 06:57:57 +0000 Subject: [PATCH 17/17] Express configured durations in seconds, not polls Durations are now integer seconds in settings ending in _secs: scale_poll_secs, cooldown_secs (now integers, were floats) scale_down_sustain_polls -> scale_down_sustain_secs scale_validation_sample_polls -> scale_validation_delay_secs scale_validation_sample_polls_per_thread -> scale_validation_delay_per_thread_secs post_revert_up_grace_polls -> post_revert_up_grace_secs post_revert_up_recovery_polls -> post_revert_up_recovery_secs post_revert_up_backoff_max_polls -> post_revert_up_backoff_max_secs scale_down_backoff_max_polls -> scale_down_backoff_max_secs Every duration must be a multiple of scale_poll_secs. The engine receives the poll period when it is built and converts its durations to polls once, so engine logic is unchanged. The daemon refuses to start if any duration, controller or engine, is not a multiple, listing all of them with the nearest valid values. The round_up_durations config option and --round-up-durations flag (both off by default) instead round such durations up with a warning, so that a new poll period can be tried without rewriting the config. The daemon now logs the effective controller, engine, and backend configuration at startup, which shows rounded values. The daemon also prints errors with Display rather than Debug. AI-generated: everything. Signed-off-by: Thanos Makatos --- README.md | 30 +- io-thread-controller.d/engines/threshold.json | 14 +- io-thread-controller.json | 5 +- src/config.rs | 253 +++++++++++-- src/controller.rs | 47 +-- src/daemon.rs | 39 +- src/engines/mod.rs | 28 +- src/engines/threshold.rs | 356 ++++++++++++------ src/main.rs | 31 +- 9 files changed, 591 insertions(+), 212 deletions(-) diff --git a/README.md b/README.md index e4ece764..e35b01c8 100644 --- a/README.md +++ b/README.md @@ -90,6 +90,20 @@ Controller, engine, and backend JSON files may contain an `include` field. Its path is resolved relative to the including file, loaded recursively, and merged before the including file's keys are applied. +## Durations + +Durations are integer seconds in settings ending in `_secs`. Every duration, +including engine ones, must be a whole multiple of `scale_poll_secs`, and zero +is allowed. Otherwise the daemon refuses to start and lists each offending +setting with the nearest valid values. Changing only `scale_poll_secs` can +therefore invalidate other settings. + +To try a new poll period without rewriting the other settings, set +`round_up_durations` or pass `--round-up-durations`. Each offending duration is +then rounded up to the next multiple with a warning. At startup the daemon logs +the effective controller, engine, and backend configuration, which shows the +rounded values. + ## Thread-name filtering Backends may select fallback `/proc` CPU samples with regular expressions and @@ -131,21 +145,21 @@ queries and thread-count changes need an independently stateful handle for each VM. This keeps the controller independent of libvirt, QMP, control sockets, and other transport details. -After a successful action, the engine waits -`scale_validation_sample_polls` complete samples. The optional -`scale_validation_sample_polls_per_thread` setting adds that many samples for -each thread in the resulting pool, allowing larger pools more time to settle. +After a successful action, the engine samples for +`scale_validation_delay_secs`. The optional +`scale_validation_delay_per_thread_secs` setting adds that much time for each +thread in the resulting pool, allowing larger pools more time to settle. A scale-up is reverted unless IOPS gained at least `scale_up_min_gain_percent`; a scale-down is reverted when IOPS lost more than `scale_down_revert_drop_percent`. Setting either percentage to zero disables that direction's validation. Repeated failures back off per VM. After a reverted scale-up, scale-up is -blocked for `post_revert_up_grace_polls` plus `post_revert_up_recovery_polls`; +blocked for `post_revert_up_grace_secs` plus `post_revert_up_recovery_secs`; that hold doubles with each consecutive failed scale-up, up to -`post_revert_up_backoff_max_polls`. After a reverted scale-down, the next -scale-down needs `scale_down_sustain_polls` low-utilisation polls doubled once -per consecutive failure, up to `scale_down_backoff_max_polls`. A kept action +`post_revert_up_backoff_max_secs`. After a reverted scale-down, the next +scale-down needs `scale_down_sustain_secs` of low utilisation doubled once per +consecutive failure, up to `scale_down_backoff_max_secs`. A kept action resets its direction's backoff, as does an applied action in the opposite direction. Setting a maximum to zero disables that direction's backoff. diff --git a/io-thread-controller.d/engines/threshold.json b/io-thread-controller.d/engines/threshold.json index 7bbbde42..110084e9 100644 --- a/io-thread-controller.d/engines/threshold.json +++ b/io-thread-controller.d/engines/threshold.json @@ -1,14 +1,14 @@ { "scale_up_threshold_percent": 95, - "scale_down_sustain_polls": 12, + "scale_down_sustain_secs": 120, "max_scale_down_step": 0, "scale_up_min_gain_percent": 5, "scale_down_revert_drop_percent": 3, - "scale_validation_sample_polls": 3, - "scale_validation_sample_polls_per_thread": 0, + "scale_validation_delay_secs": 30, + "scale_validation_delay_per_thread_secs": 0, "block_scale_up_when_cgroup_throttled": true, - "post_revert_up_grace_polls": 1, - "post_revert_up_recovery_polls": 2, - "post_revert_up_backoff_max_polls": 30, - "scale_down_backoff_max_polls": 120 + "post_revert_up_grace_secs": 10, + "post_revert_up_recovery_secs": 20, + "post_revert_up_backoff_max_secs": 300, + "scale_down_backoff_max_secs": 1200 } diff --git a/io-thread-controller.json b/io-thread-controller.json index 9bab418e..9efee460 100644 --- a/io-thread-controller.json +++ b/io-thread-controller.json @@ -1,11 +1,12 @@ { - "scale_poll_secs": 10.0, + "scale_poll_secs": 10, + "round_up_durations": false, "enable_per_vm_status_line": true, "enable_aggregate_status_line": true, "min_thread_count": 1, "max_thread_count": 8, "host_cpu_scale_up_ceiling_percent": 90, - "cooldown_secs": 10.0, + "cooldown_secs": 10, "engine": "threshold", "engine_config_dir": "/etc/io-thread-controller.d/engines", "backend_config_dir": "/etc/io-thread-controller.d/backends", diff --git a/src/config.rs b/src/config.rs index 8c7b106e..9e4b4156 100644 --- a/src/config.rs +++ b/src/config.rs @@ -5,7 +5,7 @@ //! Configuration for the io-thread-controller daemon. -use std::io; +use std::{io, time::Duration}; use serde::{Deserialize, Deserializer, Serialize, Serializer}; use thiserror::Error; @@ -17,6 +17,13 @@ pub enum ConfigError { #[error("invalid configuration value: {0}")] InvalidValue(String), + #[error( + "durations must be multiples of scale_poll_secs (set round_up_durations or pass \ + --round-up-durations to round them up):\n {}", + .0.join("\n ") + )] + NotPollMultiple(Vec), + #[error(transparent)] Io(#[from] io::Error), @@ -59,11 +66,15 @@ pub struct Config { pub host_cpu_scale_up_ceiling: f64, /// Per-VM delay after ordinary successful scale actions. #[serde(default = "default_cooldown_secs")] - pub cooldown_secs: f64, - /// Poll cadence in seconds. The controller refreshes - /// per-instance state on each tick and calls the active - /// scaling engine for a decision. - pub scale_poll_secs: f64, + pub cooldown_secs: u64, + /// Poll cadence in seconds. The controller refreshes per-instance state on + /// each tick and calls the active scaling engine for a decision. Every + /// other duration, including engine ones, must be a multiple of it. + pub scale_poll_secs: u64, + /// Round durations that are not multiples of `scale_poll_secs` up to the + /// next multiple, with a warning, instead of rejecting them. + #[serde(default)] + pub round_up_durations: bool, /// Emit a `vm=` status line per tracked instance on /// every tick (default `true`). #[serde(default = "default_true")] @@ -111,8 +122,92 @@ fn default_host_cpu_scale_up_ceiling() -> f64 { 0.9 } -fn default_cooldown_secs() -> f64 { - 30.0 +fn default_cooldown_secs() -> u64 { + 30 +} + +/// Controller poll period that durations are expressed in multiples of. +#[derive(Debug, Clone, Copy)] +pub struct PollPeriod { + secs: u64, + round_up: bool, +} + +impl PollPeriod { + /// `secs` must be non-zero. + pub fn new(secs: u64, round_up: bool) -> Self { + assert!(secs != 0, "poll period must be non-zero"); + Self { secs, round_up } + } + + /// Start converting a group of durations to poll counts. + pub fn converter(self) -> PollConverter { + PollConverter { + poll: self, + errors: Vec::new(), + } + } +} + +/// Converts durations to poll counts, collecting every invalid duration so +/// that they can be reported together. +pub struct PollConverter { + poll: PollPeriod, + errors: Vec, +} + +impl PollConverter { + /// Return the poll period. + pub fn period(&self) -> Duration { + Duration::from_secs(self.poll.secs) + } + + /// Return how many polls `secs` spans. + /// + /// A `secs` that is not a multiple of the poll period is either rounded + /// up in place, so the effective configuration shows the value in use, or + /// recorded as an error for [`Self::finish`]. + pub fn polls(&mut self, name: &str, secs: &mut u64) -> u32 { + let period = self.poll.secs; + let mut polls = *secs / period; + if !secs.is_multiple_of(period) { + let below = polls * period; + let above = below.saturating_add(period); + if !self.poll.round_up { + self.errors.push(format!( + "{name} ({secs}) is not a multiple of scale_poll_secs ({period}); \ + nearest valid values: {below} or {above}" + )); + return 0; + } + tracing::warn!( + target: "controller", + setting = name, + configured = *secs, + effective = above, + "rounded duration up to a multiple of scale_poll_secs" + ); + *secs = above; + polls += 1; + } + match u32::try_from(polls) { + Ok(polls) => polls, + Err(_) => { + self.errors + .push(format!("{name} ({secs}) spans too many polls")); + 0 + } + } + } + + /// Fail with every duration rejected by [`Self::polls`]. + pub fn finish(self) -> Result<(), ConfigError> { + if self.errors.is_empty() { + Ok(()) + } else { + Err(ConfigError::NotPollMultiple(self.errors)) + } + } } /// Deserialise a human-readable percent to a floating point value. @@ -140,11 +235,12 @@ impl Default for Config { engine: "threshold".to_string(), engine_config_dir: Path::new("/etc/io-thread-controller.d/engines"), backend_config_dir: Path::new("/etc/io-thread-controller.d/backends"), - scale_poll_secs: 10.0, + scale_poll_secs: 10, + round_up_durations: false, min_thread_count: 2, max_thread_count: 8, host_cpu_scale_up_ceiling: 0.9, - cooldown_secs: 30.0, + cooldown_secs: default_cooldown_secs(), vm_state_path: Path::new("/run/io-thread-controller/vm-ownership.json"), enable_per_vm_status_line: true, enable_per_vm_main_log: false, @@ -200,11 +296,26 @@ fn load_config_value(path: &Path) -> Result { Ok(merged) } +impl Config { + /// Return the poll period that every other duration is a multiple of. + /// + /// Panics if `scale_poll_secs` is zero, which [`validate_config`] + /// rejects. + pub fn poll_period(&self) -> PollPeriod { + PollPeriod::new(self.scale_poll_secs, self.round_up_durations) + } +} + // TODO add a RawConfig or ValidatedConfig to ensure an unvalidated config // cannot be constructed outside this module. -/// Ensures configuration is internally coherent. -pub fn validate_config(cfg: &Config) -> Result<(), ConfigError> { - if !cfg.scale_poll_secs.is_finite() || cfg.scale_poll_secs <= 0.0 { +/// Ensures configuration is internally coherent, rounding durations up to +/// poll multiples when `round_up_durations` is set. +/// +/// Invalid durations are not reported yet: the returned converter holds them +/// so that the engine can convert its own durations with it, and +/// [`PollConverter::finish`] then reports every invalid duration at once. +pub fn validate_config(cfg: &mut Config) -> Result { + if cfg.scale_poll_secs == 0 { return Err(ConfigError::InvalidValue( "scale_poll_secs must be > 0".to_string(), )); @@ -226,12 +337,9 @@ pub fn validate_config(cfg: &Config) -> Result<(), ConfigError> { "host_cpu_scale_up_ceiling_percent must be in [0, 100]".to_string(), )); } - if !(cfg.cooldown_secs.is_finite() && cfg.cooldown_secs >= 0.0) { - return Err(ConfigError::InvalidValue( - "cooldown_secs must be finite and non-negative".to_string(), - )); - } - Ok(()) + let mut converter = cfg.poll_period().converter(); + converter.polls("cooldown_secs", &mut cfg.cooldown_secs); + Ok(converter) } /// Serialise the default config as indented JSON for CLI @@ -248,10 +356,10 @@ mod tests { /// engine with a 10s poll. #[test] fn default_config_is_valid() { - let cfg = Config::default(); + let mut cfg = Config::default(); assert_eq!(cfg.engine, "threshold"); - assert!((cfg.scale_poll_secs - 10.0).abs() < f64::EPSILON); - validate_config(&cfg).unwrap(); + assert_eq!(cfg.scale_poll_secs, 10); + validate_config(&mut cfg).unwrap().finish().unwrap(); } /// Test that `validate_config` rejects an empty engine name. @@ -259,24 +367,96 @@ mod tests { fn validate_rejects_empty_engine() { let mut cfg = Config::default(); cfg.engine.clear(); - let err = validate_config(&cfg).unwrap_err().to_string(); + let err = validate_config(&mut cfg).err().unwrap().to_string(); assert!(err.contains("engine must be non-empty")); } - /// Test that `validate_config` rejects non-positive - /// `scale_poll_secs`. + /// Test that `validate_config` rejects a zero `scale_poll_secs`. #[test] - fn validate_rejects_non_positive_poll() { + fn validate_rejects_zero_poll() { let mut cfg = Config { - scale_poll_secs: 0.0, + scale_poll_secs: 0, ..Default::default() }; - let err = validate_config(&cfg).unwrap_err().to_string(); + let err = validate_config(&mut cfg).err().unwrap().to_string(); assert!(err.contains("scale_poll_secs must be > 0")); + } - cfg.scale_poll_secs = f64::NAN; - let err = validate_config(&cfg).unwrap_err().to_string(); - assert!(err.contains("scale_poll_secs must be > 0")); + /// Tests that a cooldown that is not a multiple of the poll period is + /// rejected with the nearest valid values, and left unchanged. + #[test] + fn test_validate_rejects_cooldown_not_poll_multiple() { + let mut cfg = Config { + cooldown_secs: 25, + ..Default::default() + }; + let err = validate_config(&mut cfg) + .unwrap() + .finish() + .unwrap_err() + .to_string(); + assert!( + err.contains( + "cooldown_secs (25) is not a multiple of scale_poll_secs (10); nearest valid \ + values: 20 or 30" + ), + "{err}" + ); + assert!(err.contains("--round-up-durations"), "{err}"); + assert_eq!(cfg.cooldown_secs, 25); + } + + /// Tests that `round_up_durations` rounds a non-multiple cooldown up to + /// the next poll multiple instead of rejecting it. + #[test] + fn test_validate_rounds_up_cooldown() { + let mut cfg = Config { + cooldown_secs: 25, + round_up_durations: true, + ..Default::default() + }; + validate_config(&mut cfg).unwrap().finish().unwrap(); + assert_eq!(cfg.cooldown_secs, 30); + } + + /// Tests that the poll converter counts polls, accepts zero, reports + /// every invalid duration at once, and rounds up when enabled. + #[test] + fn test_poll_converter() { + let mut converter = PollPeriod::new(4, false).converter(); + assert_eq!(converter.polls("zero", &mut 0), 0); + assert_eq!(converter.polls("exact", &mut 12), 3); + converter.polls("short", &mut 1); + converter.polls("long", &mut 30); + let err = converter.finish().unwrap_err().to_string(); + assert!( + err.contains( + "short (1) is not a multiple of scale_poll_secs (4); nearest valid values: 0 or 4" + ), + "{err}" + ); + assert!( + err.contains( + "long (30) is not a multiple of scale_poll_secs (4); nearest valid values: 28 or 32" + ), + "{err}" + ); + + let mut converter = PollPeriod::new(4, true).converter(); + let mut secs = 30; + assert_eq!(converter.polls("long", &mut secs), 8); + assert_eq!(secs, 32); + converter.finish().unwrap(); + } + + /// Tests that durations must be integer seconds. + #[test] + fn test_durations_must_be_integers() { + let err = serde_json::from_str::( + r#"{"engine": "foo", "engine_config_dir": "/engines", "backend_config_dir": "/backends", "scale_poll_secs": 7.5, "vm_state_path": "/path/to/vm-state"}"#, + ) + .unwrap_err(); + assert!(err.to_string().contains("expected u64"), "{err}"); } /// Test that minimal valid JSON loads into `Config` with the @@ -291,7 +471,7 @@ mod tests { "engine": "threshold", "engine_config_dir": "/etc/io-thread-controller/engines", "backend_config_dir": "/etc/io-thread-controller/backends", - "scale_poll_secs": 7.5, + "scale_poll_secs": 7, "vm_state_path": "/var/lib/io-thread-controller/vm-state.json" }"#, ) @@ -299,7 +479,7 @@ mod tests { let cfg: Config = load_json_with_includes(Path::new(path.to_str().unwrap())).unwrap(); assert_eq!(cfg.engine, "threshold"); - assert!((cfg.scale_poll_secs - 7.5).abs() < f64::EPSILON); + assert_eq!(cfg.scale_poll_secs, 7); assert_eq!( cfg.engine_config_dir.as_os_str(), std::ffi::OsStr::new("/etc/io-thread-controller/engines") @@ -318,7 +498,7 @@ mod tests { "engine": "threshold", "engine_config_dir": "/engines", "backend_config_dir": "/backends", - "scale_poll_secs": 1.0, + "scale_poll_secs": 1, "not_a_real_field": true }"#, ) @@ -334,7 +514,7 @@ mod tests { fn dump_default_config_contains_engine() { let dumped = dump_default_config(); assert!(dumped.contains(r#""engine": "threshold""#)); - assert!(dumped.contains(r#""scale_poll_secs": 10.0"#)); + assert!(dumped.contains(r#""scale_poll_secs": 10"#)); assert!(dumped.contains(r#""/etc/io-thread-controller.d/engines""#)); } @@ -346,7 +526,8 @@ mod tests { assert_eq!(cfg.min_thread_count, 1); assert_eq!(cfg.max_thread_count, 8); assert!((cfg.host_cpu_scale_up_ceiling - 0.9).abs() < f64::EPSILON); - assert_eq!(cfg.cooldown_secs, 30.0); + assert_eq!(cfg.cooldown_secs, 30); + assert!(!cfg.round_up_durations); let serialized = serde_json::to_string(&cfg).unwrap(); assert!(serialized.contains(r#""host_cpu_scale_up_ceiling_percent":90.0"#)); diff --git a/src/controller.rs b/src/controller.rs index 1485f28c..65bb2714 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -550,7 +550,6 @@ impl Controller { host_cpu_util: self.host_cpu_util, psi: self.psi, tick_index: self.tick_index, - scale_poll_secs: self.cfg.scale_poll_secs, }; let fleet: Vec<_> = self.instances.values().cloned().collect(); let refresh_cgroup = self.cfg.refresh_cgroup_on_each_read; @@ -808,7 +807,7 @@ impl Controller { ScaleAction::Up(_) | ScaleAction::Down(_) | ScaleAction::Revert(_) ) { instance.status.write().await.cooldown_until = - Some(Instant::now() + Duration::from_secs_f64(self.cfg.cooldown_secs)); + Some(Instant::now() + Duration::from_secs(self.cfg.cooldown_secs)); } let applied_at = Instant::now(); self.engine @@ -835,7 +834,7 @@ impl Controller { ScaleAction::Up(_) | ScaleAction::Down(_) | ScaleAction::Revert(_) ) { status.cooldown_until = - Some(applied_at + Duration::from_secs_f64(self.cfg.cooldown_secs)); + Some(applied_at + Duration::from_secs(self.cfg.cooldown_secs)); } drop(status); self.engine @@ -1088,6 +1087,14 @@ mod tests { #[cfg(feature = "threshold-engine")] use crate::engines::threshold::{ThresholdConfig, ThresholdEngine}; + #[cfg(feature = "threshold-engine")] + fn default_threshold_engine() -> ThresholdEngine { + let mut durations = Config::default().poll_period().converter(); + let engine = ThresholdEngine::new(ThresholdConfig::default(), &mut durations).unwrap(); + durations.finish().unwrap(); + engine + } + struct VcpuLimitedClient { target: Arc, } @@ -1118,7 +1125,7 @@ mod tests { let state_dir = tempfile::tempdir().unwrap(); let cfg = Config { vm_state_path: Path::new(&state_dir.path().join("ownership.json")), - cooldown_secs: 0.0, + cooldown_secs: 0, ..Default::default() }; let target = Arc::new(AtomicU32::new(2)); @@ -1137,11 +1144,7 @@ mod tests { status.vcpu_count = 4; status.ownership_classification = Some(true); } - let mut controller = Controller::new( - cfg, - Box::new(ThresholdEngine::new(ThresholdConfig::default())), - ) - .unwrap(); + let mut controller = Controller::new(cfg, Box::new(default_threshold_engine())).unwrap(); controller .instances .insert(instance.id.clone(), Arc::clone(&instance)); @@ -1200,7 +1203,7 @@ mod tests { min_thread_count: 2, max_thread_count: 3, host_cpu_scale_up_ceiling: 0.5, - cooldown_secs: 30.0, + cooldown_secs: 30, ..Default::default() }; let target = Arc::new(AtomicU32::new(2)); @@ -1218,11 +1221,7 @@ mod tests { status.thread_count = 2; status.vcpu_count = 4; } - let mut controller = Controller::new( - cfg, - Box::new(ThresholdEngine::new(ThresholdConfig::default())), - ) - .unwrap(); + let mut controller = Controller::new(cfg, Box::new(default_threshold_engine())).unwrap(); controller.host_cpu_util = 0.5; controller .instances @@ -1272,7 +1271,7 @@ mod tests { let state_dir = tempfile::tempdir().unwrap(); let cfg = Config { vm_state_path: Path::new(&state_dir.path().join("ownership.json")), - cooldown_secs: 30.0, + cooldown_secs: 30, min_thread_count: 1, max_thread_count: 8, ..Default::default() @@ -1293,11 +1292,7 @@ mod tests { status.vcpu_count = 8; status.ownership_classification = Some(true); } - let mut controller = Controller::new( - cfg, - Box::new(ThresholdEngine::new(ThresholdConfig::default())), - ) - .unwrap(); + let mut controller = Controller::new(cfg, Box::new(default_threshold_engine())).unwrap(); controller .instances .insert(instance.id.clone(), Arc::clone(&instance)); @@ -1334,7 +1329,7 @@ mod tests { let state_dir = tempfile::tempdir().unwrap(); let cfg = Config { vm_state_path: Path::new(&state_dir.path().join("ownership.json")), - cooldown_secs: 60.0, + cooldown_secs: 60, min_thread_count: 1, max_thread_count: 8, ..Default::default() @@ -1355,11 +1350,7 @@ mod tests { status.vcpu_count = 8; status.ownership_classification = Some(true); } - let mut controller = Controller::new( - cfg, - Box::new(ThresholdEngine::new(ThresholdConfig::default())), - ) - .unwrap(); + let mut controller = Controller::new(cfg, Box::new(default_threshold_engine())).unwrap(); controller .instances .insert(instance.id.clone(), Arc::clone(&instance)); @@ -1419,7 +1410,7 @@ mod tests { vm_state_path: Path::new(&state_dir.path().join("ownership.json")), ..Default::default() }; - let engine = Box::new(ThresholdEngine::new(ThresholdConfig::default())); + let engine = Box::new(default_threshold_engine()); let mut controller = Controller::new(cfg, engine).unwrap(); let good_target = Arc::new(AtomicU32::new(1)); let good = BulkClient { diff --git a/src/daemon.rs b/src/daemon.rs index ed3ce058..a733e724 100644 --- a/src/daemon.rs +++ b/src/daemon.rs @@ -22,7 +22,7 @@ use crate::{ config::Config, controller::{Controller, ControllerError}, dbus, - engines::{EngineError, load_registered_engine}, + engines::ScalingEngine, instance::Instance, util::Path, }; @@ -32,9 +32,6 @@ pub enum DaemonError { #[error(transparent)] Controller(#[from] ControllerError), - #[error(transparent)] - Engine(#[from] EngineError), - #[error("backend watch path `{0}` is not a directory")] InvalidWatchPath(String), @@ -52,7 +49,11 @@ pub enum DaemonError { const INOTIFY_EVENT_BUF_SIZE: usize = 32 * 1024; /// Run until SIGTERM or SIGINT. -pub async fn run(cfg: Config, backends: Vec>) -> Result<(), DaemonError> { +pub async fn run( + cfg: Config, + engine: Box, + backends: Vec>, +) -> Result<(), DaemonError> { let per_vm_destination = if cfg.enable_per_vm_status_line { if cfg.enable_per_vm_main_log { "main log".to_string() @@ -85,7 +86,7 @@ pub async fn run(cfg: Config, backends: Vec>) -> Result<(), Dae "# aggregate line: " ); } - let engine = load_registered_engine(&cfg.engine_config_dir, &cfg.engine)?; + log_effective_config(&cfg, engine.as_ref(), &backends); let mut controller = Controller::new(cfg.clone(), engine)?; let discovered = discover_all(&backends).await; @@ -95,7 +96,7 @@ pub async fn run(cfg: Config, backends: Vec>) -> Result<(), Dae let (dbus_tx, mut dbus_rx) = tokio::sync::mpsc::channel(1); let _dbus_connection = dbus::serve(dbus::Service::new(dbus_tx)).await?; let mut inotify_events = setup_inotify(&backends)?; - let mut poll = interval(Duration::from_secs_f64(cfg.scale_poll_secs)); + let mut poll = interval(Duration::from_secs(cfg.scale_poll_secs)); poll.set_missed_tick_behavior(MissedTickBehavior::Skip); let mut sigterm = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::terminate())?; let mut sigint = tokio::signal::unix::signal(tokio::signal::unix::SignalKind::interrupt())?; @@ -126,6 +127,30 @@ pub async fn run(cfg: Config, backends: Vec>) -> Result<(), Dae Ok(()) } +/// Log the configuration in use after defaults, command-line overrides, and +/// duration rounding were applied. +fn log_effective_config(cfg: &Config, engine: &dyn ScalingEngine, backends: &[Box]) { + tracing::info!( + target: "controller", + config = %serde_json::to_value(cfg).unwrap_or_default(), + "effective controller configuration" + ); + tracing::info!( + target: "controller", + engine = engine.name(), + config = %engine.dump_config(), + "effective engine configuration" + ); + for backend in backends { + tracing::info!( + target: "controller", + backend = backend.name(), + config = %backend.dump_config(), + "effective backend configuration" + ); + } +} + /// Discover all backend inventories concurrently. async fn discover_all(backends: &[Box]) -> Vec> { join_all(backends.iter().map(|backend| backend.discover())) diff --git a/src/engines/mod.rs b/src/engines/mod.rs index 15b8c4c9..357e94f1 100644 --- a/src/engines/mod.rs +++ b/src/engines/mod.rs @@ -12,17 +12,28 @@ use futures_util::future::join_all; use linkme::distributed_slice; use thiserror::Error; -use crate::{backends::BackendClientError, config::ConfigError, instance::Instance, util::Path}; +use crate::{ + backends::BackendClientError, + config::{ConfigError, PollConverter}, + instance::Instance, + util::Path, +}; #[cfg(feature = "threshold-engine")] pub mod threshold; +/// Build an engine from ``. Engine durations must be +/// converted with the supplied converter, which reports invalid durations +/// together with the controller's after the engine is built. +pub type EngineBuilder = + fn(&Path, &mut PollConverter) -> Result, EngineError>; + /// One in-tree (or out-of-tree) engine factory registered on [`ENGINES`]. pub struct EngineRegistration { /// Stable name matched against `Config::engine`. pub name: &'static str, - /// Build the engine from ``. - pub build: fn(&Path) -> Result, EngineError>, + /// Build the engine. + pub build: EngineBuilder, } /// Linked-in engine factories. Feature-gated modules append themselves here. @@ -103,8 +114,6 @@ pub struct EngineTickContext { pub psi: PsiSample, /// Monotonically increasing tick sequence number. pub tick_index: u64, - /// Controller poll interval in seconds (for operator-facing durations). - pub scale_poll_secs: f64, } impl Default for EngineTickContext { @@ -116,7 +125,6 @@ impl Default for EngineTickContext { host_cpu_util: 0.0, psi: PsiSample::default(), tick_index: 0, - scale_poll_secs: 10.0, } } } @@ -273,10 +281,11 @@ pub trait ScalingEngine: Send + Sync { pub fn load_registered_engine( engine_config_dir: &Path, name: &str, + durations: &mut PollConverter, ) -> Result, EngineError> { for registration in ENGINES { if registration.name == name { - return (registration.build)(engine_config_dir); + return (registration.build)(engine_config_dir, durations); } } Err(EngineError::NoSuchEngine(name.to_string())) @@ -315,10 +324,11 @@ mod tests { #[test] fn load_registered_engine_finds_threshold_and_rejects_unknown() { let dir = Path::new("/tmp/missing-engines"); - let engine = load_registered_engine(&dir, "threshold").unwrap(); + let mut durations = crate::config::Config::default().poll_period().converter(); + let engine = load_registered_engine(&dir, "threshold", &mut durations).unwrap(); assert_eq!(engine.name(), "threshold"); - match load_registered_engine(&dir, "no-such-engine") { + match load_registered_engine(&dir, "no-such-engine", &mut durations) { Err(EngineError::NoSuchEngine(name)) => assert_eq!(name, "no-such-engine"), Ok(_) => panic!("expected NoSuchEngine"), Err(other) => panic!("expected NoSuchEngine, got {other}"), diff --git a/src/engines/threshold.rs b/src/engines/threshold.rs index 3102812b..0a932356 100644 --- a/src/engines/threshold.rs +++ b/src/engines/threshold.rs @@ -5,14 +5,18 @@ //! Threshold engine: size a worker pool to carry the observed CPU load. -use std::{collections::HashMap, sync::Arc, time::Instant}; +use std::{ + collections::HashMap, + sync::Arc, + time::{Duration, Instant}, +}; use async_trait::async_trait; use serde::{Deserialize, Serialize}; use tokio::sync::Mutex; use crate::{ - config::{ConfigError, deserialize_percent, serialize_percent}, + config::{ConfigError, PollConverter, deserialize_percent, serialize_percent}, engines::{ AppliedOutcome, EngineError, EngineRegistration, EngineTickContext, ScaleAction, ScalingEngine, @@ -24,8 +28,11 @@ use crate::{ /// Registry and configuration name of the threshold engine. pub const ENGINE_NAME: &str = "threshold"; -fn build_threshold_engine(dir: &Path) -> Result, EngineError> { - Ok(Box::new(ThresholdEngine::from_config_dir(dir)?)) +fn build_threshold_engine( + dir: &Path, + durations: &mut PollConverter, +) -> Result, EngineError> { + Ok(Box::new(ThresholdEngine::from_config_dir(dir, durations)?)) } #[linkme::distributed_slice(super::ENGINES)] @@ -35,6 +42,8 @@ static THRESHOLD_ENGINE: EngineRegistration = EngineRegistration { }; /// Scale, sustain, and post-action validation settings. +/// +/// Every duration must be a multiple of the controller poll period. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(deny_unknown_fields)] pub struct ThresholdConfig { @@ -46,9 +55,9 @@ pub struct ThresholdConfig { serialize_with = "serialize_percent" )] pub scale_up_threshold: f64, - /// Consecutive eligible polls required before scale-down. - #[serde(default = "default_scale_down_sustain_polls")] - pub scale_down_sustain_polls: u32, + /// Seconds scale-down must stay eligible before it is proposed. + #[serde(default = "default_scale_down_sustain_secs")] + pub scale_down_sustain_secs: u64, /// Maximum number of threads than can be reduced in a scale down operation. /// 0 means unlimited. #[serde(default = "default_max_scale_down_step")] @@ -69,39 +78,53 @@ pub struct ThresholdConfig { serialize_with = "serialize_percent" )] pub scale_down_revert_drop: f64, - /// Complete samples to wait before validating a successful action. - #[serde(default = "default_scale_validation_sample_polls")] - pub scale_validation_sample_polls: u32, - /// Additional validation polls per target thread after a scale action. + /// Seconds to sample after a successful action before validating it. + #[serde(default = "default_scale_validation_delay_secs")] + pub scale_validation_delay_secs: u64, + /// Additional validation seconds per target thread after a scale action. #[serde(default)] - pub scale_validation_sample_polls_per_thread: u32, + pub scale_validation_delay_per_thread_secs: u64, /// Suppress scale-up after the VM's cgroup reports new throttled CPU time. #[serde(default = "default_true")] pub block_scale_up_when_cgroup_throttled: bool, - /// Polls after a failed scale-up revert during which metrics are ignored. - #[serde(default = "default_post_revert_up_grace_polls")] - pub post_revert_up_grace_polls: u32, - /// Polls after grace spent measuring a settled baseline while scale-up is - /// still blocked. - #[serde(default = "default_post_revert_up_recovery_polls")] - pub post_revert_up_recovery_polls: u32, - /// Cap on the post-revert hold (grace plus recovery), which doubles with - /// each consecutive failed scale-up. Zero disables the backoff. - #[serde(default = "default_post_revert_up_backoff_max_polls")] - pub post_revert_up_backoff_max_polls: u32, - /// Cap on the low-utilisation polls required before scale-down, which - /// doubles with each consecutive failed scale-down. Zero disables the + /// Seconds after a failed scale-up revert during which metrics are + /// ignored. + #[serde(default = "default_post_revert_up_grace_secs")] + pub post_revert_up_grace_secs: u64, + /// Seconds after grace spent measuring a settled baseline while scale-up + /// is still blocked. + #[serde(default = "default_post_revert_up_recovery_secs")] + pub post_revert_up_recovery_secs: u64, + /// Cap in seconds on the post-revert hold (grace plus recovery), which + /// doubles with each consecutive failed scale-up. Zero disables the /// backoff. - #[serde(default = "default_scale_down_backoff_max_polls")] - pub scale_down_backoff_max_polls: u32, + #[serde(default = "default_post_revert_up_backoff_max_secs")] + pub post_revert_up_backoff_max_secs: u64, + /// Cap in seconds on the low-utilisation time required before scale-down, + /// which doubles with each consecutive failed scale-down. Zero disables + /// the backoff. + #[serde(default = "default_scale_down_backoff_max_secs")] + pub scale_down_backoff_max_secs: u64, +} + +/// [`ThresholdConfig`] durations converted to controller polls. +#[derive(Debug, Clone, Copy)] +struct ThresholdPolls { + scale_down_sustain: u32, + scale_validation_delay: u32, + scale_validation_delay_per_thread: u32, + post_revert_up_grace: u32, + post_revert_up_recovery: u32, + post_revert_up_backoff_max: u32, + scale_down_backoff_max: u32, } fn default_scale_up_threshold() -> f64 { 0.95 } -fn default_scale_down_sustain_polls() -> u32 { - 3 +fn default_scale_down_sustain_secs() -> u64 { + 30 } fn default_max_scale_down_step() -> u32 { @@ -114,23 +137,23 @@ fn default_scale_up_min_gain() -> f64 { fn default_scale_down_revert_drop() -> f64 { 0.05 } -fn default_scale_validation_sample_polls() -> u32 { - 2 +fn default_scale_validation_delay_secs() -> u64 { + 20 } fn default_true() -> bool { true } -fn default_post_revert_up_grace_polls() -> u32 { - 1 +fn default_post_revert_up_grace_secs() -> u64 { + 10 } -fn default_post_revert_up_recovery_polls() -> u32 { - 2 +fn default_post_revert_up_recovery_secs() -> u64 { + 20 } -fn default_post_revert_up_backoff_max_polls() -> u32 { - 30 +fn default_post_revert_up_backoff_max_secs() -> u64 { + 300 } -fn default_scale_down_backoff_max_polls() -> u32 { - 120 +fn default_scale_down_backoff_max_secs() -> u64 { + 1200 } /// `base` doubled `doublings` times, capped at `max` (never below `base`). @@ -147,17 +170,17 @@ impl Default for ThresholdConfig { fn default() -> Self { Self { scale_up_threshold: default_scale_up_threshold(), - scale_down_sustain_polls: default_scale_down_sustain_polls(), + scale_down_sustain_secs: default_scale_down_sustain_secs(), max_scale_down_step: default_max_scale_down_step(), scale_up_min_gain: default_scale_up_min_gain(), scale_down_revert_drop: default_scale_down_revert_drop(), - scale_validation_sample_polls: default_scale_validation_sample_polls(), - scale_validation_sample_polls_per_thread: 0, + scale_validation_delay_secs: default_scale_validation_delay_secs(), + scale_validation_delay_per_thread_secs: 0, block_scale_up_when_cgroup_throttled: true, - post_revert_up_grace_polls: default_post_revert_up_grace_polls(), - post_revert_up_recovery_polls: default_post_revert_up_recovery_polls(), - post_revert_up_backoff_max_polls: default_post_revert_up_backoff_max_polls(), - scale_down_backoff_max_polls: default_scale_down_backoff_max_polls(), + post_revert_up_grace_secs: default_post_revert_up_grace_secs(), + post_revert_up_recovery_secs: default_post_revert_up_recovery_secs(), + post_revert_up_backoff_max_secs: default_post_revert_up_backoff_max_secs(), + scale_down_backoff_max_secs: default_scale_down_backoff_max_secs(), } } } @@ -187,6 +210,39 @@ impl ThresholdConfig { } Ok(()) } + + /// Convert every duration to polls, rounding durations up in place when + /// `converter` allows it. + fn resolve_polls(&mut self, converter: &mut PollConverter) -> ThresholdPolls { + ThresholdPolls { + scale_down_sustain: converter + .polls("scale_down_sustain_secs", &mut self.scale_down_sustain_secs), + scale_validation_delay: converter.polls( + "scale_validation_delay_secs", + &mut self.scale_validation_delay_secs, + ), + scale_validation_delay_per_thread: converter.polls( + "scale_validation_delay_per_thread_secs", + &mut self.scale_validation_delay_per_thread_secs, + ), + post_revert_up_grace: converter.polls( + "post_revert_up_grace_secs", + &mut self.post_revert_up_grace_secs, + ), + post_revert_up_recovery: converter.polls( + "post_revert_up_recovery_secs", + &mut self.post_revert_up_recovery_secs, + ), + post_revert_up_backoff_max: converter.polls( + "post_revert_up_backoff_max_secs", + &mut self.post_revert_up_backoff_max_secs, + ), + scale_down_backoff_max: converter.polls( + "scale_down_backoff_max_secs", + &mut self.scale_down_backoff_max_secs, + ), + } + } } /// One successful action waiting for post-action IOPS validation. @@ -390,6 +446,9 @@ fn iops_rate(start: IoSample, end: IoSample) -> u64 { /// Scaling engine driven by CPU utilisation thresholds. pub struct ThresholdEngine { cfg: ThresholdConfig, + /// `cfg` durations in polls. + polls: ThresholdPolls, + poll_period: Duration, // TODO the controller already maintains this mapping, an alternative would // be for this engine to define a private type and keep a reference to @@ -398,16 +457,25 @@ pub struct ThresholdEngine { } impl ThresholdEngine { - /// Construct an engine from validated settings. - pub fn new(cfg: ThresholdConfig) -> Self { - Self { + /// Validate `cfg` and convert its durations to polls. + /// + /// The engine must not be used before `durations` finished without error. + pub fn new( + mut cfg: ThresholdConfig, + durations: &mut PollConverter, + ) -> Result { + cfg.validate()?; + let polls = cfg.resolve_polls(durations); + Ok(Self { cfg, + polls, + poll_period: durations.period(), state: Mutex::new(HashMap::new()), - } + }) } /// Load `threshold.json`, falling back to built-in defaults when absent. - pub fn from_config_dir(dir: &Path) -> Result { + pub fn from_config_dir(dir: &Path, durations: &mut PollConverter) -> Result { let path = Path::new(&dir.join(format!("{ENGINE_NAME}.json"))); // TODO TOCTOU, blindly load and return default if ENOENT let cfg: ThresholdConfig = if path.exists() { @@ -415,20 +483,20 @@ impl ThresholdEngine { } else { ThresholdConfig::default() }; - cfg.validate()?; - Ok(Self::new(cfg)) + Ok(Self::new(cfg, durations)?) } - /// Return the engine's effective configuration. + /// Return the engine's effective configuration, with any rounded-up + /// durations. pub fn config(&self) -> &ThresholdConfig { &self.cfg } /// Return the validation polls for an action's resulting thread count. fn validation_sample_polls(&self, target_thread_count: u32) -> u32 { - self.cfg.scale_validation_sample_polls.saturating_add( - self.cfg - .scale_validation_sample_polls_per_thread + self.polls.scale_validation_delay.saturating_add( + self.polls + .scale_validation_delay_per_thread .saturating_mul(target_thread_count), ) } @@ -514,11 +582,11 @@ impl ThresholdEngine { /// failed scale-ups. The backoff extends grace so the recovery baseline /// window keeps its configured length. fn post_revert_up_grace_polls(&self, failures: u32) -> u32 { - let recovery = self.cfg.post_revert_up_recovery_polls; + let recovery = self.polls.post_revert_up_recovery; let hold = backoff_polls( - self.cfg.post_revert_up_grace_polls.saturating_add(recovery), + self.polls.post_revert_up_grace.saturating_add(recovery), failures.saturating_sub(1), - self.cfg.post_revert_up_backoff_max_polls, + self.polls.post_revert_up_backoff_max, ); hold.saturating_sub(recovery) } @@ -527,18 +595,20 @@ impl ThresholdEngine { /// consecutive failed scale-downs. fn scale_down_sustain_polls(&self, failures: u32) -> u32 { backoff_polls( - self.cfg.scale_down_sustain_polls, + self.polls.scale_down_sustain, failures, - self.cfg.scale_down_backoff_max_polls, + self.polls.scale_down_backoff_max, ) } - /// Seconds until a post-scale IOPS sample is evaluated. - fn validation_after_secs(&self, scale_poll_secs: f64, target_thread_count: u32) -> f64 { + /// Time until a post-scale IOPS sample is evaluated. + fn validation_after(&self, target_thread_count: u32) -> Duration { // Countdown consumes the effective validation polls, then the next // tick performs the comparison. - // TODO use Duration - (f64::from(self.validation_sample_polls(target_thread_count)) + 1.0) * scale_poll_secs + self.poll_period.saturating_mul( + self.validation_sample_polls(target_thread_count) + .saturating_add(1), + ) } } @@ -580,7 +650,7 @@ fn log_scale_up_decision( target: u32, current_iops: u64, min_target_iops: u64, - validation_after_secs: f64, + validation_after: Duration, ) { tracing::info!( target: "engine", @@ -590,7 +660,7 @@ fn log_scale_up_decision( thr = %format!("{thread_count}->{target}"), current_iops, min_target_iops, - validation_after_secs, + validation_after_secs = validation_after.as_secs(), "threshold scaling decision" ); } @@ -715,7 +785,7 @@ impl ScalingEngine for ThresholdEngine { &instance.id, failures, grace_polls, - self.cfg.post_revert_up_recovery_polls, + self.polls.post_revert_up_recovery, ); } else { instance_state.consecutive_failed_downs = @@ -730,8 +800,8 @@ impl ScalingEngine for ThresholdEngine { return ScaleAction::Revert(target); } - let block_scale_up = instance_state - .advance_post_revert_up_hold(sample, self.cfg.post_revert_up_recovery_polls); + let block_scale_up = + instance_state.advance_post_revert_up_hold(sample, self.polls.post_revert_up_recovery); // Grace ignores metrics; recovery updates the rate inside advance. // Once the hold ends (or if there was none), refresh the normal baseline. if !block_scale_up { @@ -760,13 +830,13 @@ impl ScalingEngine for ThresholdEngine { // or equal to the max_thread_count here. let target = (thread_count + 1).min(context.max_thread_count); let current_iops = instance_state.current_baseline_iops; - let (min_target_iops, validation_after_secs) = if self.cfg.scale_up_min_gain > 0.0 { + let (min_target_iops, validation_after) = if self.cfg.scale_up_min_gain > 0.0 { ( ((current_iops as f64) * (1.0 + self.cfg.scale_up_min_gain)).ceil() as u64, - self.validation_after_secs(context.scale_poll_secs, target), + self.validation_after(target), ) } else { - (current_iops, 0.0) + (current_iops, Duration::ZERO) }; log_scale_up_decision( instance, @@ -775,7 +845,7 @@ impl ScalingEngine for ThresholdEngine { target, current_iops, min_target_iops, - validation_after_secs, + validation_after, ); return ScaleAction::Up(target); } @@ -865,6 +935,7 @@ mod tests { }; use crate::{ backends::BackendClientError, + config::PollPeriod, engines::{AppliedOutcome, EngineTickContext, ScaleAction, ScalingEngine}, instance::{ Instance, InstanceClient, InstancePerfSample, InstanceStatus, ThreadPoolSnapshot, @@ -921,10 +992,28 @@ mod tests { instance } + const POLL_SECS: u64 = 10; + + /// Return the seconds `count` polls span. + fn polls(count: u64) -> u64 { + POLL_SECS * count + } + + /// Build an engine whose durations are converted with `round_up`. + fn try_new_engine(cfg: ThresholdConfig, round_up: bool) -> Result { + let mut durations = PollPeriod::new(POLL_SECS, round_up).converter(); + let engine = ThresholdEngine::new(cfg, &mut durations).map_err(|e| e.to_string())?; + durations.finish().map_err(|e| e.to_string())?; + Ok(engine) + } + + fn new_engine(cfg: ThresholdConfig) -> ThresholdEngine { + try_new_engine(cfg, false).unwrap() + } + fn context_at(now: Instant) -> EngineTickContext { EngineTickContext { now, - scale_poll_secs: 10.0, ..EngineTickContext::default() } } @@ -994,10 +1083,10 @@ mod tests { #[fixture] fn engine() -> ThresholdEngine { - ThresholdEngine::new(ThresholdConfig { + new_engine(ThresholdConfig { scale_up_threshold: 0.8, scale_up_min_gain: 0.05, - scale_validation_sample_polls: 0, + scale_validation_delay_secs: 0, ..Default::default() }) } @@ -1112,12 +1201,12 @@ mod tests { #[tokio::test] async fn test_failed_scale_up_grace_then_recovery_baseline(#[future] instance: Arc) { let instance = instance.await; - let engine = ThresholdEngine::new(ThresholdConfig { + let engine = new_engine(ThresholdConfig { scale_up_threshold: 0.8, scale_up_min_gain: 0.05, - scale_validation_sample_polls: 0, - post_revert_up_grace_polls: 1, - post_revert_up_recovery_polls: 2, + scale_validation_delay_secs: 0, + post_revert_up_grace_secs: polls(1), + post_revert_up_recovery_secs: polls(2), ..Default::default() }); let t0 = Instant::now(); @@ -1193,13 +1282,13 @@ mod tests { #[tokio::test] async fn test_scale_down_allowed_during_post_revert_up_hold(#[future] instance: Arc) { let instance = instance.await; - let engine = ThresholdEngine::new(ThresholdConfig { + let engine = new_engine(ThresholdConfig { scale_up_threshold: 0.8, scale_up_min_gain: 0.05, - scale_validation_sample_polls: 0, - scale_down_sustain_polls: 1, - post_revert_up_grace_polls: 2, - post_revert_up_recovery_polls: 2, + scale_validation_delay_secs: 0, + scale_down_sustain_secs: polls(1), + post_revert_up_grace_secs: polls(2), + post_revert_up_recovery_secs: polls(2), ..Default::default() }); let t0 = Instant::now(); @@ -1240,11 +1329,11 @@ mod tests { #[tokio::test] async fn scale_down_keeps_shrink_when_util_stays_low(#[future] instance: Arc) { let instance = instance.await; - let engine = ThresholdEngine::new(ThresholdConfig { + let engine = new_engine(ThresholdConfig { scale_up_threshold: 0.8, scale_down_revert_drop: 0.05, - scale_validation_sample_polls: 0, - scale_down_sustain_polls: 1, + scale_validation_delay_secs: 0, + scale_down_sustain_secs: polls(1), ..Default::default() }); let t0 = Instant::now(); @@ -1280,10 +1369,10 @@ mod tests { #[future] instance: Arc, ) { let instance = instance.await; - let engine = ThresholdEngine::new(ThresholdConfig { + let engine = new_engine(ThresholdConfig { scale_up_threshold: 0.8, scale_down_revert_drop: 0.05, - scale_validation_sample_polls: 0, + scale_validation_delay_secs: 0, ..Default::default() }); let t0 = Instant::now(); @@ -1364,13 +1453,13 @@ mod tests { #[tokio::test] async fn test_failed_scale_ups_back_off_until_kept(#[future] instance: Arc) { let instance = instance.await; - let engine = ThresholdEngine::new(ThresholdConfig { + let engine = new_engine(ThresholdConfig { scale_up_threshold: 0.8, scale_up_min_gain: 0.05, - scale_validation_sample_polls: 0, - post_revert_up_grace_polls: 1, - post_revert_up_recovery_polls: 2, - post_revert_up_backoff_max_polls: 12, + scale_validation_delay_secs: 0, + post_revert_up_grace_secs: polls(1), + post_revert_up_recovery_secs: polls(2), + post_revert_up_backoff_max_secs: polls(12), ..Default::default() }); engine.on_instance_added(&instance).await; @@ -1409,12 +1498,12 @@ mod tests { #[tokio::test] async fn test_failed_scale_downs_back_off_until_scale_up(#[future] instance: Arc) { let instance = instance.await; - let engine = ThresholdEngine::new(ThresholdConfig { + let engine = new_engine(ThresholdConfig { scale_up_threshold: 0.8, scale_down_revert_drop: 0.05, - scale_validation_sample_polls: 0, - scale_down_sustain_polls: 3, - scale_down_backoff_max_polls: 20, + scale_validation_delay_secs: 0, + scale_down_sustain_secs: polls(3), + scale_down_backoff_max_secs: polls(20), ..Default::default() }); engine.on_instance_added(&instance).await; @@ -1509,9 +1598,9 @@ mod tests { #[test] fn test_validation_polls_increase_with_target_thread_count() { - let engine = ThresholdEngine::new(ThresholdConfig { - scale_validation_sample_polls: 2, - scale_validation_sample_polls_per_thread: 3, + let engine = new_engine(ThresholdConfig { + scale_validation_delay_secs: polls(2), + scale_validation_delay_per_thread_secs: polls(3), ..Default::default() }); @@ -1547,7 +1636,7 @@ mod tests { set_observation(&instance, 5, 0.88, 220_000).await; - for i in 0..engine.cfg.scale_validation_sample_polls { + for i in 0..engine.polls.scale_validation_delay { assert_eq!( engine .evaluate( @@ -1585,10 +1674,10 @@ mod tests { #[future] instance: Arc, ) { let instance = instance.await; - let engine = ThresholdEngine::new(ThresholdConfig { + let engine = new_engine(ThresholdConfig { scale_up_threshold: 0.8, scale_up_min_gain: 0.05, - scale_validation_sample_polls: 2, + scale_validation_delay_secs: polls(2), ..Default::default() }); let t0 = Instant::now(); @@ -1609,7 +1698,7 @@ mod tests { .await; set_observation(&instance, 5, 0.88, 140_000).await; - for i in 0..engine.cfg.scale_validation_sample_polls { + for i in 0..engine.polls.scale_validation_delay { assert_eq!( engine .evaluate( @@ -1670,19 +1759,64 @@ mod tests { #[test] fn missing_config_file_uses_defaults() { let dir = tempfile::tempdir().unwrap(); - let engine = - ThresholdEngine::from_config_dir(&Path::new(dir.path().to_str().unwrap())).unwrap(); + let mut durations = PollPeriod::new(POLL_SECS, false).converter(); + let engine = ThresholdEngine::from_config_dir( + &Path::new(dir.path().to_str().unwrap()), + &mut durations, + ) + .unwrap(); + durations.finish().unwrap(); assert_eq!(engine.name(), super::ENGINE_NAME); let _ = engine.config(); } + /// Tests that the default durations are multiples of the default + /// controller poll period. + #[test] + fn test_defaults_are_multiples_of_default_poll_period() { + let mut durations = crate::config::Config::default().poll_period().converter(); + ThresholdEngine::new(ThresholdConfig::default(), &mut durations).unwrap(); + durations.finish().unwrap(); + } + + /// Tests that every duration that is not a multiple of the poll period is + /// reported in a single error. + #[test] + fn test_durations_not_poll_multiples_are_all_rejected() { + let cfg = ThresholdConfig { + scale_down_sustain_secs: 25, + post_revert_up_grace_secs: 5, + ..Default::default() + }; + let err = try_new_engine(cfg, false).err().unwrap(); + assert!(err.contains("scale_down_sustain_secs (25)"), "{err}"); + assert!(err.contains("post_revert_up_grace_secs (5)"), "{err}"); + } + + /// Tests that rounding converts durations to the next whole poll and + /// that the effective configuration shows the rounded values. + #[test] + fn test_rounded_durations_are_used_and_dumped() { + let cfg = ThresholdConfig { + scale_down_sustain_secs: 25, + scale_validation_delay_secs: 1, + ..Default::default() + }; + let engine = try_new_engine(cfg, true).unwrap(); + assert_eq!(engine.polls.scale_down_sustain, 3); + assert_eq!(engine.polls.scale_validation_delay, 1); + let dumped = engine.dump_config(); + assert_eq!(dumped["scale_down_sustain_secs"], 30); + assert_eq!(dumped["scale_validation_delay_secs"], 10); + } + /// Test that CPU utilisation below scale-up/down thresholds yields /// Hold/`None`. #[rstest] #[tokio::test] async fn evaluate_holds_when_util_below_thresholds(#[future] instance: Arc) { let instance = instance.await; - let engine = ThresholdEngine::new(ThresholdConfig::default()); + let engine = new_engine(ThresholdConfig::default()); let action = engine .evaluate(&instance, &context_at(Instant::now())) .await; diff --git a/src/main.rs b/src/main.rs index feaff3f0..05b42bd5 100644 --- a/src/main.rs +++ b/src/main.rs @@ -5,7 +5,7 @@ //! `io-thread-controller` daemon entry point. -use std::io::ErrorKind; +use std::{io::ErrorKind, process::ExitCode}; use clap::{self, CommandFactory, FromArgMatches, Parser}; use io_thread_controller::{ @@ -13,6 +13,7 @@ use io_thread_controller::{ backends::registered_backends, config::{Config, ConfigError, dump_default_config, load_json_with_includes, validate_config}, daemon::{DaemonError, VERSION, run}, + engines::{EngineError, load_registered_engine}, per_vm_log::{self, PER_VM_LOG_DIR, PerVmLog}, util::Path, }; @@ -34,6 +35,9 @@ enum IoThreadControllerError { #[error(transparent)] Daemon(#[from] DaemonError), + #[error(transparent)] + Engine(#[from] EngineError), + #[error("no backend `{0}`")] NoSuchBackend(String), } @@ -83,6 +87,10 @@ struct Cli { /// Log scaling decisions without calling set_thread_count. #[arg(long)] dry_run: bool, + /// Round durations that are not multiples of `scale_poll_secs` up to the + /// next multiple, with a warning, instead of refusing to start. + #[arg(long)] + round_up_durations: bool, } /// See `Cli::log_style`. @@ -95,7 +103,17 @@ enum LogStyle { } #[tokio::main] -async fn main() -> Result<(), IoThreadControllerError> { +async fn main() -> ExitCode { + match run_daemon().await { + Ok(()) => ExitCode::SUCCESS, + Err(error) => { + eprintln!("error: {error}"); + ExitCode::FAILURE + } + } +} + +async fn run_daemon() -> Result<(), IoThreadControllerError> { let bootstrap_cfg = Config::default(); let bootstrap_backends = registered_backends(&bootstrap_cfg)?; let mut command = Cli::command(); @@ -140,13 +158,18 @@ async fn main() -> Result<(), IoThreadControllerError> { if cli.dry_run { cfg.dry_run = true; } - validate_config(&cfg)?; + if cli.round_up_durations { + cfg.round_up_durations = true; + } + let mut durations = validate_config(&mut cfg)?; + let engine = load_registered_engine(&cfg.engine_config_dir, &cfg.engine, &mut durations)?; + durations.finish()?; per_vm_log.set_dir( (cfg.enable_per_vm_status_line && !cfg.enable_per_vm_main_log) .then(|| Path::from(PER_VM_LOG_DIR)), ); let backends = registered_backends(&cfg)?; - run(cfg, backends).await?; + run(cfg, engine, backends).await?; Ok(()) }