From 18af41d362679b24350dc032f2f23c94b08027a4 Mon Sep 17 00:00:00 2001 From: Leonardo Forchini Date: Thu, 15 Oct 2026 12:07:09 +0100 Subject: [PATCH] Pass immutable data to ScalingEngines Fixes #89, #90 The controller constructs a struct from tracking mutable state and backend collected performance counters which it then passes to the engine. Previously the engine was given a reference to the Instance itself, potentially allowing the engine to bypass the controller. Now, only the controller can command the Instance with the engine advising the controller. This split also makes testing easier. Instead of having to construct full Instances, now only fake data needs to be passed to the engine to evaluate its decisions. Also, the controller now derives thread count and utilisation from procfs instead of relying on the backend. Signed-off-by: Leonardo Forchini Co-authored-by: Cursor --- src/backends/fake/mod.rs | 14 +- src/backends/qemu/client.rs | 34 +-- src/backends/qemu/helpers.rs | 96 +------ src/backends/qemu/mod.rs | 19 +- src/bin/mockfsd.rs | 47 +++ src/controller.rs | 300 ++++++++++++-------- src/engines/mod.rs | 86 +++--- src/engines/threshold.rs | 362 ++++++++++-------------- src/instance.rs | 128 +++------ src/rolling.rs | 43 --- tests/component/conftest.py | 12 +- tests/component/harness/fake_backend.py | 79 +++++- tests/component/harness/mockfs.py | 10 +- tests/component/test_controller.py | 8 +- 14 files changed, 577 insertions(+), 661 deletions(-) diff --git a/src/backends/fake/mod.rs b/src/backends/fake/mod.rs index 65568a6e..1a426163 100644 --- a/src/backends/fake/mod.rs +++ b/src/backends/fake/mod.rs @@ -87,7 +87,6 @@ impl FakeConfig { struct VmFile { thread_count: u32, vcpu_count: u32, - per_thread_util: f64, /// First-snapshot ownership. `true` lets the controller scale the VM. managed: bool, fail_snapshot: bool, @@ -101,7 +100,6 @@ impl Default for VmFile { Self { thread_count: 1, vcpu_count: 1, - per_thread_util: 0.0, managed: true, fail_snapshot: false, read_io_count: 0, @@ -162,6 +160,7 @@ fn discover_vms(state_dir: &Path) -> Vec> { let Some(vm) = read_vm(&path, id) else { continue; }; + let pid = fake_pid(id); let client = FakeClient { vm_path: path.clone(), calls_path: state_dir.join("calls.json"), @@ -170,7 +169,7 @@ fn discover_vms(state_dir: &Path) -> Vec> { instances.push(Arc::new(Instance::new( id.to_string(), Path::new(id), - 0, + pid, client, ))); } @@ -191,6 +190,13 @@ struct FakeClient { managed: bool, } +fn fake_pid(id: &str) -> i32 { + let hash = id.bytes().fold(0u32, |hash, byte| { + hash.wrapping_mul(31).wrapping_add(u32::from(byte)) + }); + 10_000 + (hash % 1_000_000) as i32 +} + /// `.json` shares `.lock` with the pytest harness. fn lock_path_for(vm_path: &std::path::Path) -> std::path::PathBuf { vm_path.with_extension("lock") @@ -267,7 +273,6 @@ impl InstanceClient for FakeClient { ))); } Ok(ThreadPoolSnapshot { - thread_count: vm.thread_count, vcpu_count: vm.vcpu_count, perf: Some(InstancePerfSample { read_io_count: vm.read_io_count, @@ -277,7 +282,6 @@ impl InstanceClient for FakeClient { write_bytes_total: 0, ..Default::default() }), - per_thread_util: Some(vm.per_thread_util), }) } diff --git a/src/backends/qemu/client.rs b/src/backends/qemu/client.rs index c2a339b8..8263e479 100644 --- a/src/backends/qemu/client.rs +++ b/src/backends/qemu/client.rs @@ -12,8 +12,7 @@ //! * the [`super::libvirt::LibvirtQmp`] transport (QMP multiplexed through //! libvirt's `virDomainQemuMonitorCommand`); //! * the per-VM topology cache (iothread ids + TIDs, virtio-scsi device path, -//! virtqueue count); -//! * the per-tick CPU sample used to compute per-IOThread utilisation. +//! virtqueue count). //! //! Everything the controller used to keep in `QemuInstance` / //! `QemuState` now lives here, behind the generic per-VM interface. @@ -29,9 +28,7 @@ use crate::{ backends::{ BackendClientError, IoThreadProperties, VqMapping, qemu::{ - helpers::{ - CpuSample, QemuError, QemuTopology, next_qemu_iothread_id, round_robin_vq_mapping, - }, + helpers::{QemuError, QemuTopology, next_qemu_iothread_id, round_robin_vq_mapping}, libvirt::LibvirtQmp, }, }, @@ -69,8 +66,6 @@ struct Inner { /// Number of command virtqueues exposed by the device. vq_count: u16, - /// Previous per-task CPU sample used to calculate utilisation. - last_sample: Option, /// Whether we've already warned that prometheus metrics are unsupported. warned_unsupported_metrics: bool, } @@ -180,10 +175,7 @@ impl QemuInstanceClient { }; Ok(ThreadPoolSnapshot { - thread_count, perf, - // Cannot derive per-thread util without fresh topology/TID data. - per_thread_util: None, vcpu_count: self.vcpu_count, }) } @@ -271,13 +263,6 @@ impl InstanceClient for QemuInstanceClient { "invalid pid {pid} from tid {first_tid}" ))); } - let sample = CpuSample::new(pid, &topo.iothread_tids)?; - if sample.jiffies.is_empty() { - return Err(BackendClientError::InvalidState( - "qemu per-thread sample empty".into(), - )); - } - let mut s = self.inner.lock().await; s.pid = pid; s.iothreads = topo.iothreads.clone(); @@ -285,17 +270,6 @@ impl InstanceClient for QemuInstanceClient { s.device_path = topo.device_path.clone(); s.vq_count = topo.vq_count; - let per_thread_util = if let Some(prev) = s.last_sample.take() { - let util = &sample - &prev; - s.last_sample = Some(sample); - Some(util) - } else { - // Warm-up tick: report None so the engine leaves - // per_thread_util at 0.0. Next tick has a `prev` - // sample and returns a real delta. - s.last_sample = Some(sample); - None - }; // Feed block statistics to the engine so // revert-on-drop validation compares apples to apples // across backends. A best-effort call: an error here @@ -320,11 +294,9 @@ impl InstanceClient for QemuInstanceClient { { perf.per_vq_depth = Some(topo.vq_inflight.clone()); } - let thread_count = topo.iothreads.len() as u32; + Ok(ThreadPoolSnapshot { - thread_count, perf, - per_thread_util, vcpu_count: self.vcpu_count, }) } diff --git a/src/backends/qemu/helpers.rs b/src/backends/qemu/helpers.rs index cd898990..49a975ec 100644 --- a/src/backends/qemu/helpers.rs +++ b/src/backends/qemu/helpers.rs @@ -6,8 +6,8 @@ //! Shared QEMU helpers. //! //! Everything here is transport-neutral: prometheus-body topology -//! parsing, /proc jiffies sampling, VQ round-robin, iothread id -//! allocation, and the [`QemuError`] shape used by the QMP +//! parsing, VQ round-robin, iothread id allocation, and the [`QemuError`] +//! shape used by the QMP //! transport in [`crate::backends::qemu::libvirt`]. The libvirt- //! backed transport lives in `backends/qemu/libvirt.rs`; the //! per-VM state that used to live in this module now belongs to @@ -16,23 +16,16 @@ use std::{ collections::{BTreeMap, HashMap, HashSet}, io, - ops::Sub, sync::{ LazyLock, atomic::{AtomicU64, Ordering}, }, - time::Instant, }; -use procfs::process::Process; use regex::Regex; -use rustix::param; use thiserror::Error; -use crate::{ - backends::{BackendClientError, VqMapping}, - util::Path, -}; +use crate::backends::{BackendClientError, VqMapping}; /// Errors produced while querying or reconfiguring a QEMU vm. #[derive(Debug, Error)] @@ -198,89 +191,6 @@ impl QemuTopology { } } -// --------------------------------------------------------------------- -// /proc helpers -// --------------------------------------------------------------------- - -/// Per-IOThread cumulative CPU counters sampled at one wall-clock instant. -#[derive(Debug, Clone)] -pub struct CpuSample { - /// Cumulative user-plus-system jiffies keyed by host task ID. - pub jiffies: HashMap, - /// Wall-clock time at which the counters were sampled. - // FIXME rename to sampled_at - pub wall: Instant, -} - -impl Default for CpuSample { - fn default() -> Self { - Self { - jiffies: HashMap::new(), - wall: Instant::now(), - } - } -} - -impl CpuSample { - /// Sample cumulative CPU jiffies for every discovered IOThread task. - pub fn new( - pid: i32, - iothread_tids: &BTreeMap, - ) -> Result { - let mut out = Self { - jiffies: HashMap::new(), - wall: Instant::now(), - }; - - let p = Process::new_with_root(Path::proc_pid(pid))?; - for &tid in iothread_tids.values() { - let stat = p.task_from_tid(tid)?.stat()?; - out.jiffies.insert(tid, stat.utime + stat.stime); - } - Ok(out) - } - - /// Compute average per-IOThread CPU utilisation between two samples. - // TODO add unit test for this function and the call site, and that it can be - // folded into the Sub. - pub fn diff_jiffies_sample(prev: &Self, cur: &Self) -> f64 { - let Some(elapsed) = cur.wall.checked_duration_since(prev.wall) else { - return 0.0; - }; - let elapsed = elapsed.as_secs_f64(); - if elapsed <= 0.0 { - return 0.0; - } - let hz = param::clock_ticks_per_second() as f64; - assert!(hz > 0.0); - let mut total = 0.0; - let mut counted = 0u32; - for (&tid, &after) in &cur.jiffies { - let Some(&before) = prev.jiffies.get(&tid) else { - continue; - }; - if after < before { - continue; - } - let delta = (after - before) as f64; - let frac = ((delta / hz) / elapsed).clamp(0.0, 1.0); - total += frac; - counted += 1; - } - if counted == 0 { - return 0.0; - } - total / counted as f64 - } -} - -impl Sub<&CpuSample> for &CpuSample { - type Output = f64; - fn sub(self, rhs: &CpuSample) -> Self::Output { - CpuSample::diff_jiffies_sample(rhs, self) - } -} - // --------------------------------------------------------------------- // VQ mapping helpers // --------------------------------------------------------------------- diff --git a/src/backends/qemu/mod.rs b/src/backends/qemu/mod.rs index 6d9f73e9..a2afa3e9 100644 --- a/src/backends/qemu/mod.rs +++ b/src/backends/qemu/mod.rs @@ -32,7 +32,7 @@ use serde::{Deserialize, Serialize}; use crate::{ backends::{Backend, BackendClientError, BackendRegistration}, config::{self, ConfigError}, - instance::{Instance, InstanceClient}, + instance::{Instance, InstanceClient, ThreadNameFilter}, util::Path, }; @@ -153,10 +153,7 @@ impl Backend for QemuBackend { /// in a generic [`Instance`] whose wire client is a /// [`client::QemuInstanceClient`]. We seed /// [`Instance::pid`] with libvirt's per-VM pidfile so the - /// engine's cgroup / procfs sampling kicks in immediately; - /// the QEMU backend still supplies its own per-thread util on - /// every snapshot so the /proc walker's thread-name filter is - /// never consulted for this transport. + /// controller's cgroup and procfs sampling kicks in immediately. async fn discover(&self) -> Vec> { let conn = match self.get_or_open_conn(&self.cfg.libvirt_uri).await { Some(c) => c, @@ -208,12 +205,12 @@ impl Backend for QemuBackend { // it is purely informational (log context), so we // stuff the libvirt URI in there. let sock_path = Path::new(&self.cfg.libvirt_uri); - out.push(Arc::new(Instance::new( - uuid.to_string(), - sock_path, - pid, - client, - ))); + let filter = ThreadNameFilter::new("^IO iothread", &[]) + .expect("static QEMU IOThread regex must compile"); + out.push(Arc::new( + Instance::new(uuid.to_string(), sock_path, pid, client) + .with_thread_name_filter(filter), + )); } out } diff --git a/src/bin/mockfsd.rs b/src/bin/mockfsd.rs index 4d6ac055..b1dfc435 100644 --- a/src/bin/mockfsd.rs +++ b/src/bin/mockfsd.rs @@ -111,6 +111,42 @@ impl Inner { self.callbacks.insert(parent, callback); } + /// Attempt to unregister a file, returns true if successful. + fn unregister(&mut self, name: &Path) -> bool { + assert!(name.is_absolute()); + let mut parent = INodeNo::ROOT; + let mut target = None; + for part in name.components() { + let Component::Normal(part) = part else { + continue; + }; + let Some(inode) = self.children.get(&(parent, part.to_owned())).copied() else { + return false; + }; + target = Some((parent, part.to_owned(), inode)); + parent = inode; + } + let Some((target_parent, target_name, target_inode)) = target else { + return false; + }; + let mut removed = vec![target_inode]; + let mut index = 0; + while index < removed.len() { + let inode = removed[index]; + removed.extend( + self.children + .iter() + .filter_map(|((parent, _), child)| (*parent == inode).then_some(*child)), + ); + index += 1; + } + self.children + .retain(|(parent, _), child| !removed.contains(parent) && !removed.contains(child)); + self.children.remove(&(target_parent, target_name)); + self.callbacks.retain(|inode, _| !removed.contains(inode)); + true + } + /// Get child [`INodeNo`] given its name and parents [`INodeNo`]. fn lookup(&self, parent: INodeNo, name: &OsStr) -> Option { self.children.get(&(parent, name.to_owned())).copied() @@ -312,6 +348,11 @@ impl MockFsHandle { self.register_iterator(name, values); Ok(()) } + + fn unregister(&self, name: &Path) -> Result { + validate_registration_path(name)?; + Ok(self.inner.lock().expect("mockfs lock").unregister(name)) + } } /// Ensure a given [`Path`] is well-formed: @@ -648,6 +689,12 @@ impl Control { .map_err(invalid_args) } + async fn unregister(&self, mount_path: String, path: String) -> zbus::fdo::Result { + self.mounted_handle(&mount_path)? + .unregister(Path::new(&path)) + .map_err(invalid_args) + } + // TODO register callback, how to implement? async fn ping(&self) -> String { diff --git a/src/controller.rs b/src/controller.rs index d28cbcc0..97b04256 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -24,9 +24,10 @@ use crate::{ config::Config, dbus::DbusRequest, engines::{ - AppliedOutcome, BlockedReason, EngineTickContext, PsiSample, ScaleAction, ScalingEngine, + AppliedOutcome, BlockedReason, EngineInstanceSnapshot, EngineTickContext, PsiSample, + ScaleAction, ScalingEngine, }, - instance::{CgroupError, Instance, InstanceStatus, SnapshotLatency}, + instance::{CgroupError, Instance, InstanceStatus, SnapshotLatency, ThreadPoolSnapshot}, rolling::format_1_5_15, state::{StateError, VmOwnership, VmStateStore}, util::Path, @@ -102,8 +103,10 @@ pub struct SnapshotVm { } /// Average, median, and aggregate utilisation across sampled workers. -#[derive(Debug, Clone, Copy)] +#[derive(Debug, Default, Clone, Copy)] struct CpuStats { + /// Mean per-worker utilisation as a 0..1 fraction. + avg_util: f64, /// Mean per-worker utilisation in percent. avg_pct: u64, /// Median per-worker utilisation in percent. @@ -112,22 +115,20 @@ struct CpuStats { total_pct: u64, } -/// Summarise per-worker utilisation, falling back to aggregate samples. -fn compute_cpu_stats(s: &crate::instance::InstanceStatus) -> CpuStats { - if !s.per_worker_util.is_empty() { - let total = s.per_worker_util.iter().sum::(); - let median = median(&s.per_worker_util); - return CpuStats { - avg_pct: (total / s.per_worker_util.len() as f64 * 100.0).round() as u64, - median_pct: (median * 100.0).round() as u64, - total_pct: (total * 100.0).round() as u64, - }; +/// Summarise controller-sampled worker utilisation. +fn compute_cpu_stats(status: &InstanceStatus) -> CpuStats { + if status.per_worker_util.is_empty() { + return CpuStats::default(); } + let total = status.per_worker_util.iter().sum::(); + let median = median(&status.per_worker_util); + let avg_util = total / status.per_worker_util.len() as f64; CpuStats { - avg_pct: (s.per_thread_util * 100.0).round() as u64, - median_pct: (s.per_thread_util * 100.0).round() as u64, - total_pct: (s.per_thread_util * s.thread_count as f64 * 100.0).round() as u64, + avg_util, + avg_pct: (avg_util * 100.0).round() as u64, + median_pct: (median * 100.0).round() as u64, + total_pct: (total * 100.0).round() as u64, } } @@ -264,10 +265,8 @@ impl Controller { continue; } let classification = self.vm_ownership.classification(&instance.id); - let mut status = instance.status.write().await; - status.ownership_classification = classification; - drop(status); - self.engine.on_instance_added(&instance).await; + instance.status.write().await.ownership_classification = classification; + self.engine.on_instance_added(&instance.id).await; self.instances.insert(instance.id.clone(), instance); added += 1; } @@ -335,8 +334,15 @@ impl Controller { DbusRequest::GetStats { reply } => { let mut vms = Vec::with_capacity(self.instances.len()); for (id, instance) in &self.instances { + let snapshot = instance.snapshot.read().await; let status = instance.status.read().await; - let (read_io_count, write_io_count, other_io_count) = match &status.perf { + let cpu = compute_cpu_stats(&status); + let thread_count = status.thread_count; + let vcpu_count = snapshot.as_ref().map_or(0, |snapshot| snapshot.vcpu_count); + let (read_io_count, write_io_count, other_io_count) = match snapshot + .as_ref() + .and_then(|snapshot| snapshot.perf.as_ref()) + { Some(perf) => { (perf.read_io_count, perf.write_io_count, perf.other_io_count) } @@ -344,11 +350,11 @@ impl Controller { }; vms.push(serde_json::json!({ "vm": id, - "thread_count": status.thread_count, + "thread_count": thread_count, "manual_scaling_sticky": status.manual_scaling_sticky, "scaling_allowed": status.scaling_allowed, - "vcpu_count": status.vcpu_count, - "per_thread_util": status.per_thread_util, + "vcpu_count": vcpu_count, + "per_thread_util": cpu.avg_util, // FIXME omit if status.perf.is_none()? "read_io_count": read_io_count, "write_io_count": write_io_count, @@ -429,7 +435,12 @@ impl Controller { .get(vm) .cloned() .ok_or_else(|| ControllerError::VmError("unknown VM {vm}".to_string()))?; - let vcpu_count = instance.status.read().await.vcpu_count; + let vcpu_count = instance + .snapshot + .read() + .await + .as_ref() + .map_or(0, |snapshot| snapshot.vcpu_count); if threads == 0 { return Err(ControllerError::ThreadCountError( "target thread count must be greater than zero".to_string(), @@ -442,9 +453,7 @@ impl Controller { } instance.client.set_thread_count(threads).await?; - let mut status = instance.status.write().await; - status.thread_count = threads; - status.manual_scaling_sticky = sticky; + instance.status.write().await.manual_scaling_sticky = sticky; Ok(()) } @@ -486,43 +495,33 @@ impl Controller { let ids: Vec<_> = self.instances.keys().collect(); for id in ids { let instance = &self.instances[id]; - let s = instance.status.read().await; - let cpu_stats = compute_cpu_stats(&s); - let (num_queues, qd_total, qd_avg_x100, qd_median) = - summarize_vq_depths(s.per_vq_depth.as_deref()); + let snapshot = instance.snapshot.read().await; + let status = instance.status.read().await; + let cpu_stats = compute_cpu_stats(&status); + let thread_count = status.thread_count; + let (vcpu_count, perf) = snapshot.as_ref().map_or((0, None), |snapshot| { + (snapshot.vcpu_count, snapshot.perf.as_ref()) + }); + let per_vq_depth = perf.and_then(|perf| perf.per_vq_depth.as_deref()); + let (num_queues, qd_total, qd_avg_x100, qd_median) = summarize_vq_depths(per_vq_depth); payload.vms.push(SnapshotVm { id: id.clone(), - vcpu_count: s.vcpu_count, + vcpu_count, pid: instance.pid, - thread_count: s.thread_count, - alive: s.alive, - perf_available: s.perf.is_some(), - // Guard JSON serialization: NaN/inf are not valid JSON numbers. - per_thread_util: if s.per_thread_util.is_finite() { - s.per_thread_util - } else { - 0.0 - }, - manual_scaling_sticky: s.manual_scaling_sticky, - read_iops: s.read_iops, - write_iops: s.write_iops, - other_iops: s.other_iops, - read_bw_bps: s.read_bytes_per_second, - write_bw_bps: s.write_bytes_per_second, - read_latency_us: s.read_latency_us.map(|latency| SnapshotLatency { - p50: latency.p50, - p95: latency.p95, - p99: latency.p99, - avg: latency.avg, - }), - write_latency_us: s.write_latency_us.map(|latency| SnapshotLatency { - p50: latency.p50, - p95: latency.p95, - p99: latency.p99, - avg: latency.avg, - }), + thread_count, + alive: status.alive, + perf_available: perf.is_some(), + per_thread_util: cpu_stats.avg_util, + manual_scaling_sticky: status.manual_scaling_sticky, + read_iops: status.read_iops, + write_iops: status.write_iops, + other_iops: status.other_iops, + read_bw_bps: status.read_bytes_per_second, + write_bw_bps: status.write_bytes_per_second, + read_latency_us: perf.and_then(|perf| perf.read_latency_us), + write_latency_us: perf.and_then(|perf| perf.write_latency_us), num_queues, - per_vq_depth: s.per_vq_depth.clone(), + per_vq_depth: per_vq_depth.map(<[u64]>::to_vec), qd_total, qd_avg_x100, qd_median, @@ -534,6 +533,34 @@ impl Controller { serde_json::to_string(&payload).unwrap_or_else(|e| format!("{{\"error\":\"{e}\"}}")) } + /// Convert one coherent raw snapshot/status pair into engine policy input. + async fn engine_snapshot(instance: &Instance) -> EngineInstanceSnapshot { + let snapshot = instance.snapshot.read().await; + let status = instance.status.read().await; + let cpu = compute_cpu_stats(&status); + let thread_count = status.thread_count; + let (vcpu_count, io_count_total) = snapshot.as_ref().map_or((0, 0), |snapshot| { + ( + snapshot.vcpu_count, + snapshot + .perf + .as_ref() + .map_or(0, |perf| perf.total_io_count()), + ) + }); + EngineInstanceSnapshot { + id: instance.id.clone(), + thread_count, + vcpu_count, + per_thread_util: cpu.avg_util, + cpu_available: !status.per_worker_util.is_empty(), + io_count_total, + throttled_recently: status.throttled_usec_delta > 0, + scaling_allowed: status.scaling_allowed, + manual_scaling_sticky: status.manual_scaling_sticky, + } + } + /// Refresh the fleet, evaluate one coherent plan, and apply eligible work. pub async fn tick(&mut self) -> Result<(), ControllerError> { self.tick_index = self.tick_index.wrapping_add(1); @@ -578,7 +605,12 @@ impl Controller { let fleet: Vec<_> = self.instances.values().cloned().collect(); self.persist_new_classifications(&fleet).await?; - let decisions = self.engine.evaluate_fleet(&fleet, &context).await; + let engine_snapshots = + join_all(fleet.iter().map(|instance| Self::engine_snapshot(instance))).await; + let decisions = self + .engine + .evaluate_fleet(&engine_snapshots, &context) + .await; let mut planned_ids = HashSet::new(); let mut plan = Vec::new(); for (sequence, item) in decisions.into_iter().enumerate() { @@ -602,8 +634,7 @@ impl Controller { let Some(target) = action.target() else { continue; }; - let status = instance.status.read().await; - let current = status.thread_count; + let current = instance.status.read().await.thread_count; if target == current { continue; } @@ -671,17 +702,19 @@ impl Controller { return Ok(false); }; + let snapshot = instance.snapshot.read().await; let status = instance.status.read().await; let previous_count = status.thread_count; - let previous_io_count = match &status.perf { - Some(perf) => perf.total_io_count(), - None => 0, - }; + let previous_io_count = snapshot + .as_ref() + .and_then(|snapshot| snapshot.perf.as_ref()) + .map_or(0, |perf| perf.total_io_count()); let sticky = status.manual_scaling_sticky; let scaling_allowed = status.scaling_allowed(); - let vcpu_count = status.vcpu_count; + let vcpu_count = snapshot.as_ref().map_or(0, |snapshot| snapshot.vcpu_count); let cooldown_until = status.cooldown_until; drop(status); + drop(snapshot); if sticky { self.report_blocked_scale(&instance.id, action, BlockedReason::ManualOverride) @@ -854,7 +887,6 @@ impl Controller { Ok(()) => { let applied_at = Instant::now(); let mut status = instance.status.write().await; - status.thread_count = target; if matches!( action, ScaleAction::Up(_) | ScaleAction::Down(_) | ScaleAction::Revert(_) @@ -937,6 +969,7 @@ impl Controller { let mut fleet: Vec<_> = self.instances.values().collect(); fleet.sort_unstable_by(|left, right| left.id.cmp(&right.id)); for instance in fleet { + let snapshot = instance.snapshot.read().await; let status = instance.status.read().await; let iops_windows = STATUS_WINDOWS.map(|window| status.rolling.iops_over(window)); for (index, value) in iops_windows.iter().enumerate() { @@ -946,12 +979,12 @@ 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); + Self::emit_instance_status_main(instance, &status, iops_windows, snapshot.as_ref()); } 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, snapshot.as_ref()); } if self.cfg.enable_per_vm_main_log { - Self::emit_instance_status_main(instance, &status, iops_windows); + Self::emit_instance_status_main(instance, &status, iops_windows, snapshot.as_ref()); } total_threads += u64::from(status.thread_count); } @@ -972,8 +1005,10 @@ impl Controller { fn format_instance_status_line( status: &InstanceStatus, iops_windows: [Option; 3], + snapshot: Option<&ThreadPoolSnapshot>, ) -> String { let cpu = compute_cpu_stats(status); + let perf = snapshot.and_then(|snapshot| snapshot.perf.as_ref()); format!( "status={} {}/{}/{} {} {}/{} {}/{}/{} {} {} {}", status.thread_count, @@ -986,8 +1021,8 @@ impl Controller { cpu.avg_pct, cpu.median_pct, cpu.total_pct, - format_latency_cell(status.read_latency_us), - format_latency_cell(status.write_latency_us), + format_latency_cell(perf.and_then(|perf| perf.read_latency_us)), + format_latency_cell(perf.and_then(|perf| perf.write_latency_us)), format_1_5_15(&status.rolling, |rolling, window| rolling .cpu_us_per_io_over(window)), ) @@ -998,10 +1033,11 @@ impl Controller { instance: &Instance, status: &InstanceStatus, iops_windows: [Option; 3], + snapshot: Option<&ThreadPoolSnapshot>, ) { self.log_per_vm_line( &instance.id, - &Self::format_instance_status_line(status, iops_windows), + &Self::format_instance_status_line(status, iops_windows, snapshot), ); } @@ -1010,20 +1046,23 @@ impl Controller { instance: &Instance, status: &InstanceStatus, iops_windows: [Option; 3], + snapshot: Option<&ThreadPoolSnapshot>, ) { + let thread_count = status.thread_count; + let perf = snapshot.and_then(|snapshot| snapshot.perf.as_ref()); tracing::info!( target: "controller", event = "vm_status", vm = instance.to_string(), - thread_count = status.thread_count, + thread_count, read_iops = status.read_iops, write_iops = status.write_iops, other_iops = status.other_iops, 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(perf.and_then(|perf| perf.read_latency_us)), + write_latency_us = %format_latency_cell(perf.and_then(|perf| perf.write_latency_us)) ); } @@ -1206,6 +1245,28 @@ mod tests { #[cfg(feature = "threshold-engine")] use crate::engines::threshold::{ThresholdConfig, ThresholdEngine}; + #[test] + fn cpu_stats_summarize_procfs_workers() { + let status = InstanceStatus { + per_worker_util: vec![0.25, 0.75], + ..Default::default() + }; + let stats = compute_cpu_stats(&status); + assert_eq!(stats.avg_util, 0.5); + assert_eq!(stats.avg_pct, 50); + assert_eq!(stats.median_pct, 50); + assert_eq!(stats.total_pct, 100); + } + + #[test] + fn cpu_stats_are_zero_before_a_complete_procfs_interval() { + let stats = compute_cpu_stats(&InstanceStatus::default()); + assert_eq!(stats.avg_util, 0.0); + assert_eq!(stats.avg_pct, 0); + assert_eq!(stats.median_pct, 0); + assert_eq!(stats.total_pct, 0); + } + struct VcpuLimitedClient { target: Arc, } @@ -1219,16 +1280,29 @@ mod tests { async fn get_thread_pool_snapshot(&self) -> Result { Ok(ThreadPoolSnapshot { - thread_count: self.target.load(Ordering::Relaxed), vcpu_count: 4, perf: None, - per_thread_util: None, }) } async fn close(&self) {} } + async fn set_test_state( + instance: &Instance, + thread_count: u32, + vcpu_count: u32, + ownership_classification: Option, + ) { + *instance.snapshot.write().await = Some(ThreadPoolSnapshot { + vcpu_count, + perf: None, + }); + let mut status = instance.status.write().await; + status.thread_count = thread_count; + status.ownership_classification = ownership_classification; + } + /// Test that actuation will not raise the pool above the VM vCPU /// count. #[rstest] @@ -1249,12 +1323,7 @@ mod tests { 0, client, )); - { - let mut status = instance.status.write().await; - status.thread_count = 2; - status.vcpu_count = 4; - status.ownership_classification = Some(true); - } + set_test_state(&instance, 2, 4, Some(true)).await; let mut controller = Controller::new( cfg, Box::new(ThresholdEngine::new(ThresholdConfig::default())), @@ -1322,11 +1391,7 @@ mod tests { 0, client, )); - { - let mut status = instance.status.write().await; - status.thread_count = 2; - status.vcpu_count = 4; - } + set_test_state(&instance, 2, 4, None).await; let mut controller = Controller::new( cfg, Box::new(ThresholdEngine::new(ThresholdConfig::default())), @@ -1398,12 +1463,7 @@ mod tests { 0, client, )); - { - let mut status = instance.status.write().await; - status.thread_count = 4; - status.vcpu_count = 8; - status.ownership_classification = Some(true); - } + set_test_state(&instance, 4, 8, Some(true)).await; let mut controller = Controller::new( cfg, Box::new(ThresholdEngine::new(ThresholdConfig::default())), @@ -1460,12 +1520,7 @@ mod tests { 0, client, )); - { - let mut status = instance.status.write().await; - status.thread_count = 3; - status.vcpu_count = 8; - status.ownership_classification = Some(true); - } + set_test_state(&instance, 3, 8, Some(true)).await; let mut controller = Controller::new( cfg, Box::new(ThresholdEngine::new(ThresholdConfig::default())), @@ -1511,10 +1566,8 @@ mod tests { async fn get_thread_pool_snapshot(&self) -> Result { Ok(ThreadPoolSnapshot { - thread_count: self.target.load(Ordering::Relaxed), vcpu_count: 16, perf: None, - per_thread_util: Some(0.0), }) } @@ -1546,8 +1599,8 @@ mod tests { }; let good_instance = Arc::new(Instance::new("visible".to_string(), Path::new(""), 0, good)); let bad_instance = Arc::new(Instance::new("hidden".to_string(), Path::new(""), 0, bad)); - good_instance.status.write().await.vcpu_count = 16; - bad_instance.status.write().await.vcpu_count = 16; + set_test_state(&good_instance, 1, 16, None).await; + set_test_state(&bad_instance, 1, 16, None).await; controller .instances .insert("visible".to_string(), good_instance.clone()); @@ -1567,7 +1620,6 @@ mod tests { } struct SnapshotClient { - threads: u32, closed: Arc, } @@ -1579,11 +1631,9 @@ mod tests { async fn get_thread_pool_snapshot(&self) -> Result { Ok(ThreadPoolSnapshot { - thread_count: self.threads, // FIXME these weren't required, looked like broken due to rebase vcpu_count: 2, perf: None, - per_thread_util: None, }) } @@ -1625,14 +1675,14 @@ mod tests { async fn evaluate( &self, - _instance: &Arc, + _instance: &EngineInstanceSnapshot, _context: &EngineTickContext, ) -> ScaleAction { self.evaluated.fetch_add(1, Ordering::Relaxed); ScaleAction::None } - async fn on_instance_added(&self, _instance: &Arc) { + async fn on_instance_added(&self, _instance_id: &str) { self.added.fetch_add(1, Ordering::Relaxed); } @@ -1641,12 +1691,12 @@ mod tests { } } - fn instance(id: &str, threads: u32, closed: Arc) -> Arc { + fn instance(id: &str, closed: Arc) -> Arc { Arc::new(Instance::new( id.to_string(), Path::new(""), 1, - SnapshotClient { threads, closed }, + SnapshotClient { closed }, )) } @@ -1675,7 +1725,7 @@ mod tests { .unwrap(); let closed = Arc::new(AtomicUsize::new(0)); - let first = instance("vm-a", 1, Arc::clone(&closed)); + let first = instance("vm-a", Arc::clone(&closed)); let (n_added, n_removed) = controller .sync_instances(vec![Arc::clone(&first)]) .await @@ -1684,7 +1734,7 @@ mod tests { assert_eq!(controller.instances.len(), 1); assert_eq!(added.load(Ordering::Relaxed), 1); - let duplicate = instance("vm-a", 9, Arc::clone(&closed)); + let duplicate = instance("vm-a", Arc::clone(&closed)); let (n_added, n_removed) = controller.sync_instances(vec![duplicate]).await.unwrap(); assert_eq!((n_added, n_removed), (0, 0)); assert_eq!(controller.instances.len(), 1); @@ -1722,7 +1772,7 @@ mod tests { .unwrap(); let closed = Arc::new(AtomicUsize::new(0)); - let ok = instance("ok", 4, Arc::clone(&closed)); + let ok = instance("ok", Arc::clone(&closed)); let bad = Arc::new(Instance::new( "bad".to_string(), Path::new(""), @@ -1732,14 +1782,20 @@ mod tests { controller.sync_instances(vec![ok, bad]).await.unwrap(); assert_eq!(added.load(Ordering::Relaxed), 2); + let task_dir = Path::proc_pid(1).join("task/10"); + std::fs::create_dir_all(&task_dir).unwrap(); + std::fs::write( + task_dir.join("stat"), + "10 (worker0) S 1 1 1 0 -1 0 0 0 0 0 5 3 0 0 20 0 1 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 \ + 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0 0\n", + ) + .unwrap(); + controller.tick().await.unwrap(); assert_eq!(controller.tick_index, 1); assert_eq!(controller.instances.len(), 1); assert!(controller.instances.contains_key("ok")); - assert_eq!( - controller.instances["ok"].status.read().await.thread_count, - 4 - ); + assert!(controller.instances["ok"].status.read().await.thread_count > 0); assert_eq!(evaluated.load(Ordering::Relaxed), 1); assert_eq!(removed.load(Ordering::Relaxed), 1); } diff --git a/src/engines/mod.rs b/src/engines/mod.rs index 3ee27d3d..3cef8449 100644 --- a/src/engines/mod.rs +++ b/src/engines/mod.rs @@ -5,14 +5,14 @@ //! Pluggable fleet scaling decisions. -use std::{sync::Arc, time::Instant}; +use std::time::Instant; use async_trait::async_trait; 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, util::Path}; #[cfg(feature = "threshold-engine")] pub mod threshold; @@ -121,6 +121,29 @@ impl Default for EngineTickContext { } } +/// Immutable, controller-processed input for one engine evaluation. +#[derive(Debug, Default, Clone)] +pub struct EngineInstanceSnapshot { + /// Stable VM identifier. + pub id: String, + /// Current backend worker count. + pub thread_count: u32, + /// Current backend-reported vCPU count. + pub vcpu_count: u32, + /// Canonical average utilisation per worker, from 0.0 to 1.0. + pub per_thread_util: f64, + /// Whether a complete procfs interval contributed CPU utilisation. + pub cpu_available: bool, + /// Latest cumulative completed I/O count. + pub io_count_total: u64, + /// Whether cgroup CPU throttling increased in the latest interval. + pub throttled_recently: bool, + /// Whether ownership policy permits automatic scaling. + pub scaling_allowed: bool, + /// Whether a manual override suppresses automatic scaling. + pub manual_scaling_sticky: bool, +} + /// One VM's entry in a fleet-wide plan. #[derive(Debug, Clone, PartialEq, Eq)] pub struct InstanceDecision { @@ -220,7 +243,11 @@ pub trait ScalingEngine: Send + Sync { /// /// The controller needs to pass `Instances` to the engine for it to /// evaluate and decide. - async fn evaluate(&self, instance: &Arc, context: &EngineTickContext) -> ScaleAction; + async fn evaluate( + &self, + instance: &EngineInstanceSnapshot, + context: &EngineTickContext, + ) -> ScaleAction; /// Evaluate eligible VMs concurrently and collect one coherent plan. /// @@ -229,16 +256,13 @@ pub trait ScalingEngine: Send + Sync { /// not evaluate them. async fn evaluate_fleet( &self, - instances: &[Arc], + instances: &[EngineInstanceSnapshot], context: &EngineTickContext, ) -> Vec { // FIXME concurrent evaluation might not be a good idea, an engine might want // to first look at all the instances first and then make decisions join_all(instances.iter().map(|instance| async move { - let status = instance.status.read().await; - let scaling_allowed = status.scaling_allowed; - let sticky = status.manual_scaling_sticky; - if sticky || !scaling_allowed { + if instance.manual_scaling_sticky || !instance.scaling_allowed { return None; } Some(InstanceDecision::new( @@ -263,7 +287,7 @@ pub trait ScalingEngine: Send + Sync { async fn on_applied(&self, _instance_id: &str, _outcome: AppliedOutcome) {} /// Notify the engine when a VM is added. - async fn on_instance_added(&self, _instance: &Arc) {} + async fn on_instance_added(&self, _instance_id: &str) {} /// Notify the engine when a VM is removed. async fn on_instance_removed(&self, _instance_id: &str) {} @@ -327,31 +351,6 @@ mod tests { use std::sync::atomic::{AtomicUsize, Ordering}; - use crate::{ - backends::BackendClientError, - instance::{InstanceClient, ThreadPoolSnapshot}, - }; - - struct SnapshotClient; - - #[async_trait] - impl InstanceClient for SnapshotClient { - async fn set_thread_count(&self, _count: u32) -> Result<(), BackendClientError> { - Ok(()) - } - - async fn get_thread_pool_snapshot(&self) -> Result { - Ok(ThreadPoolSnapshot { - thread_count: 1, - vcpu_count: 2, - perf: None, - per_thread_util: Some(0.0), - }) - } - - async fn close(&self) {} - } - struct PerInstanceEngine { calls: AtomicUsize, } @@ -368,7 +367,7 @@ mod tests { async fn evaluate( &self, - _instance: &Arc, + _instance: &EngineInstanceSnapshot, _context: &EngineTickContext, ) -> ScaleAction { self.calls.fetch_add(1, Ordering::Relaxed); @@ -376,13 +375,12 @@ mod tests { } } - fn test_instance(id: &str) -> Arc { - Arc::new(Instance::new( - id.to_string(), - Path::new(""), - 0, - SnapshotClient, - )) + fn test_instance(id: &str) -> EngineInstanceSnapshot { + EngineInstanceSnapshot { + id: id.to_string(), + scaling_allowed: true, + ..Default::default() + } } /// Test that default fleet `evaluate` skips ineligible VMs and only @@ -393,8 +391,8 @@ mod tests { calls: AtomicUsize::new(0), }; let first = test_instance("first"); - let sticky = test_instance("sticky"); - sticky.status.write().await.manual_scaling_sticky = true; + let mut sticky = test_instance("sticky"); + sticky.manual_scaling_sticky = true; let plan = engine .evaluate_fleet(&[first.clone(), sticky], &EngineTickContext::default()) diff --git a/src/engines/threshold.rs b/src/engines/threshold.rs index 4a55aa2b..1e6969e0 100644 --- a/src/engines/threshold.rs +++ b/src/engines/threshold.rs @@ -5,7 +5,7 @@ //! Threshold engine: size a worker pool to carry the observed CPU load. -use std::{collections::HashMap, sync::Arc, time::Instant}; +use std::{collections::HashMap, time::Instant}; use async_trait::async_trait; use serde::{Deserialize, Serialize}; @@ -14,10 +14,9 @@ use tokio::sync::Mutex; use crate::{ config::{self, ConfigError, deserialize_percent, serialize_percent}, engines::{ - AppliedOutcome, EngineError, EngineRegistration, EngineTickContext, ScaleAction, - ScalingEngine, + AppliedOutcome, EngineError, EngineInstanceSnapshot, EngineRegistration, EngineTickContext, + ScaleAction, ScalingEngine, }, - instance::Instance, util::Path, }; @@ -225,11 +224,6 @@ enum PostRevertUpHold { /// Per-VM counters owned by this engine. #[derive(Clone)] struct InstanceState { - /// Original controller record retained so lifecycle-owned state can refer - /// to the same VM without duplicating its identity or backend handle. - // FIXME This is probably not needed. Only .evaluate() uses it, and that already takes - // a &Arc. It seems like this contradicts the controller owning inventory design. - instance: Arc, /// Number of consecutive polls eligible for scale-down. low_util_polls: u32, /// Successful action waiting for its performance validation sample. @@ -250,10 +244,9 @@ struct InstanceState { } impl InstanceState { - /// Create lifecycle state for the exact controller instance. - fn new(instance: Arc) -> Self { + /// Create lifecycle state for one controller-owned instance. + fn new() -> Self { Self { - instance, low_util_polls: 0, pending_validation: None, baseline_window_start: None, @@ -569,7 +562,7 @@ fn log_performance_revert( /// Emit one operator-visible scale-up decision with validation expectations. fn log_scale_up_decision( - instance: &Instance, + instance: &EngineInstanceSnapshot, per_thread_util: f64, thread_count: u32, target: u32, @@ -580,7 +573,7 @@ fn log_scale_up_decision( tracing::info!( target: "engine", event = "scale_up", - vm = instance.to_string(), + vm = %instance.id, util = per_thread_util, thr = %format!("{thread_count}->{target}"), current_iops, @@ -592,7 +585,7 @@ fn log_scale_up_decision( /// Emit one operator-visible scale-down decision. fn log_scale_down_decision( - instance: &Instance, + instance: &EngineInstanceSnapshot, per_thread_util: f64, thread_count: u32, target: u32, @@ -600,7 +593,7 @@ fn log_scale_down_decision( tracing::info!( target: "engine", event = "scale_down", - vm = instance.to_string(), + vm = %instance.id, util = per_thread_util, thr = %format!("{thread_count}->{target}"), "threshold scaling decision" @@ -648,25 +641,24 @@ impl ScalingEngine for ThresholdEngine { serde_json::to_value(&self.cfg).unwrap_or(serde_json::Value::Null) } - async fn evaluate(&self, instance: &Arc, context: &EngineTickContext) -> ScaleAction { - let (per_thread_util, thread_count, io_count_total, throttled_recently) = { - let status = instance.status.read().await; - ( - status.per_thread_util, - status.thread_count, - match &status.perf { - Some(perf) => perf.total_io_count(), - None => 0, - }, - status.throttled_usec_delta > 0, - ) - }; + async fn evaluate( + &self, + instance: &EngineInstanceSnapshot, + context: &EngineTickContext, + ) -> ScaleAction { + if !instance.cpu_available { + return ScaleAction::None; + } + let per_thread_util = instance.per_thread_util; + let thread_count = instance.thread_count; + let io_count_total = instance.io_count_total; + let throttled_recently = instance.throttled_recently; let down_target = self.scale_down_target(per_thread_util, thread_count, context.min_thread_count); let mut state = self.state.lock().await; let instance_state = state .entry(instance.id.clone()) - .or_insert_with(|| InstanceState::new(Arc::clone(instance))); + .or_insert_with(InstanceState::new); let sample = IoSample { io_count: io_count_total, @@ -682,7 +674,7 @@ impl ScalingEngine for ThresholdEngine { let observed_iops = iops_rate(pending.validation_start, sample); let failed_up = matches!(pending.action, ScaleAction::Up(_)); let revert_target = self.validation_revert_target( - &instance_state.instance.id, + &instance.id, thread_count, observed_iops, per_thread_util, @@ -830,11 +822,11 @@ impl ScalingEngine for ThresholdEngine { } } - async fn on_instance_added(&self, instance: &Arc) { + async fn on_instance_added(&self, instance_id: &str) { let mut state = self.state.lock().await; state - .entry(instance.id.clone()) - .or_insert_with(|| InstanceState::new(Arc::clone(instance))); + .entry(instance_id.to_string()) + .or_insert_with(InstanceState::new); } async fn on_instance_removed(&self, instance_id: &str) { @@ -850,7 +842,6 @@ mod tests { time::{Duration, Instant}, }; - use async_trait::async_trait; use rstest::{fixture, rstest}; use test_log::test; @@ -859,10 +850,8 @@ mod tests { log_performance_revert, }; use crate::{ - backends::BackendClientError, - engines::{AppliedOutcome, EngineTickContext, ScaleAction, ScalingEngine}, - instance::{ - Instance, InstanceClient, InstancePerfSample, InstanceStatus, ThreadPoolSnapshot, + engines::{ + AppliedOutcome, EngineInstanceSnapshot, EngineTickContext, ScaleAction, ScalingEngine, }, util::Path, }; @@ -881,39 +870,34 @@ mod tests { } } - struct SnapshotClient; + const VM_ID: &str = "vm-1"; - #[async_trait] - impl InstanceClient for SnapshotClient { - async fn set_thread_count(&self, _count: u32) -> Result<(), BackendClientError> { - Ok(()) - } - - async fn get_thread_pool_snapshot(&self) -> Result { - Ok(ThreadPoolSnapshot { - thread_count: 1, - vcpu_count: 8, - perf: None, - per_thread_util: Some(0.0), - }) + fn observation( + thread_count: u32, + per_thread_util: f64, + io_count_total: u64, + ) -> EngineInstanceSnapshot { + EngineInstanceSnapshot { + id: VM_ID.to_string(), + thread_count, + vcpu_count: 8, + per_thread_util, + io_count_total, + scaling_allowed: true, + cpu_available: true, + ..Default::default() } - - async fn close(&self) {} } - #[fixture] - async fn instance() -> Arc { - let instance = Arc::new(Instance::new( - "vm-1".to_string(), - Path::new(""), - 0, - SnapshotClient, - )); - let mut status = instance.status.write().await; - *status = InstanceStatus::default(); - status.perf = Some(InstancePerfSample::default()); - drop(status); - instance + fn unavailable_observation() -> EngineInstanceSnapshot { + EngineInstanceSnapshot { + id: VM_ID.to_string(), + thread_count: 4, + vcpu_count: 8, + per_thread_util: 1.0, + scaling_allowed: true, + ..Default::default() + } } fn context_at(now: Instant) -> EngineTickContext { @@ -924,38 +908,25 @@ mod tests { } } - async fn set_observation( - instance: &Arc, - thread_count: u32, - per_thread_util: f64, - io_count: u64, - ) { - let mut status = instance.status.write().await; - status.thread_count = thread_count; - status.per_thread_util = per_thread_util; - status.perf.as_mut().unwrap().read_io_count = io_count; - } - /// Establish a baseline window of `baseline_iops` ops/sec ending at `t1`. async fn seed_baseline_rate( engine: &ThresholdEngine, - instance: &Arc, thread_count: u32, baseline_iops: u64, t0: Instant, t1: Instant, ) { - engine.on_instance_added(instance).await; + engine.on_instance_added(VM_ID).await; // Park util exactly at the up threshold so neither up nor down fires. let steady_util = engine.cfg.scale_up_threshold; - set_observation(instance, thread_count, steady_util, 0).await; + let first = observation(thread_count, steady_util, 0); assert_eq!( - engine.evaluate(instance, &context_at(t0)).await, + engine.evaluate(&first, &context_at(t0)).await, ScaleAction::None ); - set_observation(instance, thread_count, steady_util, baseline_iops).await; + let second = observation(thread_count, steady_util, baseline_iops); assert_eq!( - engine.evaluate(instance, &context_at(t1)).await, + engine.evaluate(&second, &context_at(t1)).await, ScaleAction::None ); } @@ -1000,20 +971,16 @@ mod tests { /// Test that an IOPS-rate drop after scale-up triggers revert. #[rstest] #[tokio::test] - async fn test_scale_up_revert_fires_on_iops_rate_drop( - engine: ThresholdEngine, - #[future] instance: Arc, - ) { - let instance = instance.await; + async fn test_scale_up_revert_fires_on_iops_rate_drop(engine: ThresholdEngine) { let t0 = Instant::now(); let t1 = t0 + Duration::from_secs(1); let t_apply = t1; let t_validate = t1 + Duration::from_secs(1); - seed_baseline_rate(&engine, &instance, 5, 100_000, t0, t1).await; + seed_baseline_rate(&engine, 5, 100_000, t0, t1).await; engine .on_applied( - &instance.id, + VM_ID, AppliedOutcome::Success { action: ScaleAction::Up(6), prev_thread_count: 5, @@ -1023,10 +990,10 @@ mod tests { ) .await; // Validation window: only 50k ops over 1s → 50k IOPS (< 105k required). - set_observation(&instance, 6, 0.6, 150_000).await; + let observed = observation(6, 0.6, 150_000); assert_eq!( - engine.evaluate(&instance, &context_at(t_validate)).await, + engine.evaluate(&observed, &context_at(t_validate)).await, ScaleAction::Revert(5) ); } @@ -1035,20 +1002,16 @@ mod tests { /// revert. #[rstest] #[tokio::test] - async fn test_scale_up_flat_rate_reverts( - engine: ThresholdEngine, - #[future] instance: Arc, - ) { - let instance = instance.await; + async fn test_scale_up_flat_rate_reverts(engine: ThresholdEngine) { let t0 = Instant::now(); let t1 = t0 + Duration::from_secs(1); let t_apply = t1; let t_validate = t1 + Duration::from_secs(1); - seed_baseline_rate(&engine, &instance, 3, 100_000, t0, t1).await; + seed_baseline_rate(&engine, 3, 100_000, t0, t1).await; engine .on_applied( - &instance.id, + VM_ID, AppliedOutcome::Success { action: ScaleAction::Up(4), prev_thread_count: 3, @@ -1058,10 +1021,10 @@ mod tests { ) .await; // +1% ops over the validation second → still below 5% gain. - set_observation(&instance, 4, 0.6, 101_000).await; + let observed = observation(4, 0.6, 101_000); assert_eq!( - engine.evaluate(&instance, &context_at(t_validate)).await, + engine.evaluate(&observed, &context_at(t_validate)).await, ScaleAction::Revert(3) ); } @@ -1069,20 +1032,16 @@ mod tests { /// Test that a sufficient post-scale IOPS-rate gain keeps the scale-up. #[rstest] #[tokio::test] - async fn test_scale_up_rate_gain_keeps_scale( - engine: ThresholdEngine, - #[future] instance: Arc, - ) { - let instance = instance.await; + async fn test_scale_up_rate_gain_keeps_scale(engine: ThresholdEngine) { let t0 = Instant::now(); let t1 = t0 + Duration::from_secs(1); let t_apply = t1; let t_validate = t1 + Duration::from_secs(1); - seed_baseline_rate(&engine, &instance, 3, 100_000, t0, t1).await; + seed_baseline_rate(&engine, 3, 100_000, t0, t1).await; engine .on_applied( - &instance.id, + VM_ID, AppliedOutcome::Success { action: ScaleAction::Up(4), prev_thread_count: 3, @@ -1092,10 +1051,10 @@ mod tests { ) .await; // +10% ops/sec during validation → above 5% min gain. - set_observation(&instance, 4, 0.6, 210_000).await; + let observed = observation(4, 0.6, 210_000); assert_eq!( - engine.evaluate(&instance, &context_at(t_validate)).await, + engine.evaluate(&observed, &context_at(t_validate)).await, ScaleAction::None ); } @@ -1103,10 +1062,8 @@ mod tests { /// Test that a failed scale-up enters grace then recovery: scale-up is /// blocked while a settled baseline is measured, then the next up uses /// that recovery-window IOPS (not a depressed post-revert blip). - #[rstest] #[tokio::test] - async fn test_failed_scale_up_grace_then_recovery_baseline(#[future] instance: Arc) { - let instance = instance.await; + async fn test_failed_scale_up_grace_then_recovery_baseline() { let engine = ThresholdEngine::new(ThresholdConfig { scale_up_threshold: 0.8, scale_up_min_gain: 0.05, @@ -1120,10 +1077,10 @@ mod tests { let t_apply = t1; let t_validate = t1 + Duration::from_secs(1); - seed_baseline_rate(&engine, &instance, 3, 100_000, t0, t1).await; + seed_baseline_rate(&engine, 3, 100_000, t0, t1).await; engine .on_applied( - &instance.id, + VM_ID, AppliedOutcome::Success { action: ScaleAction::Up(4), prev_thread_count: 3, @@ -1132,51 +1089,51 @@ mod tests { }, ) .await; - set_observation(&instance, 4, 0.95, 101_000).await; + let observed = observation(4, 0.95, 101_000); assert_eq!( - engine.evaluate(&instance, &context_at(t_validate)).await, + engine.evaluate(&observed, &context_at(t_validate)).await, ScaleAction::Revert(3) ); // Grace: util blip / depressed IOPS must not produce a scale-up. - set_observation(&instance, 3, 0.4, 150_000).await; + let observed = observation(3, 0.4, 150_000); assert_eq!( engine - .evaluate(&instance, &context_at(t_validate + Duration::from_secs(1))) + .evaluate(&observed, &context_at(t_validate + Duration::from_secs(1))) .await, ScaleAction::None ); // Recovery tick 1: enter recovery window (anchor), still blocked. - set_observation(&instance, 3, 0.95, 150_000).await; + let observed = observation(3, 0.95, 150_000); assert_eq!( engine - .evaluate(&instance, &context_at(t_validate + Duration::from_secs(2))) + .evaluate(&observed, &context_at(t_validate + Duration::from_secs(2))) .await, ScaleAction::None ); // Recovery tick 2: measure ~100k IOPS over 1s, still blocked. - set_observation(&instance, 3, 0.95, 250_000).await; + let observed = observation(3, 0.95, 250_000); assert_eq!( engine - .evaluate(&instance, &context_at(t_validate + Duration::from_secs(3))) + .evaluate(&observed, &context_at(t_validate + Duration::from_secs(3))) .await, ScaleAction::None ); // Hold complete: scale-up allowed; baseline is the recovery-window rate // (~100k), so min_target reflects that — not the grace-period blip. - set_observation(&instance, 3, 0.95, 350_000).await; + let observed = observation(3, 0.95, 350_000); assert_eq!( engine - .evaluate(&instance, &context_at(t_validate + Duration::from_secs(4))) + .evaluate(&observed, &context_at(t_validate + Duration::from_secs(4))) .await, ScaleAction::Up(4) ); { let state = engine.state.lock().await; - let instance_state = state.get(&instance.id).unwrap(); + let instance_state = state.get(VM_ID).unwrap(); // Recovery anchored at 150k @ +2s through 350k @ +4s → 100k IOPS. assert_eq!(instance_state.current_baseline_iops, 100_000); assert!(instance_state.post_revert_up_hold.is_none()); @@ -1184,10 +1141,8 @@ mod tests { } /// Test that scale-down remains allowed during the post-revert up hold. - #[rstest] #[tokio::test] - async fn test_scale_down_allowed_during_post_revert_up_hold(#[future] instance: Arc) { - let instance = instance.await; + async fn test_scale_down_allowed_during_post_revert_up_hold() { let engine = ThresholdEngine::new(ThresholdConfig { scale_up_threshold: 0.8, scale_up_min_gain: 0.05, @@ -1202,10 +1157,10 @@ mod tests { let t_apply = t1; let t_validate = t1 + Duration::from_secs(1); - seed_baseline_rate(&engine, &instance, 3, 100_000, t0, t1).await; + seed_baseline_rate(&engine, 3, 100_000, t0, t1).await; engine .on_applied( - &instance.id, + VM_ID, AppliedOutcome::Success { action: ScaleAction::Up(4), prev_thread_count: 3, @@ -1214,27 +1169,25 @@ mod tests { }, ) .await; - set_observation(&instance, 4, 0.95, 101_000).await; + let observed = observation(4, 0.95, 101_000); assert_eq!( - engine.evaluate(&instance, &context_at(t_validate)).await, + engine.evaluate(&observed, &context_at(t_validate)).await, ScaleAction::Revert(3) ); // During grace, low util can still scale down. - set_observation(&instance, 3, 0.0, 101_000).await; + let observed = observation(3, 0.0, 101_000); assert_eq!( engine - .evaluate(&instance, &context_at(t_validate + Duration::from_secs(1))) + .evaluate(&observed, &context_at(t_validate + Duration::from_secs(1))) .await, ScaleAction::Down(2) ); } /// Test that scale-down validation does not revert when util stays low /// (demand left; IOPS collapse is expected). - #[rstest] #[tokio::test] - async fn test_scale_down_keeps_shrink_when_util_stays_low(#[future] instance: Arc) { - let instance = instance.await; + async fn test_scale_down_keeps_shrink_when_util_stays_low() { let engine = ThresholdEngine::new(ThresholdConfig { scale_up_threshold: 0.8, scale_down_revert_drop: 0.05, @@ -1247,10 +1200,10 @@ mod tests { let t_apply = t1; let t_validate = t1 + Duration::from_secs(1); - seed_baseline_rate(&engine, &instance, 4, 100_000, t0, t1).await; + seed_baseline_rate(&engine, 4, 100_000, t0, t1).await; engine .on_applied( - &instance.id, + VM_ID, AppliedOutcome::Success { action: ScaleAction::Down(1), prev_thread_count: 4, @@ -1260,21 +1213,17 @@ mod tests { ) .await; // IOPS collapsed, but util remains below the up threshold. - set_observation(&instance, 1, 0.0, 100_001).await; + let observed = observation(1, 0.0, 100_001); assert_eq!( - engine.evaluate(&instance, &context_at(t_validate)).await, + engine.evaluate(&observed, &context_at(t_validate)).await, ScaleAction::None ); } /// Test that scale-down validation reverts when util is still high (demand /// remains) but IOPS fell beyond the tolerated drop. - #[rstest] #[tokio::test] - async fn test_scale_down_reverts_when_util_stays_high_and_iops_drop( - #[future] instance: Arc, - ) { - let instance = instance.await; + async fn test_scale_down_reverts_when_util_stays_high_and_iops_drop() { let engine = ThresholdEngine::new(ThresholdConfig { scale_up_threshold: 0.8, scale_down_revert_drop: 0.05, @@ -1286,10 +1235,10 @@ mod tests { let t_apply = t1; let t_validate = t1 + Duration::from_secs(1); - seed_baseline_rate(&engine, &instance, 4, 100_000, t0, t1).await; + seed_baseline_rate(&engine, 4, 100_000, t0, t1).await; engine .on_applied( - &instance.id, + VM_ID, AppliedOutcome::Success { action: ScaleAction::Down(2), prev_thread_count: 4, @@ -1299,9 +1248,9 @@ mod tests { ) .await; // Demand still present (util above up threshold) but IOPS collapsed. - set_observation(&instance, 2, 0.95, 110_000).await; + let observed = observation(2, 0.95, 110_000); assert_eq!( - engine.evaluate(&instance, &context_at(t_validate)).await, + engine.evaluate(&observed, &context_at(t_validate)).await, ScaleAction::Revert(4) ); } @@ -1321,7 +1270,6 @@ mod tests { /// it one second later with `io_delta` completed operations. async fn apply_and_validate( engine: &ThresholdEngine, - instance: &Arc, action: ScaleAction, prev_thread_count: u32, io_delta: u64, @@ -1331,12 +1279,12 @@ mod tests { .state .lock() .await - .get_mut(&instance.id) + .get_mut(VM_ID) .unwrap() .current_baseline_iops = 100_000; engine .on_applied( - &instance.id, + VM_ID, AppliedOutcome::Success { action, prev_thread_count, @@ -1346,19 +1294,17 @@ mod tests { ) .await; let target = action.target().unwrap(); - set_observation(instance, target, 0.95, io_delta).await; + let observed = observation(target, 0.95, io_delta); engine - .evaluate(instance, &context_at(at + Duration::from_secs(1))) + .evaluate(&observed, &context_at(at + Duration::from_secs(1))) .await } /// Test that each consecutive failed scale-up doubles the post-revert /// hold up to the cap, extending grace while recovery stays fixed, and /// that a kept scale-up resets the backoff. - #[rstest] #[tokio::test] - async fn test_failed_scale_ups_back_off_until_kept(#[future] instance: Arc) { - let instance = instance.await; + async fn test_failed_scale_ups_back_off_until_kept() { let engine = ThresholdEngine::new(ThresholdConfig { scale_up_threshold: 0.8, scale_up_min_gain: 0.05, @@ -1368,18 +1314,18 @@ mod tests { post_revert_up_backoff_max_polls: 12, ..Default::default() }); - engine.on_instance_added(&instance).await; + engine.on_instance_added(VM_ID).await; let start = Instant::now(); // Holds of 3, 6, 12, 12 polls minus the 2 recovery polls. for (i, expected_grace) in [1, 4, 10, 10].into_iter().enumerate() { let at = start + Duration::from_secs(10 * i as u64); assert_eq!( - apply_and_validate(&engine, &instance, ScaleAction::Up(4), 3, 1_000, at).await, + apply_and_validate(&engine, ScaleAction::Up(4), 3, 1_000, at).await, ScaleAction::Revert(3) ); let state = engine.state.lock().await; - let instance_state = state.get(&instance.id).unwrap(); + let instance_state = state.get(VM_ID).unwrap(); assert_eq!(instance_state.consecutive_failed_ups, i as u32 + 1); assert!(matches!( instance_state.post_revert_up_hold, @@ -1389,21 +1335,19 @@ mod tests { let at = start + Duration::from_secs(100); assert_eq!( - apply_and_validate(&engine, &instance, ScaleAction::Up(4), 3, 200_000, at).await, + apply_and_validate(&engine, ScaleAction::Up(4), 3, 200_000, at).await, ScaleAction::None ); let state = engine.state.lock().await; - assert_eq!(state.get(&instance.id).unwrap().consecutive_failed_ups, 0); + assert_eq!(state.get(VM_ID).unwrap().consecutive_failed_ups, 0); drop(state); assert_eq!(engine.post_revert_up_grace_polls(1), 1); } /// Test that each consecutive failed scale-down doubles the sustain /// requirement up to the cap, and that an applied scale-up resets it. - #[rstest] #[tokio::test] - async fn test_failed_scale_downs_back_off_until_scale_up(#[future] instance: Arc) { - let instance = instance.await; + async fn test_failed_scale_downs_back_off_until_scale_up() { let engine = ThresholdEngine::new(ThresholdConfig { scale_up_threshold: 0.8, scale_down_revert_drop: 0.05, @@ -1412,20 +1356,20 @@ mod tests { scale_down_backoff_max_polls: 20, ..Default::default() }); - engine.on_instance_added(&instance).await; + engine.on_instance_added(VM_ID).await; let start = Instant::now(); for (i, expected_sustain) in [6, 12, 20].into_iter().enumerate() { let at = start + Duration::from_secs(10 * i as u64); assert_eq!( - apply_and_validate(&engine, &instance, ScaleAction::Down(2), 4, 1_000, at).await, + apply_and_validate(&engine, ScaleAction::Down(2), 4, 1_000, at).await, ScaleAction::Revert(4) ); let failures = engine .state .lock() .await - .get(&instance.id) + .get(VM_ID) .unwrap() .consecutive_failed_downs; assert_eq!(failures, i as u32 + 1); @@ -1434,7 +1378,7 @@ mod tests { engine .on_applied( - &instance.id, + VM_ID, AppliedOutcome::Success { action: ScaleAction::Up(5), prev_thread_count: 4, @@ -1444,7 +1388,7 @@ mod tests { ) .await; let state = engine.state.lock().await; - assert_eq!(state.get(&instance.id).unwrap().consecutive_failed_downs, 0); + assert_eq!(state.get(VM_ID).unwrap().consecutive_failed_downs, 0); } /// Test that an applied action resets only the evidence for its own @@ -1456,14 +1400,13 @@ mod tests { #[case::revert(ScaleAction::Revert(3), 5, 2, 3, true)] #[tokio::test] async fn test_reset_after_applied( - #[future] instance: Arc, #[case] action: ScaleAction, #[case] low_util_polls: u32, #[case] consecutive_failed_ups: u32, #[case] consecutive_failed_downs: u32, #[case] post_revert_up_hold: bool, ) { - let mut instance_state = InstanceState::new(instance.await); + let mut instance_state = InstanceState::new(); instance_state.low_util_polls = 5; instance_state.consecutive_failed_ups = 2; instance_state.consecutive_failed_downs = 3; @@ -1518,19 +1461,15 @@ mod tests { /// scale-ups are suppressed. #[rstest] #[test(tokio::test)] - async fn test_no_scale_up_during_pending_validation( - engine: ThresholdEngine, - #[future] instance: Arc, - ) { - let instance = instance.await; + async fn test_no_scale_up_during_pending_validation(engine: ThresholdEngine) { let t0 = Instant::now(); let t1 = t0 + Duration::from_secs(1); let t_apply = t1; - seed_baseline_rate(&engine, &instance, 4, 100_000, t0, t1).await; + seed_baseline_rate(&engine, 4, 100_000, t0, t1).await; engine .on_applied( - &instance.id, + VM_ID, AppliedOutcome::Success { action: ScaleAction::Up(5), prev_thread_count: 4, @@ -1540,13 +1479,13 @@ mod tests { ) .await; - set_observation(&instance, 5, 0.88, 220_000).await; + let observed = observation(5, 0.88, 220_000); for i in 0..engine.cfg.scale_validation_sample_polls { assert_eq!( engine .evaluate( - &instance, + &observed, &context_at(t_apply + Duration::from_secs(u64::from(i) + 1)) ) .await, @@ -1557,15 +1496,15 @@ mod tests { // Validation tick settles the scale-up without proposing another action. let t_settle = t_apply + Duration::from_secs(1); assert_eq!( - engine.evaluate(&instance, &context_at(t_settle)).await, + engine.evaluate(&observed, &context_at(t_settle)).await, ScaleAction::None ); // Next tick with sustained high util may scale again. - set_observation(&instance, 5, 0.88, 330_000).await; + let observed = observation(5, 0.88, 330_000); assert_eq!( engine - .evaluate(&instance, &context_at(t_settle + Duration::from_secs(1))) + .evaluate(&observed, &context_at(t_settle + Duration::from_secs(1))) .await, ScaleAction::Up(6) ); @@ -1573,13 +1512,8 @@ mod tests { /// Test that a scale up is revert after the validation period if /// performance doesn't increase much. - #[rstest] #[test(tokio::test)] - async fn test_pending_validation_reverts_regressive_scale( - _engine: ThresholdEngine, - #[future] instance: Arc, - ) { - let instance = instance.await; + async fn test_pending_validation_reverts_regressive_scale() { let engine = ThresholdEngine::new(ThresholdConfig { scale_up_threshold: 0.8, scale_up_min_gain: 0.05, @@ -1590,10 +1524,10 @@ mod tests { let t1 = t0 + Duration::from_secs(1); let t_apply = t1; - seed_baseline_rate(&engine, &instance, 4, 100_000, t0, t1).await; + seed_baseline_rate(&engine, 4, 100_000, t0, t1).await; engine .on_applied( - &instance.id, + VM_ID, AppliedOutcome::Success { action: ScaleAction::Up(5), prev_thread_count: 4, @@ -1602,13 +1536,13 @@ mod tests { }, ) .await; - set_observation(&instance, 5, 0.88, 140_000).await; + let observed = observation(5, 0.88, 140_000); for i in 0..engine.cfg.scale_validation_sample_polls { assert_eq!( engine .evaluate( - &instance, + &observed, &context_at(t_apply + Duration::from_secs(u64::from(i) + 1)) ) .await, @@ -1618,7 +1552,7 @@ mod tests { // 40k ops over ~3s ≈ 13k IOPS ≪ 105k required. assert_eq!( engine - .evaluate(&instance, &context_at(t_apply + Duration::from_secs(3))) + .evaluate(&observed, &context_at(t_apply + Duration::from_secs(3))) .await, ScaleAction::Revert(4) ); @@ -1627,11 +1561,7 @@ mod tests { /// Test that scale-up decision logs current/min IOPS and validation delay. #[rstest] #[tokio::test] - async fn test_scale_up_log_includes_iops_targets( - engine: ThresholdEngine, - #[future] instance: Arc, - ) { - let instance = instance.await; + async fn test_scale_up_log_includes_iops_targets(engine: ThresholdEngine) { let output = Arc::new(Mutex::new(Vec::new())); let writer_output = Arc::clone(&output); let subscriber = tracing_subscriber::fmt() @@ -1645,11 +1575,11 @@ mod tests { let t0 = Instant::now(); let t1 = t0 + Duration::from_secs(1); - seed_baseline_rate(&engine, &instance, 2, 100_000, t0, t1).await; - set_observation(&instance, 2, 0.9, 100_000).await; + seed_baseline_rate(&engine, 2, 100_000, t0, t1).await; + let observed = observation(2, 0.9, 100_000); assert_eq!( - engine.evaluate(&instance, &context_at(t1)).await, + engine.evaluate(&observed, &context_at(t1)).await, ScaleAction::Up(3) ); @@ -1673,13 +1603,21 @@ mod tests { /// Test that CPU utilisation below scale-up/down thresholds yields /// Hold/`None`. - #[rstest] #[tokio::test] - async fn test_evaluate_holds_when_util_below_thresholds(#[future] instance: Arc) { - let instance = instance.await; + async fn test_evaluate_holds_when_util_below_thresholds() { + let engine = ThresholdEngine::new(ThresholdConfig::default()); + let observed = observation(4, 0.75, 0); + let action = engine + .evaluate(&observed, &context_at(Instant::now())) + .await; + assert_eq!(action, ScaleAction::None); + } + + #[tokio::test] + async fn test_evaluate_holds_until_cpu_is_available() { let engine = ThresholdEngine::new(ThresholdConfig::default()); let action = engine - .evaluate(&instance, &context_at(Instant::now())) + .evaluate(&unavailable_observation(), &context_at(Instant::now())) .await; assert_eq!(action, ScaleAction::None); } diff --git a/src/instance.rs b/src/instance.rs index 1ed0d6b0..6998c9c4 100644 --- a/src/instance.rs +++ b/src/instance.rs @@ -129,10 +129,10 @@ pub struct Instance { pub thread_name_filter: ThreadNameFilter, /// Per-VM backend implementation. pub client: Box, - /// Latest snapshot shared by refresh, evaluation, and status output. - /// - /// The lock permits all VM refresh and evaluation futures to share their - /// records safely. + /// Latest raw observation returned by the backend. + pub snapshot: RwLock>, + /// Sampling baselines, policy state, and metrics maintained by the + /// controller. pub status: RwLock, } @@ -189,6 +189,7 @@ impl Instance { pid, thread_name_filter: ThreadNameFilter::default(), client: Box::new(client), + snapshot: RwLock::new(None), status: RwLock::new(InstanceStatus { scaling_allowed: true, ..Default::default() @@ -229,20 +230,16 @@ impl Instance { snapshot: ThreadPoolSnapshot, refresh_cgroup: bool, ) -> bool { - let cpu = if snapshot.per_thread_util.is_none() { - match read_cpu_sample(self.pid, &self.thread_name_filter) { - Ok(sample) => Some(sample), - Err(error) => { - tracing::warn!( - target: "controller", - %error, - "failed to sample backend task CPU" - ); - None - } + let cpu = match read_cpu_sample(self.pid, &self.thread_name_filter) { + Ok(sample) => Some(sample), + Err(error) => { + tracing::warn!( + target: "controller", + %error, + "failed to sample backend task CPU" + ); + None } - } else { - None }; let previous_cgroup = self.status.read().await.cgroup; let cgroup = if refresh_cgroup || previous_cgroup.is_none() { @@ -251,17 +248,22 @@ impl Instance { previous_cgroup }; let now = Instant::now(); + // Always acquire the raw snapshot lock before the status lock. Readers + // that need a coherent view use the same order. + let mut latest_snapshot = self.snapshot.write().await; let mut status = self.status.write().await; - if status.ownership_classification.is_none() { + if let Some(cpu) = cpu.as_ref() { + status.thread_count = cpu.thread_count; + } + if status.ownership_classification.is_none() + && let Some(cpu) = cpu.as_ref() + { status.ownership_classification = Some( self.client - .initial_pool_is_managed(snapshot.thread_count, snapshot.vcpu_count), + .initial_pool_is_managed(cpu.thread_count, snapshot.vcpu_count), ); } Self::update_perf_rates(&mut status, &snapshot.perf, now); - status.thread_count = snapshot.thread_count; - status.vcpu_count = snapshot.vcpu_count; - status.perf = snapshot.perf.clone(); status.read_latency_us = snapshot.perf.as_ref().and_then(|perf| perf.read_latency_us); status.write_latency_us = snapshot .perf @@ -272,14 +274,7 @@ impl Instance { .as_ref() .and_then(|perf| perf.per_vq_depth.clone()); status.alive = true; - let backend_util = snapshot.per_thread_util.map(|util| util.clamp(0.0, 1.0)); - if let Some(backend_util) = backend_util { - status.per_worker_util.clear(); - status.last_worker_names = None; - status.per_thread_util = backend_util; - } else if let (Some(previous), Some(current)) = - (status.last_cpu_sample.as_ref(), cpu.as_ref()) - { + if let (Some(previous), Some(current)) = (status.last_cpu_sample.as_ref(), cpu.as_ref()) { let wall_ticks = current .sampled_at .checked_duration_since(previous.sampled_at) @@ -291,26 +286,19 @@ impl Instance { update_worker_utilisation(&mut status, worker_util); } } - let io_ops_total = match snapshot.perf { + let io_ops_total = match &snapshot.perf { Some(perf) => perf .read_io_count .saturating_add(perf.write_io_count) .saturating_add(perf.other_io_count), None => 0, }; - if let Some(per_thread_util) = backend_util { - status.rolling.push_from_backend_util( - now, - io_ops_total, - per_thread_util, - snapshot.thread_count, - ); - } else if let Some(current) = cpu.as_ref() { + if let Some(current) = cpu.as_ref() { status .rolling .push_from_procfs_delta(now, io_ops_total, current.cpu_ticks); } - status.last_cpu_sample = if backend_util.is_none() { cpu } else { None }; + status.last_cpu_sample = cpu; status.throttled_usec_delta = match (status.cgroup, cgroup) { (Some(previous), Some(current)) => current .throttled_usec @@ -318,6 +306,7 @@ impl Instance { _ => 0, }; status.cgroup = cgroup; + *latest_snapshot = Some(snapshot); true } @@ -423,14 +412,12 @@ pub struct CgroupSample { /// Most recent mutable state for one VM. #[derive(Debug, Default, Clone)] pub struct InstanceStatus { - /// Current backend worker count. + /// Current worker count pub thread_count: u32, - /// Current backend-reported vCPU count. - pub vcpu_count: u32, /// Whether the last refresh succeeded. pub alive: bool, /// Whether backend ownership policy permits automatic scaling. - /// FIXME this was removed no back again? + // FIXME this was removed no back again? pub scaling_allowed: bool, /// Persisted ownership classification, or `None` before the first usable /// snapshot of a previously unknown VM. @@ -441,11 +428,6 @@ pub struct InstanceStatus { /// /// The override is deliberately cleared when the daemon restarts. pub manual_scaling_sticky: bool, - /// Latest backend performance counters, [`None`] if the backend didn't - /// provide any. - pub perf: Option, - /// Latest average CPU utilisation per backend task, from 0.0 to 1.0. - pub per_thread_util: f64, /// Previous cumulative CPU sample used to compute a delta. pub last_cpu_sample: Option, /// Latest comparable utilisation fraction for each sampled worker. @@ -573,20 +555,13 @@ impl InstancePerfSample { } } -/// One backend snapshot consumed by the controller and engine. +/// One raw backend snapshot consumed by the controller. #[derive(Debug, Clone)] pub struct ThreadPoolSnapshot { - /// Number of active I/O workers. - pub thread_count: u32, /// Current backend-reported vCPU count. pub vcpu_count: u32, /// Performance counters captured with the current worker count. pub perf: Option, - /// Backend-computed per-thread CPU utilisation. - /// - /// `None` asks the controller to sample the backend process through - /// `/proc` using [`Instance::thread_name_filter`]. - pub per_thread_util: Option, } /// Store one set of per-worker utilisation samples. @@ -607,11 +582,6 @@ fn update_worker_utilisation(status: &mut InstanceStatus, worker_util: Vec<(Stri } status.per_worker_util = worker_util.into_iter().map(|(_, util)| util).collect(); - status.per_thread_util = if status.per_worker_util.is_empty() { - 0.0 - } else { - status.per_worker_util.iter().sum::() / status.per_worker_util.len() as f64 - }; } /// Match task counters by TID and convert deltas to utilisation fractions. @@ -761,9 +731,7 @@ mod tests { use super::*; use crate::backends::BackendClientError; - struct SnapshotClient { - threads: u32, - } + struct SnapshotClient; #[async_trait] impl InstanceClient for SnapshotClient { @@ -773,10 +741,8 @@ mod tests { async fn get_thread_pool_snapshot(&self) -> Result { Ok(ThreadPoolSnapshot { - thread_count: self.threads, vcpu_count: 2, perf: None, - per_thread_util: None, }) } @@ -802,20 +768,16 @@ mod tests { } } - /// Test that a successful client snapshot updates - /// alive/thread_count state. + /// Test that a successful client snapshot updates liveness and raw state. #[tokio::test] async fn test_instance_refresh_state_applies_successful_snapshot() { - let instance = Instance::new( - "vm-ok".to_string(), - Path::new(""), - 7, - SnapshotClient { threads: 3 }, - ); + let instance = Instance::new("vm-ok".to_string(), Path::new(""), 7, SnapshotClient); assert!(instance.refresh_state(false).await); - let status = instance.status.read().await; - assert!(status.alive); - assert_eq!(status.thread_count, 3); + assert!(instance.status.read().await.alive); + assert_eq!( + instance.snapshot.read().await.as_ref().unwrap().vcpu_count, + 2 + ); } /// Test that client errors mark the instance dead and close the @@ -832,20 +794,14 @@ mod tests { }, ); assert!(!instance.refresh_state(false).await); - let status = instance.status.read().await; - assert!(!status.alive); + assert!(!instance.status.read().await.alive); assert!(closed.load(Ordering::Relaxed)); } /// Test that `Display` for an instance prints the bare VM id. #[test] fn test_display_trims_instance_id() { - let instance = Instance::new( - " vm-1 ".to_string(), - Path::new(""), - 1, - SnapshotClient { threads: 1 }, - ); + let instance = Instance::new(" vm-1 ".to_string(), Path::new(""), 1, SnapshotClient); assert_eq!(instance.to_string(), "vm-1"); } diff --git a/src/rolling.rs b/src/rolling.rs index fd9245ce..f4277311 100644 --- a/src/rolling.rs +++ b/src/rolling.rs @@ -81,43 +81,6 @@ impl RollingMetrics { self.trim(now); } - /// Append an interval from backend-computed per-thread utilisation. - pub fn push_from_backend_util( - &mut self, - now: Instant, - io_ops_cumulative: u64, - per_thread_util: f64, - thread_count: u32, - ) { - let (Some(previous_wall_ts), Some(previous_io_ops)) = - (self.previous_wall_ts, self.previous_io_ops) - else { - self.set_backend_baseline(now, io_ops_cumulative); - return; - }; - let Some(elapsed) = now.checked_duration_since(previous_wall_ts) else { - self.set_backend_baseline(now, io_ops_cumulative); - return; - }; - if io_ops_cumulative < previous_io_ops { - self.set_backend_baseline(now, io_ops_cumulative); - return; - } - - let wall_dt_ns = elapsed.as_nanos().min(u128::from(u64::MAX)) as u64; - let cpu_ns_delta = - (wall_dt_ns as f64 * per_thread_util.clamp(0.0, 1.0) * f64::from(thread_count)) - .min(u64::MAX as f64) as u64; - self.samples.push_back(TickSample { - wall_ts: now, - io_ops_delta: io_ops_cumulative - previous_io_ops, - cpu_ns_delta, - wall_dt_ns, - }); - self.set_backend_baseline(now, io_ops_cumulative); - self.trim(now); - } - /// Append an already-computed delta, used by fleet aggregation. pub fn push_delta(&mut self, now: Instant, io_ops_delta: u64, cpu_ns_delta: u64) { let Some(previous_wall_ts) = self.previous_wall_ts else { @@ -176,12 +139,6 @@ impl RollingMetrics { self.previous_cpu_ticks = Some(cpu_ticks); } - fn set_backend_baseline(&mut self, now: Instant, io_ops: u64) { - self.previous_wall_ts = Some(now); - self.previous_io_ops = Some(io_ops); - self.previous_cpu_ticks = None; - } - fn trim(&mut self, now: Instant) { // If `now` is still within MAX_WINDOW of the // opaque Instant origin, keep every sample. diff --git a/tests/component/conftest.py b/tests/component/conftest.py index a8d61c42..024e05ea 100644 --- a/tests/component/conftest.py +++ b/tests/component/conftest.py @@ -14,11 +14,6 @@ from harness.mockfs import MockProc -@pytest.fixture -def fake_backend(tmp_path): - return FakeBackend(tmp_path) - - @pytest.fixture def dbus_server(tmp_path): server = DBusServer(tmp_path) @@ -34,6 +29,13 @@ def mock_proc(tmp_path, dbus_server): proc.stop() +@pytest.fixture +def fake_backend(tmp_path, mock_proc): + backend = FakeBackend(tmp_path, mock_proc) + yield backend + backend.stop() + + @pytest.fixture def controller(tmp_path, fake_backend, dbus_server): del fake_backend # created first so the state directory exists diff --git a/tests/component/harness/fake_backend.py b/tests/component/harness/fake_backend.py index f6cbafad..dfd6197c 100644 --- a/tests/component/harness/fake_backend.py +++ b/tests/component/harness/fake_backend.py @@ -6,30 +6,46 @@ import fcntl import json import os +import threading -from harness.common import physical +from harness.common import POLL_S, physical DEFAULT_VM = "vm-a" STATE_DIR = "/var/fake" OWNERSHIP_FILE = "/run/io-thread-controller/vm-ownership.json" +def _task_stat(tid, name, ticks): + fields = ["0"] * 49 + fields[10] = str(ticks) # field 14: utime + fields[16] = "1" # field 20: num_threads + fields[18] = "1" # field 22: starttime + return ("%d (%s) S %s\n" % (tid, name, " ".join(fields))).encode() + + class FakeBackend: """JSON VM directory the ``fake`` backend discovers.""" - def __init__(self, root): + def __init__(self, root, mock_proc): self.root = root + self.mock_proc = mock_proc physical(root, STATE_DIR).mkdir(parents=True, exist_ok=True) self.state_dir = physical(root, STATE_DIR) self.default_vm = DEFAULT_VM + self._util = {} + self._published = {} + self._state_lock = threading.Lock() + self._stop = threading.Event() self.add_vm(DEFAULT_VM) + self._monitor = threading.Thread(target=self._monitor_procfs, daemon=True) + self._monitor.start() def add_vm(self, vm_id, **fields): """Create ``vm_id`` with idle defaults overridden by ``fields``.""" + util = fields.pop("per_thread_util", 0.0) defaults = { "thread_count": 1, "vcpu_count": 1, - "per_thread_util": 0.0, "managed": True, "fail_snapshot": False, "read_io_count": 0, @@ -38,10 +54,14 @@ def add_vm(self, vm_id, **fields): } defaults.update(fields) self._update(vm_id, **defaults) + self.set_util(util, vm_id) return vm_id def set_util(self, util, vm=None): - self._update(vm or self.default_vm, per_thread_util=util) + vm_id = vm or self.default_vm + with self._state_lock: + self._util[vm_id] = util + self._sync_procfs(vm_id) def set_threads(self, count, vm=None): self._update(vm or self.default_vm, thread_count=count) @@ -74,6 +94,17 @@ def prepare_healthy_vm(self, threads, util, io=0, vcpu=16): def thread_count(self, vm=None): return self._read(vm or self.default_vm)["thread_count"] + def pid(self, vm=None): + """Return the deterministic process ID used by the fake backend.""" + value = 0 + for byte in (vm or self.default_vm).encode(): + value = ((value * 31) + byte) & 0xFFFFFFFF + return 10_000 + (value % 1_000_000) + + def stop(self): + self._stop.set() + self._monitor.join(timeout=5) + def clear_calls(self): """Forget recorded ``set_thread_count`` calls for every VM.""" path = self.state_dir / "calls.json" @@ -119,3 +150,43 @@ def _update(self, vm_id, **fields): temporary.write_text(json.dumps(data)) os.replace(temporary, path) fcntl.flock(lock, fcntl.LOCK_UN) + + def _monitor_procfs(self): + while not self._stop.wait(0.02): + for path in self.state_dir.glob("*.json"): + if path.name == "calls.json": + continue + self._sync_procfs(path.stem) + + def _sync_procfs(self, vm_id): + path = self._vm_path(vm_id) + if not path.exists(): + return + data = self._read(vm_id) + thread_count = data["thread_count"] + with self._state_lock: + util = self._util.get(vm_id, 0.0) + previous = self._published.get(vm_id) + signature = (thread_count, util) + if previous == signature: + return + self._published[vm_id] = signature + + pid = self.pid(vm_id) + previous_count = previous[0] if previous is not None else 0 + for index in range(thread_count, previous_count): + self.mock_proc.unregister("/%d/task/%d" % (pid, pid + index + 1)) + + ticks_per_poll = round( + POLL_S * os.sysconf("SC_CLK_TCK") * max(0.0, min(1.0, util)) + ) + for index in range(thread_count): + tid = pid + index + 1 + values = [ + _task_stat(tid, "fake-worker-%d" % index, sample * ticks_per_poll) + for sample in range(1000) + ] + self.mock_proc.register_sequence( + "/%d/task/%d/stat" % (pid, tid), + values, + ) diff --git a/tests/component/harness/mockfs.py b/tests/component/harness/mockfs.py index 03017ad0..a508226a 100644 --- a/tests/component/harness/mockfs.py +++ b/tests/component/harness/mockfs.py @@ -6,6 +6,7 @@ import os import signal import subprocess +import threading import dbus @@ -29,6 +30,7 @@ def __init__(self, root, binary, dbus_server): self._proc = None self._bus = None self._iface = None + self._lock = threading.Lock() self._stderr_path = root / "mockfsd.stderr" self._stderr = open(self._stderr_path, "w") try: @@ -79,7 +81,13 @@ def register_sequence(self, path, values): [dbus.ByteArray(bytes(value)) for value in values], signature="ay", ) - self._iface.RegisterSequence(str(self.mount), path, payload) + with self._lock: + self._iface.RegisterSequence(str(self.mount), path, payload) + + def unregister(self, path): + """Remove ``path`` and any descendants from this mount.""" + with self._lock: + self._iface.Unregister(str(self.mount), path) def stop(self): """Unmount if possible, stop mockfsd, and close its stderr log.""" diff --git a/tests/component/test_controller.py b/tests/component/test_controller.py index 5fd3c68e..bfe9d032 100644 --- a/tests/component/test_controller.py +++ b/tests/component/test_controller.py @@ -141,14 +141,14 @@ def test_actuation_enforces_host_cpu_scale_up_ceiling( cooldown_secs=0 ) wait_for( - lambda: fake_backend.calls() == [2], + lambda: controller.stats() + and controller.stats()["vms"][0]["per_thread_util"] > 0.6, timeout=10, - description="first scale-up before the host ceiling is known", + description="procfs CPU sample while host ceiling is active", ) - fake_backend.clear_calls() time.sleep(POLL_S * 5) assert fake_backend.calls() == [], controller.logs() - assert fake_backend.thread_count() == 2 + assert fake_backend.thread_count() == 1 def test_tick_refreshes_evaluates_and_drops_failed_instances(