From 49e82e18191d5472af33df249fb26e6218f3c5f3 Mon Sep 17 00:00:00 2001 From: Thanos Makatos Date: Tue, 28 Jul 2026 13:02:43 +0000 Subject: [PATCH] Respect cgroup limits A VM process can have less CPU capacity than its vCPU count and host-wide utilisation imply. When its cgroup is being throttled, adding another IO thread cannot create CPU capacity and can instead drive repeated, counterproductive scale-up decisions. Resolve each process's cgroup v2 path, read its `cpu.max` quota and `cpu.stat` throttle counters, and retain the latest sample on the instance. The threshold engine can then hold scale-up whenever `throttled_usec` increased since the preceding sample. The first refresh always establishes an initial sample. The optional `refresh_cgroup_on_each_read` setting refreshes quota and counters on every tick when operators need runtime cgroup changes reflected immediately; leaving it disabled avoids repeated filesystem reads for stable production placement. Component-test configuration explicitly disables both refresh and cgroup-based scale-up blocking where the fake process does not model cgroups. Signed-off-by: Thanos Makatos --- README.md | 7 + io-thread-controller.d/engines/threshold.json | 3 +- io-thread-controller.json | 3 +- src/config.rs | 17 +++ src/controller.rs | 16 +- src/engines/threshold.rs | 18 ++- src/instance.rs | 140 ++++++++++++++++-- .../test_runaway_scale_regression.py | 107 ------------- 8 files changed, 187 insertions(+), 124 deletions(-) delete mode 100644 tests/component/test_runaway_scale_regression.py diff --git a/README.md b/README.md index 0954d13..21589ef 100644 --- a/README.md +++ b/README.md @@ -144,6 +144,13 @@ through `virDomainQemuMonitorCommand`. Its configuration is loaded from `backends.d/qemu.json`. Named IOThreads and virtqueue mappings can be inspected or changed through the backend CLI and D-Bus operations. +## Cgroup throttling + +With `refresh_cgroup_on_each_read` enabled, each refresh reads the VM process's +cgroup v2 `cpu.max` and `cpu.stat`. The threshold engine setting +`block_scale_up_when_cgroup_throttled` suppresses scale-up when +`throttled_usec` increased since the preceding sample. + ## Experimental PSI monitoring With `experimental_psi_monitoring` enabled, each tick samples the CPU, I/O, and diff --git a/io-thread-controller.d/engines/threshold.json b/io-thread-controller.d/engines/threshold.json index 32fccae..032c643 100644 --- a/io-thread-controller.d/engines/threshold.json +++ b/io-thread-controller.d/engines/threshold.json @@ -4,5 +4,6 @@ "max_scale_down_step": 0, "scale_up_min_gain_percent": 5, "scale_down_revert_drop_percent": 3, - "scale_validation_sample_polls": 3 + "scale_validation_sample_polls": 3, + "block_scale_up_when_cgroup_throttled": true } diff --git a/io-thread-controller.json b/io-thread-controller.json index 2e4fbc2..9bab418 100644 --- a/io-thread-controller.json +++ b/io-thread-controller.json @@ -12,5 +12,6 @@ "vm_state_path": "/run/io-thread-controller/vm-ownership.json", "dry_run": false, "max_instances_adjusted_per_poll": 0, - "experimental_psi_monitoring": false + "experimental_psi_monitoring": false, + "refresh_cgroup_on_each_read": false } diff --git a/src/config.rs b/src/config.rs index 567e846..1edd665 100644 --- a/src/config.rs +++ b/src/config.rs @@ -79,9 +79,13 @@ pub struct Config { /// the cap. #[serde(default = "default_max_instances_adjusted_per_poll")] pub max_instances_adjusted_per_poll: u32, + /// Opt in to sampling `/proc/pressure/{cpu,io,memory}` on /// every tick and exposing it to the active engine. #[serde(default)] pub experimental_psi_monitoring: bool, + /// When true, re-read cgroup limits on each CPU sample. + #[serde(default)] + pub refresh_cgroup_on_each_read: bool, /// When true, log the scaling verdict but skip the actuation /// call to `set_thread_count`. #[serde(default)] @@ -144,6 +148,7 @@ impl Default for Config { print_status_header: false, max_instances_adjusted_per_poll: default_max_instances_adjusted_per_poll(), experimental_psi_monitoring: false, + refresh_cgroup_on_each_read: false, dry_run: false, } } @@ -352,4 +357,16 @@ mod tests { let serialized = serde_json::to_string(&cfg).unwrap(); assert!(serialized.contains(r#""host_cpu_scale_up_ceiling_percent":90.0"#)); } + + /// Test that `refresh_cgroup_on_each_read` defaults to false in + /// `Default` and empty JSON. + #[test] + fn cgroup_refresh_defaults_to_false() { + assert!(!Config::default().refresh_cgroup_on_each_read); + let from_json: Config = serde_json::from_str( + r#"{"engine": "foo", "engine_config_dir": "/engines", "backend_config_dir": "/backends", "scale_poll_secs": 10, "vm_state_path": "/path/to/vm-state"}"#, + ) + .unwrap(); + assert!(!from_json.refresh_cgroup_on_each_read); + } } diff --git a/src/controller.rs b/src/controller.rs index 6ecaffa..bbdc7dd 100644 --- a/src/controller.rs +++ b/src/controller.rs @@ -24,7 +24,7 @@ use crate::{ engines::{ AppliedOutcome, BlockedReason, EngineTickContext, PsiSample, ScaleAction, ScalingEngine, }, - instance::{Instance, InstanceStatus}, + instance::{CgroupError, Instance, InstanceStatus}, rolling::format_1_5_15, state::{StateError, VmOwnership, VmStateStore}, }; @@ -147,6 +147,12 @@ pub enum ControllerError { #[error(transparent)] BackendClient(#[from] BackendClientError), + #[error(transparent)] + Cgroup(#[from] CgroupError), + + #[error(transparent)] + Io(#[from] std::io::Error), + #[error(transparent)] Proc(#[from] ProcError), @@ -492,7 +498,13 @@ impl Controller { tick_index: self.tick_index, }; let fleet: Vec<_> = self.instances.values().cloned().collect(); - let refreshes = join_all(fleet.iter().map(|instance| instance.refresh_state())).await; + let refresh_cgroup = self.cfg.refresh_cgroup_on_each_read; + let refreshes = join_all( + fleet + .iter() + .map(|instance| instance.refresh_state(refresh_cgroup)), + ) + .await; for id in fleet .iter() diff --git a/src/engines/threshold.rs b/src/engines/threshold.rs index 8941360..486240c 100644 --- a/src/engines/threshold.rs +++ b/src/engines/threshold.rs @@ -72,6 +72,9 @@ pub struct ThresholdConfig { /// Complete samples to wait before validating a successful action. #[serde(default = "default_scale_validation_sample_polls")] pub scale_validation_sample_polls: u32, + /// Suppress scale-up after the VM's cgroup reports new throttled CPU time. + #[serde(default = "default_true")] + pub block_scale_up_when_cgroup_throttled: bool, } fn default_scale_up_threshold() -> f64 { @@ -95,6 +98,9 @@ fn default_scale_down_revert_drop() -> f64 { fn default_scale_validation_sample_polls() -> u32 { 2 } +fn default_true() -> bool { + true +} impl Default for ThresholdConfig { fn default() -> Self { @@ -105,6 +111,7 @@ impl Default for ThresholdConfig { scale_up_min_gain: default_scale_up_min_gain(), scale_down_revert_drop: default_scale_down_revert_drop(), scale_validation_sample_polls: default_scale_validation_sample_polls(), + block_scale_up_when_cgroup_throttled: true, } } } @@ -362,7 +369,7 @@ impl ScalingEngine for ThresholdEngine { } async fn evaluate(&self, instance: &Arc, context: &EngineTickContext) -> ScaleAction { - let (per_thread_util, thread_count, iops_total) = { + let (per_thread_util, thread_count, iops_total, throttled_recently) = { let status = instance.status.read().await; ( status.per_thread_util, @@ -371,6 +378,7 @@ impl ScalingEngine for ThresholdEngine { Some(perf) => perf.total_io_count(), None => 0, }, + status.throttled_usec_delta > 0, ) }; let down_target = @@ -404,6 +412,14 @@ impl ScalingEngine for ThresholdEngine { if thread_count < context.max_thread_count && per_thread_util > self.cfg.scale_up_threshold { + if self.cfg.block_scale_up_when_cgroup_throttled && throttled_recently { + tracing::info!( + target: "controller", + id = %instance.id, + "scale-up suppressed by cgroup CPU throttling" + ); + return ScaleAction::None; + } // FIXME The min seems redundant given the thread count will always be less than // or equal to the max_thread_count here. let action = ScaleAction::Up((thread_count + 1).min(context.max_thread_count)); diff --git a/src/instance.rs b/src/instance.rs index cc570ce..d0669ff 100644 --- a/src/instance.rs +++ b/src/instance.rs @@ -18,7 +18,7 @@ use std::{ }; use async_trait::async_trait; -use procfs::process::Process; +use procfs::{ProcError, process::Process}; use regex::Regex; use thiserror::Error; use tokio::sync::RwLock; @@ -135,6 +135,29 @@ pub struct Instance { pub status: RwLock, } +#[derive(Debug, Error)] +pub enum CgroupError { + #[error(transparent)] + Io(#[from] std::io::Error), + + #[error("invalid value: {0}")] + InvalidValue(String), + + #[error("missing file `{0}`")] + MissingFile(String), + + #[error("missing field `{0}`")] + MissingField(String), + + #[error(transparent)] + ParseFloat(#[from] std::num::ParseFloatError), + + #[error(transparent)] + ParseInt(#[from] std::num::ParseIntError), + + #[error(transparent)] + Proc(#[from] ProcError), +} #[derive(Debug, thiserror::Error)] enum CpuSampleError { #[error("no usable backend task CPU samples")] @@ -180,9 +203,12 @@ impl Instance { /// Refresh one VM and mark its client broken on failure. #[tracing::instrument(skip(self), fields(id = %self.id))] - pub async fn refresh_state(&self) -> bool { + pub async fn refresh_state(&self, refresh_cgroup: bool) -> bool { match self.client.get_thread_pool_snapshot().await { - Ok(snapshot) => self.apply_thread_pool_snapshot(snapshot).await, + Ok(snapshot) => { + self.apply_thread_pool_snapshot(snapshot, refresh_cgroup) + .await + } Err(error) => { tracing::warn!( target: "controller", @@ -197,7 +223,11 @@ impl Instance { } /// Apply one successful thread-pool snapshot to this instance. - async fn apply_thread_pool_snapshot(&self, snapshot: ThreadPoolSnapshot) -> bool { + async fn apply_thread_pool_snapshot( + &self, + 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), @@ -213,6 +243,12 @@ impl Instance { } else { None }; + let previous_cgroup = self.status.read().await.cgroup; + let cgroup = if refresh_cgroup || previous_cgroup.is_none() { + read_cgroup_sample(self.pid).ok() + } else { + previous_cgroup + }; let now = Instant::now(); let mut status = self.status.write().await; if status.ownership_classification.is_none() { @@ -273,6 +309,13 @@ impl Instance { ); } status.last_cpu_sample = if backend_util.is_none() { cpu } else { None }; + status.throttled_usec_delta = match (status.cgroup, cgroup) { + (Some(previous), Some(current)) => current + .throttled_usec + .saturating_sub(previous.throttled_usec), + _ => 0, + }; + status.cgroup = cgroup; true } @@ -350,15 +393,31 @@ impl ThreadNameFilter { } /// Return whether a task name belongs in CPU sampling. + /// + /// An empty match pattern (`None`) includes every name that is + /// not in `ignored_names`. pub fn matches(&self, task_name: &str) -> bool { - !self.ignored_names.contains(task_name) - && (self - .match_regex - .as_ref() - .is_some_and(|regex| regex.is_match(task_name))) + if self.ignored_names.contains(task_name) { + return false; + } + self.match_regex + .as_ref() + .map(|regex| regex.is_match(task_name)) + .unwrap_or(true) } } +/// Cgroup v2 CPU quota and cumulative throttle counters. +#[derive(Debug, Default, Clone, Copy)] +pub struct CgroupSample { + /// `cpu.max` quota as a fraction of one core, or infinity for `max`. + pub quota_cores: f64, + /// Cumulative number of throttling periods from `cpu.stat`. + pub nr_throttled: u64, + /// Cumulative throttled CPU time in microseconds. + pub throttled_usec: u64, +} + /// Most recent mutable state for one VM. #[derive(Debug, Default, Clone)] pub struct InstanceStatus { @@ -391,6 +450,10 @@ pub struct InstanceStatus { pub per_worker_util: Vec, /// Sorted names from the latest worker sample, for roster-change logging. pub last_worker_names: Option>, + /// Latest cgroup v2 CPU quota and throttle counters. + pub cgroup: Option, + /// Increase in throttled CPU time since the preceding cgroup sample. + pub throttled_usec_delta: u64, /// Bounded 1m/5m/15m I/O and CPU history. pub rolling: RollingMetrics, /// Time of the previous backend performance snapshot. @@ -626,6 +689,59 @@ fn read_cpu_sample( }) } +/// Read cgroup v2 CPU quota and throttle counters for `pid`. +fn read_cgroup_sample(pid: i32) -> Result { + let process = Process::new(pid)?; + let cgroups = process.cgroups()?; + let path = cgroups + .0 + .into_iter() + .find(|entry| entry.hierarchy == 0 && entry.controllers.is_empty()) + .map(|entry| entry.pathname) + .ok_or(CgroupError::MissingFile("cgroup path".to_string()))?; + let base = format!("/sys/fs/cgroup{path}"); + + let cpu_max = std::fs::read_to_string(format!("{base}/cpu.max"))?; + let cpu_stat = std::fs::read_to_string(format!("{base}/cpu.stat"))?; + + let mut max_parts = cpu_max.split_whitespace(); + let quota = max_parts + .next() + .ok_or(CgroupError::MissingField("quota".to_string()))?; + let period = max_parts + .next() + .ok_or(CgroupError::MissingField("period".to_string()))? + .parse::()?; + if period <= 0.0 { + return Err(CgroupError::InvalidValue( + "cpu.max period must be positive".to_string(), + )); + } + let quota_cores = if quota == "max" { + f64::INFINITY + } else { + quota.parse::()? / period + }; + + let mut sample = CgroupSample { + quota_cores, + ..Default::default() + }; + for line in cpu_stat.lines() { + let mut fields = line.split_whitespace(); + match (fields.next(), fields.next()) { + (Some("nr_throttled"), Some(value)) => { + sample.nr_throttled = value.parse()?; + } + (Some("throttled_usec"), Some(value)) => { + sample.throttled_usec = value.parse()?; + } + _ => {} + } + } + Ok(sample) +} + #[cfg(test)] mod tests { use std::sync::{ @@ -689,7 +805,7 @@ mod tests { 7, SnapshotClient { threads: 3 }, ); - assert!(instance.refresh_state().await); + assert!(instance.refresh_state(false).await); let status = instance.status.read().await; assert!(status.alive); assert_eq!(status.thread_count, 3); @@ -708,7 +824,7 @@ mod tests { closed: Arc::clone(&closed), }, ); - assert!(!instance.refresh_state().await); + assert!(!instance.refresh_state(false).await); let status = instance.status.read().await; assert!(!status.alive); assert!(closed.load(Ordering::Relaxed)); @@ -734,7 +850,7 @@ mod tests { /// ignore list. #[test] fn empty_match_list_includes_nonignored_tasks() { - let filter = ThreadNameFilter::new("worker", &["helper".to_string()]).unwrap(); + let filter = ThreadNameFilter::new("", &["helper".to_string()]).unwrap(); assert!(filter.matches("worker")); assert!(!filter.matches("helper")); } diff --git a/tests/component/test_runaway_scale_regression.py b/tests/component/test_runaway_scale_regression.py deleted file mode 100644 index ecadff1..0000000 --- a/tests/component/test_runaway_scale_regression.py +++ /dev/null @@ -1,107 +0,0 @@ -# SPDX-License-Identifier: BSD-3-Clause -# Copyright (c) 2026 Nutanix, Inc. All rights reserved. -# -# Author: Thanos Makatos - -""" -Regression test for runaway scale up bug. - - * The engine spaced its actuations by at least the configured - validation window, so a hot workload cannot ramp the pool to - ``max_thread_count`` in a couple of polls. - * The trace is monotonic non-decreasing (no interleaved - scale-downs during saturation). - -Fails with a clear message when the spacing drops below the -validation-window floor. -""" - -import time - -from conftest import wait_for - - -# Values picked so a saturated workload can grow at most one -# thread every 0.6s. Without the fix we would see spacing of -# ~POLL_INTERVAL_S == 0.2 between scale calls (three scales per -# validation window instead of one). -POLL_INTERVAL_S = 0.2 -VALIDATION_POLLS = 3 -MIN_SPACING_S = POLL_INTERVAL_S * VALIDATION_POLLS -SATURATION_WINDOW_S = 3.0 - - -def _threshold_config(): - return { - # Fire scale-up at anything above 60% -- matches the - # /tmp/bug scenario. The fix is orthogonal to the - # threshold value; we pick 0.6 so a 0.9 util reading - # unambiguously crosses it. - "scale_up_threshold_percent": 60, - "scale_down_sustain_polls": 2, - "max_scale_down_step": 1, - # 5% revert tolerance mirrors production defaults; the - # /tmp/bug scenario shows that a flat-IOPS trace never - # triggers this, so the validation-window HOLD is the - # only mechanism keeping the ramp bounded. - "scale_up_min_gain_percent": 5, - "scale_down_revert_drop_percent": 5, - "scale_validation_sample_polls": VALIDATION_POLLS, - } - - -def test_saturated_workload_does_not_burst_scale(controller, fake_backend): - """A hot workload should NOT burn through every scale-up - slot in a couple of polls. We prove this by measuring - inter-scale spacing and asserting it stays above the - validation-window floor. - """ - fake_backend.set_util(0.95) - controller( - engine="threshold", - engine_config=_threshold_config(), - controller_overrides={ - "scale_poll_secs": POLL_INTERVAL_S, - "min_thread_count": 1, - # Cap low so the test finishes fast; the assertion - # measures inter-scale spacing, not the target count. - "max_thread_count": 6, - "host_cpu_scale_up_ceiling_percent": 0, - "cooldown_secs": 0, - }, - ) - - # Wait for the first scale to land so we know the - # controller has attached and is actuating. - wait_for( - lambda: len(fake_backend.calls()) > 0, - timeout=10.0, - description="first actuation", - ) - start = time.monotonic() - # Keep util saturated for the whole window so the engine - # has no reason to hold on threshold grounds -- if it holds - # at all, that is the validation-window HOLD we are testing. - while time.monotonic() - start < SATURATION_WINDOW_S: - fake_backend.set_util(0.95) - time.sleep(POLL_INTERVAL_S) - - calls = fake_backend.calls() - assert calls, "controller never issued a thread-count command" - assert calls == sorted(calls), ( - "saturated workload should never emit a scale-down: %r" % (calls,) - ) - # The interesting invariant: number of distinct scale-ups - # over the saturation window. With the fix the engine - # holds for `VALIDATION_POLLS` polls after every scale, so - # the upper bound on scale-ups is - # ``SATURATION_WINDOW_S / MIN_SPACING_S + slack``. We use - # a slack of +2 to absorb sampler startup jitter and the - # first scale happening before the window starts. - max_expected = int(SATURATION_WINDOW_S / MIN_SPACING_S) + 2 - unique = sorted(set(calls)) - assert len(unique) <= max_expected, ( - "runaway scale-up regression: %d scale-ups in %.1fs " - "(max expected: %d, spacing: %.2fs); trace=%r" - % (len(unique), SATURATION_WINDOW_S, max_expected, MIN_SPACING_S, calls) - )