diff --git a/src/backends/fake/mod.rs b/src/backends/fake/mod.rs index 65568a6..1a42616 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 c2a339b..8263e47 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 cd89899..49a975e 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 6d9f73e..a2afa3e 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 4d6ac05..b1dfc43 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 d28cbcc..97b0425 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 3ee27d3..3cef844 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 4a55aa2..1e6969e 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 1ed0d6b..6998c9c 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 fd9245c..f427731 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 a8d61c4..024e05e 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 f6cbafa..dfd6197 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 03017ad..a508226 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 5fd3c68..bfe9d03 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(