Respect cgroup limits

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

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
 }

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
 }

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);
+    }
 }

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()

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<Instance>, 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));

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<InstanceStatus>,
 }
 
+#[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<f64>,
     /// Sorted names from the latest worker sample, for roster-change logging.
     pub last_worker_names: Option<Vec<String>>,
+    /// Latest cgroup v2 CPU quota and throttle counters.
+    pub cgroup: Option<CgroupSample>,
+    /// 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<CgroupSample, CgroupError> {
+    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::<f64>()?;
+    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::<f64>()? / 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"));
     }