From 1250cbe18237cd0ff88a4231ab8e128d7c8a775e Mon Sep 17 00:00:00 2001 From: Xuewei Niu Date: Fri, 11 Jul 2025 20:55:23 +0800 Subject: [PATCH] manager: Introduce FsManager `Manager` is a trait to unify the interface of cgroups. It is designed for OCI containers. Its `set()` takes Linux resources of the OCI spec to set cgroups. The `FsManager`, the concrete implementation of `Manager`, manipulates cgroups through cgroupfs, and supports both cgroups v1 and v2. Signed-off-by: Xuewei Niu --- Cargo.toml | 2 + src/fs/freezer.rs | 13 +- src/fs/mod.rs | 21 +- src/lib.rs | 42 ++ src/manager/conv.rs | 69 +++ src/manager/error.rs | 20 + src/manager/fs.rs | 1088 ++++++++++++++++++++++++++++++++++++++++++ src/manager/mod.rs | 77 +++ src/stats.rs | 156 ++++++ 9 files changed, 1456 insertions(+), 32 deletions(-) create mode 100644 src/manager/conv.rs create mode 100644 src/manager/error.rs create mode 100644 src/manager/fs.rs create mode 100644 src/manager/mod.rs create mode 100644 src/stats.rs diff --git a/Cargo.toml b/Cargo.toml index db0f7c5..4448783 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -17,9 +17,11 @@ nix = { version = "0.25.0", default-features = false, features = ["event", "fs", libc = "0.2" serde = { version = "1.0", features = ["derive"], optional = true } thiserror = "1" +oci-spec = { version = "0.8.1", optional = true } [dev-dependencies] libc = "0.2.76" [features] default = [] +oci = ["oci-spec"] diff --git a/src/fs/freezer.rs b/src/fs/freezer.rs index 907dffb..036f6bb 100644 --- a/src/fs/freezer.rs +++ b/src/fs/freezer.rs @@ -13,8 +13,8 @@ use std::path::PathBuf; use crate::fs::error::ErrorKind::*; use crate::fs::error::*; - use crate::fs::{ControllIdentifier, ControllerInternal, Controllers, Resources, Subsystem}; +use crate::FreezerState; /// A controller that allows controlling the `freezer` subsystem of a Cgroup. /// @@ -31,17 +31,6 @@ pub struct FreezerController { v2: bool, } -/// The current state of the control group -#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] -pub enum FreezerState { - /// The processes in the control group are _not_ frozen. - Thawed, - /// The processes in the control group are in the processes of being frozen. - Freezing, - /// The processes in the control group are frozen. - Frozen, -} - impl ControllerInternal for FreezerController { fn control_type(&self) -> Controllers { Controllers::Freezer diff --git a/src/fs/mod.rs b/src/fs/mod.rs index 81aa115..f5255a4 100644 --- a/src/fs/mod.rs +++ b/src/fs/mod.rs @@ -245,6 +245,7 @@ mod sealed { } pub(crate) use crate::fs::sealed::{ControllerInternal, CustomizedAttribute}; +use crate::CgroupPid; /// A Controller is a subsystem attached to the control group. /// @@ -771,26 +772,6 @@ pub struct Resources { pub blkio: BlkIoResources, } -/// A structure representing a `pid`. Currently implementations exist for `u64` and -/// `std::process::Child`. -#[derive(Debug, PartialEq, Eq, PartialOrd, Ord)] -pub struct CgroupPid { - /// The process identifier - pub pid: u64, -} - -impl From for CgroupPid { - fn from(u: u64) -> CgroupPid { - CgroupPid { pid: u } - } -} - -impl From<&std::process::Child> for CgroupPid { - fn from(u: &std::process::Child) -> CgroupPid { - CgroupPid { pid: u.id() as u64 } - } -} - impl Subsystem { fn enter(self, path: &Path) -> Self { match self { diff --git a/src/lib.rs b/src/lib.rs index 2528ecf..4bb4a3b 100644 --- a/src/lib.rs +++ b/src/lib.rs @@ -5,3 +5,45 @@ // pub mod fs; +#[cfg(feature = "oci")] +pub mod manager; +#[cfg(feature = "oci")] +pub use manager::{FsManager, Manager}; +pub mod stats; +pub use stats::CgroupStats; + +/// The maximum value for CPU shares in cgroups v1 +pub const CPU_SHARES_V1_MAX: u64 = 262144; +/// The maximum value for CPU weight in cgroups v2 +pub const CPU_WEIGHT_V2_MAX: u64 = 10000; + +/// The current state of the control group +#[cfg_attr(feature = "serde", derive(serde::Serialize, serde::Deserialize))] +pub enum FreezerState { + /// The processes in the control group are _not_ frozen. + Thawed, + /// The processes in the control group are in the processes of being frozen. + Freezing, + /// The processes in the control group are frozen. + Frozen, +} + +/// A structure representing a `pid`. Currently implementations exist for `u64` and +/// `std::process::Child`. +#[derive(Debug, PartialEq, Eq, PartialOrd, Ord, Clone, Copy)] +pub struct CgroupPid { + /// The process identifier + pub pid: u64, +} + +impl From for CgroupPid { + fn from(u: u64) -> CgroupPid { + CgroupPid { pid: u } + } +} + +impl From<&std::process::Child> for CgroupPid { + fn from(u: &std::process::Child) -> CgroupPid { + CgroupPid { pid: u.id() as u64 } + } +} diff --git a/src/manager/conv.rs b/src/manager/conv.rs new file mode 100644 index 0000000..168c40c --- /dev/null +++ b/src/manager/conv.rs @@ -0,0 +1,69 @@ +// Copyright (c) 2025 Ant Group +// +// SPDX-License-Identifier: Apache-2.0 or MIT +// + +use crate::manager::error::{Error, Result}; +use crate::{CPU_SHARES_V1_MAX, CPU_WEIGHT_V2_MAX}; + +// Converts CPU shares, used by cgroup v1, to CPU weight, used by cgroup +// v2. +// +// Cgroup v1 CPU shares has a range of [2^1...2^18], i.e. [2...262144], +// and the default value is 1024. +// +// Cgroup v2 CPU weight has a range of [10^0...10^4], i.e. [1...10000], +// and the default value is 100. +pub(crate) fn cpu_shares_to_cgroup_v2(shares: u64) -> u64 { + if shares == 0 { + return 0; + } + if shares <= 2 { + return 1; + } + if shares >= CPU_SHARES_V1_MAX { + return CPU_WEIGHT_V2_MAX; + } + + (((shares - 2) * 9999) / 262142) + 1 +} + +// ConvertMemorySwapToCgroupV2Value converts MemorySwap value from OCI spec +// for use by cgroup v2 drivers. A conversion is needed since +// Resources.MemorySwap is defined as memory+swap combined, while in cgroup +// v2 swap is a separate value. +pub(crate) fn memory_swap_to_cgroup_v2(memswap_limit: i64, mem_limit: i64) -> Result { + // For compatibility with cgroup1 controller, set swap to unlimited in + // case the memory is set to unlimited, and swap is not explicitly set, + // treating the request as "set both memory and swap to unlimited". + if mem_limit == -1 && memswap_limit == 0 { + return Ok(-1); + } + + // -1 is "max", 0 is "unset", so treat as is + if memswap_limit == -1 || memswap_limit == 0 { + return Ok(memswap_limit); + } + + // Unlimited memory, so treat swap as is. + if mem_limit == -1 { + return Ok(memswap_limit); + } + + // Unset or unknown memory, can't calculate swap. + if mem_limit == 0 { + return Err(Error::InvalidLinuxResource); + } + + // Does not make sense to subtract a negative value. + if mem_limit < 0 { + return Err(Error::InvalidLinuxResource); + } + + // Sanity check. + if memswap_limit < mem_limit { + return Err(Error::InvalidLinuxResource); + } + + Ok(memswap_limit - mem_limit) +} diff --git a/src/manager/error.rs b/src/manager/error.rs new file mode 100644 index 0000000..5b4b3cb --- /dev/null +++ b/src/manager/error.rs @@ -0,0 +1,20 @@ +// Copyright (c) 2025 Ant Group +// +// SPDX-License-Identifier: Apache-2.0 or MIT +// + +use crate::fs::error::Error as CgroupfsError; + +pub type Result = std::result::Result; + +#[derive(thiserror::Error, Debug)] +pub enum Error { + #[error("invalid argument")] + InvalidArgument, + + #[error("invalid linux resource")] + InvalidLinuxResource, + + #[error("cgroupfs error: {0}")] + Cgroupfs(#[from] CgroupfsError), +} diff --git a/src/manager/fs.rs b/src/manager/fs.rs new file mode 100644 index 0000000..b1316b2 --- /dev/null +++ b/src/manager/fs.rs @@ -0,0 +1,1088 @@ +// Copyright (c) 2025 Ant Group +// +// SPDX-License-Identifier: Apache-2.0 or MIT +// + +use std::collections::HashMap; +use std::fs; +use std::path::Path; +use std::str::FromStr; + +use oci_spec::runtime::{ + LinuxBlockIo, LinuxCpu, LinuxDeviceCgroup, LinuxHugepageLimit, LinuxMemory, LinuxNetwork, + LinuxPids, LinuxResources, +}; + +use crate::fs::blkio::{BlkIoController, BlkIoData, IoService, IoStat}; +use crate::fs::cgroup::UNIFIED_MOUNTPOINT; +use crate::fs::cpu::CpuController; +use crate::fs::cpuacct::CpuAcctController; +use crate::fs::cpuset::CpuSetController; +use crate::fs::devices::{DevicePermissions, DeviceType, DevicesController}; +use crate::fs::error::{Error as FsError, ErrorKind as FsErrorKind, Result as FsResult}; +use crate::fs::freezer::FreezerController; +use crate::fs::hugetlb::HugeTlbController; +use crate::fs::memory::MemController; +use crate::fs::net_cls::NetClsController; +use crate::fs::net_prio::NetPrioController; +use crate::fs::pid::PidController; +use crate::fs::{hierarchies, Cgroup, ControllIdentifier, Controller, MaxValue, Subsystem}; +use crate::manager::error::Error; +use crate::manager::{conv, Manager, Result}; +use crate::stats::{ + BlkioCgroupStats, BlkioStat, CpuAcctStats, CpuCgroupStats, CpuThrottlingStats, + HugeTlbCgroupStats, HugeTlbStat, MemoryCgroupStats, MemoryStats, PidsCgroupStats, +}; +use crate::{CgroupPid, CgroupStats, FreezerState}; + +const CGROUP_PATH: &str = "/proc/self/cgroup"; +const MOUNTINFO_PATH: &str = "/proc/self/mountinfo"; + +/// FsManager manages cgroups using the cgroup filesystem (cgroupfs). +/// +/// This manager deals with `LinuxResources` conformed to the OCI runtime +/// specification, so that it allows users not to do type conversions. +#[derive(Debug, Clone)] +pub struct FsManager { + /// Cgroup subsystem paths read from `/proc/self/cgroup` + /// - cgroup v1: -> + /// - cgroup v2: "" -> + paths: HashMap, + /// Cgroup mountpoints read from `/proc/self/mountinfo`. + mounts: HashMap, + /// Base path of the cgroup filesystem, the complete path would be: + /// - cgroup v1: "/sys/fs/cgroup//" + /// - cgroup v2: "/sys/fs/cgroup/" + base: String, + /// Cgroup managed by this manager. + cgroup: Cgroup, +} + +impl FsManager { + /// Check if the cgroup exists or not. + pub fn exists(&self) -> bool { + self.cgroup.exists() + } + + /// Create an instance of FsManager. The cgroups won't be created until + /// `apply()` is called. + pub fn new(base: &str) -> Result { + let paths = parse_cgroup_subsystems()?; + let mounts = parse_cgroup_mountinfo(&paths)?; + let cgroup = Cgroup::load(hierarchies::auto(), base); + let base = base.to_string(); + + Ok(Self { + paths, + mounts, + base, + cgroup, + }) + } +} + +impl FsManager { + /// Create the cgroups if they are not created yet. + pub(crate) fn create_cgroups(&mut self) -> Result<()> { + if self.exists() { + return Ok(()); + } + self.cgroup.create()?; + Ok(()) + } + + /// Get the subcgroup path, which is useful for Docker-in-Docker (DinD) + /// with cgroup v2, see [1]. + /// + /// 1: https://github.com/kata-containers/kata-containers/issues/10733 + pub fn subcgroup(&self) -> &str { + // Check if we're in a Docker-in-Docker setup by verifying: + // 1. We're using cgroups v2 (which restricts direct process control) + // 2. An "init" subdirectory exists (used by DinD for process + // delegation) + let init_exists = hierarchies::auto() + .root() + .join(&self.base) + .join("init") + .exists(); + let is_dind = self.v2() && init_exists; + + if is_dind { + "/init/" + } else { + "/" + } + } + + fn controller<'a, T>(&'a self) -> FsResult<&'a T> + where + &'a T: From<&'a Subsystem>, + T: Controller + ControllIdentifier, + { + let controller: &T = self + .cgroup + .controller_of() + .ok_or(FsError::new(FsErrorKind::SubsystemsEmpty))?; + + Ok(controller) + } + + fn set_cpuset(&self, linux_cpu: &LinuxCpu) -> Result<()> { + let controller: &CpuSetController = self.controller()?; + + if let Some(cpus) = linux_cpu.cpus() { + controller.set_cpus(cpus)?; + } + + if let Some(mems) = linux_cpu.mems() { + controller.set_mems(mems)?; + } + + Ok(()) + } + + fn set_cpu(&self, linux_cpu: &LinuxCpu) -> Result<()> { + let controller: &CpuController = self.controller()?; + + if let Some(shares) = linux_cpu.shares() { + let shares = if self.v2() { + conv::cpu_shares_to_cgroup_v2(shares) + } else { + shares + }; + if shares != 0 { + controller.set_shares(shares)?; + } + } + + if let Some(quota) = linux_cpu.quota() { + controller.set_cfs_quota(quota)?; + } + + if let Some(period) = linux_cpu.period() { + controller.set_cfs_period(period)?; + } + + if let Some(rt_runtime) = linux_cpu.realtime_runtime() { + controller.set_rt_runtime(rt_runtime)?; + } + + if let Some(rt_period) = linux_cpu.realtime_period() { + controller.set_rt_period_us(rt_period)?; + } + + Ok(()) + } + + fn set_mem_and_memswap_v1(&self, limit: i64, mut swap_limit: i64) -> Result<()> { + let controller: &MemController = self.controller()?; + + // If the memory update is set to -1 and the swap is not set + // explicitly, we should also set swap to -1, it means + // unlimited memory. + if limit == -1 && swap_limit == 0 { + swap_limit = -1; + } + + if limit != 0 && swap_limit != 0 { + let memory = controller.memory_stat(); + let limit_actual = memory.limit_in_bytes; + + // When update memory limit, we should adapt the write sequence + // for memory and swap memory, so it won't fail because the new + // value and the old value don't fit kernel's validation. + if swap_limit == -1 || limit_actual < swap_limit { + controller.set_memswap_limit(swap_limit)?; + controller.set_limit(limit)?; + + return Ok(()); + } + } + + if limit != 0 { + controller.set_limit(limit)?; + } + if swap_limit != 0 { + controller.set_memswap_limit(swap_limit)?; + } + + Ok(()) + } + + fn set_memory_v1(&self, linux_memory: &LinuxMemory) -> Result<()> { + let controller: &MemController = self.controller()?; + + let mem_limit = linux_memory.limit().unwrap_or(0); + let memswap_limit = linux_memory.swap().unwrap_or(0); + + self.set_mem_and_memswap_v1(mem_limit, memswap_limit)?; + + if let Some(reservation) = linux_memory.reservation() { + controller.set_soft_limit(reservation)?; + } + + if linux_memory.disable_oom_killer().unwrap_or_default() { + controller.disable_oom_killer()?; + } + + if let Some(swappiness) = linux_memory.swappiness() { + if swappiness <= 100 { + controller.set_swappiness(swappiness)?; + } else { + return Err(Error::InvalidLinuxResource); + }; + } + + Ok(()) + } + + fn set_memory_v2(&self, linux_memory: &LinuxMemory) -> Result<()> { + let controller: &MemController = self.controller()?; + + if linux_memory.reservation().is_none() + && linux_memory.limit().is_none() + && linux_memory.swap().is_none() + { + return Ok(()); + } + + let mem_limit = linux_memory.limit().unwrap_or(0); + let memswap_limit = linux_memory.swap().unwrap_or(0); + + // Check memory usage + if mem_limit <= 0 && memswap_limit <= 0 { + return Ok(()); + } + + let memory_stat = controller.memory_stat(); + let usage_actual = memory_stat.usage_in_bytes; + + // Rejecting: memory+swap limit <= usage + if memswap_limit > 0 && memswap_limit as u64 <= usage_actual { + return Err(Error::InvalidLinuxResource); + } + + // Rejecting: memory limit <= usage + if mem_limit > 0 && mem_limit as u64 <= usage_actual { + return Err(Error::InvalidLinuxResource); + } + + let swap_limit = conv::memory_swap_to_cgroup_v2(memswap_limit, mem_limit)?; + controller.set_memswap_limit(swap_limit)?; + + if mem_limit != 0 { + controller.set_limit(mem_limit)?; + } + + if let Some(reservation) = linux_memory.reservation() { + controller.set_soft_limit(reservation)?; + } + + Ok(()) + } + + /// Set memory resources. + /// + /// Ignore kernel memory and kernel memory TCP, as runc does, see [1]. + /// + /// 1: https://github.com/opencontainers/cgroups/blob/d36d371fe756a30d2e21d83c6b42e86af77bf4a2/fs/memory.go#L36 + fn set_memory(&self, linux_memory: &LinuxMemory) -> Result<()> { + if self.v2() { + self.set_memory_v2(linux_memory)?; + } else { + self.set_memory_v1(linux_memory)?; + } + + Ok(()) + } + + fn set_pids(&self, pids: &LinuxPids) -> Result<()> { + let controller: &PidController = self.controller()?; + let value = if pids.limit() > 0 { + MaxValue::Value(pids.limit()) + } else { + MaxValue::Max + }; + controller.set_pid_max(value)?; + + Ok(()) + } + + fn set_blkio(&self, blkio: &LinuxBlockIo) -> Result<()> { + let controller: &BlkIoController = self.controller()?; + + if let Some(weight) = blkio.weight() { + controller.set_weight(weight as u64)?; + } + + if let Some(leaf_weight) = blkio.leaf_weight() { + controller.set_leaf_weight(leaf_weight as u64)?; + } + + if let Some(devices) = blkio.weight_device() { + for device in devices.iter() { + let major = device.major() as u64; + let minor = device.minor() as u64; + if let Some(weight) = device.weight() { + controller.set_weight_for_device(major, minor, weight as u64)?; + } + if let Some(leaf_weight) = device.leaf_weight() { + controller.set_leaf_weight_for_device(major, minor, leaf_weight as u64)?; + } + } + } + + if let Some(devices) = blkio.throttle_read_bps_device() { + for device in devices.iter() { + let major = device.major() as u64; + let minor = device.minor() as u64; + let rate = device.rate(); + controller.throttle_read_bps_for_device(major, minor, rate)?; + } + } + + if let Some(devices) = blkio.throttle_write_bps_device() { + for device in devices.iter() { + let major = device.major() as u64; + let minor = device.minor() as u64; + let rate = device.rate(); + controller.throttle_write_bps_for_device(major, minor, rate)?; + } + } + + if let Some(devices) = blkio.throttle_read_iops_device() { + for device in devices.iter() { + let major = device.major() as u64; + let minor = device.minor() as u64; + let rate = device.rate(); + controller.throttle_read_iops_for_device(major, minor, rate)?; + } + } + + if let Some(devices) = blkio.throttle_write_iops_device() { + for device in devices.iter() { + let major = device.major() as u64; + let minor = device.minor() as u64; + let rate = device.rate(); + controller.throttle_write_iops_for_device(major, minor, rate)?; + } + } + + Ok(()) + } + + fn set_hugepages(&self, hugepage_limits: &[LinuxHugepageLimit]) -> Result<()> { + let controller: &HugeTlbController = self.controller()?; + + for limit in hugepage_limits.iter() { + // ignore not supported page size + if !controller.size_supported(limit.page_size()) { + continue; + } + let page_size = limit.page_size(); + let limit = limit.limit() as u64; + controller.set_limit_in_bytes(page_size, limit)?; + } + + Ok(()) + } + + fn set_network(&self, network: &LinuxNetwork) -> Result<()> { + if let Some(class_id) = network.class_id() { + let controller: &NetClsController = self.controller()?; + controller.set_class(class_id as u64)?; + } + + if let Some(priorities) = network.priorities() { + let controller: &NetPrioController = self.controller()?; + for priority in priorities.iter() { + let eif = priority.name(); + let prio = priority.priority() as u64; + controller.set_if_prio(eif, prio)?; + } + } + + Ok(()) + } + + fn set_devices(&self, devices: &[LinuxDeviceCgroup]) -> Result<()> { + let controller: &DevicesController = self.controller()?; + + for device in devices.iter() { + let devtype = + DeviceType::from_char(device.typ().unwrap_or_default().as_str().chars().next()) + .ok_or(Error::InvalidLinuxResource)?; + + let perm = device + .access() + .as_ref() + .unwrap_or(&String::new()) + .chars() + .filter_map(|perm| match perm { + 'r' => Some(DevicePermissions::Read), + 'w' => Some(DevicePermissions::Write), + 'm' => Some(DevicePermissions::MkNod), + _ => None, + }) + .collect::>(); + + let major = device.major().unwrap_or(0); + let minor = device.minor().unwrap_or(0); + + if device.allow() { + controller.allow_device(devtype, major, minor, &perm)?; + } else { + controller.deny_device(devtype, major, minor, &perm)?; + } + } + + Ok(()) + } + + /// Set the controller topdown from root in cgroup hierarchy. The `f` + /// is going to be applied to: + /// -> root [not included] + /// -> root's child + /// -> ... + /// -> self.cgroup's parent + /// -> self.cgroup [not included] + /// + /// Please see `enable_cpus_topdown()` for more details. + /// + /// Please note that `self.cgroup` is not included. If you really want + /// that, you should do it manually. + fn set_controller_topdown(&self, f: F) -> Result<()> + where + for<'a> &'a T: From<&'a Subsystem>, + T: Controller + ControllIdentifier, + for<'a> F: Fn(&'a T) -> Result<()>, + { + let root = hierarchies::auto().root_control_group(); + let controller: &T = root + .controller_of() + .ok_or(FsError::new(FsErrorKind::SubsystemsEmpty))?; + let root_path = Path::new(controller.path()); + let root_path_str = root_path.to_string_lossy().to_string(); + + let controller: &T = self.controller()?; + let path = Path::new(controller.path()); + + // Push path's ancestors onto a stack, so the stack looks like: + // path's parent, path's grandparent, ..., root + let mut path_stack = vec![]; + for parent in path.ancestors() { + if parent == root_path { + break; + } + path_stack.push(parent); + } + + // Pop from the stack + while let Some(p) = path_stack.pop() { + let relative_path = p + .to_str() + .unwrap() + .trim_start_matches(&root_path_str) + // Makes sure the starting slash is removed + .trim_start_matches("/"); + let cgroup = Cgroup::new(hierarchies::auto(), relative_path)?; + let controller: &T = cgroup + .controller_of() + .ok_or(FsError::new(FsErrorKind::SubsystemsEmpty))?; + f(controller)?; + } + + Ok(()) + } + + fn cpu_acct_stats(&self) -> Result { + let controller: &CpuAcctController = self.controller()?; + let cpu_acct = controller.cpuacct(); + + let user_usage = parse_value_from_tuples(&cpu_acct.stat, "user").unwrap_or_default(); + + let system_usage = parse_value_from_tuples(&cpu_acct.stat, "system").unwrap_or_default(); + + let usage_percpu: Vec = cpu_acct + .usage_percpu + .lines() + .filter_map(|line| line.parse::().ok()) + .collect(); + + Ok(CpuAcctStats { + user_usage, + system_usage, + total_usage: cpu_acct.usage, + usage_percpu, + }) + } + + fn cpu_throttling_stats(&self) -> Result { + let controller: &CpuController = self.controller()?; + let stats = controller.cpu().stat; + + let periods = parse_value_from_tuples(&stats, "nr_periods").unwrap_or_default(); + let throttled_periods = parse_value_from_tuples(&stats, "nr_throttled").unwrap_or_default(); + let throttled_time = parse_value_from_tuples(&stats, "throttled_time").unwrap_or_default(); + + Ok(CpuThrottlingStats { + periods, + throttled_periods, + throttled_time, + }) + } + + fn cpu_cgroup_stats(&self) -> CpuCgroupStats { + CpuCgroupStats { + cpu_acct: self.cpu_acct_stats().ok(), + cpu_throttling: self.cpu_throttling_stats().ok(), + } + } + + fn memory_stats(&self) -> Result { + let controller: &MemController = self.controller()?; + let memory_stats = controller.memory_stat(); + + Ok(MemoryStats { + usage: memory_stats.usage_in_bytes, + max_usage: memory_stats.max_usage_in_bytes, + limit: memory_stats.limit_in_bytes, + fail_cnt: memory_stats.fail_cnt, + }) + } + + fn memory_swap_stats(&self) -> Result { + let controller: &MemController = self.controller()?; + let memory_swap_stats = controller.memswap(); + + Ok(MemoryStats { + usage: memory_swap_stats.usage_in_bytes, + max_usage: memory_swap_stats.max_usage_in_bytes, + limit: memory_swap_stats.limit_in_bytes, + fail_cnt: memory_swap_stats.fail_cnt, + }) + } + + fn kernel_memory_stats(&self) -> Result { + let controller: &MemController = self.controller()?; + let kmem_stats = controller.kmem_stat(); + + Ok(MemoryStats { + usage: kmem_stats.usage_in_bytes, + max_usage: kmem_stats.max_usage_in_bytes, + limit: kmem_stats.limit_in_bytes, + fail_cnt: kmem_stats.fail_cnt, + }) + } + + fn memory_cgroup_stats(&self) -> MemoryCgroupStats { + let memory = self.memory_stats().ok(); + let memory_swap = self.memory_swap_stats().ok(); + let kernel_memory = self.kernel_memory_stats().ok(); + + let mut memory = MemoryCgroupStats { + memory, + memory_swap, + kernel_memory, + ..Default::default() + }; + + let memory_stats = self + .controller::() + .map(|c| c.memory_stat()) + .ok(); + if let Some(memstats) = &memory_stats { + memory.use_hierarchy = memstats.use_hierarchy == 1; + // Copy items from memstats.stat + memory.cache = memstats.stat.cache; + memory.rss = memstats.stat.rss; + memory.rss_huge = memstats.stat.rss_huge; + memory.shmem = memstats.stat.shmem; + memory.mapped_file = memstats.stat.mapped_file; + memory.dirty = memstats.stat.dirty; + memory.writeback = memstats.stat.writeback; + memory.swap = memstats.stat.swap; + memory.pgpgin = memstats.stat.pgpgin; + memory.pgpgout = memstats.stat.pgpgout; + memory.pgfault = memstats.stat.pgfault; + memory.pgmajfault = memstats.stat.pgmajfault; + memory.inactive_anon = memstats.stat.inactive_anon; + memory.active_anon = memstats.stat.active_anon; + memory.inactive_file = memstats.stat.inactive_file; + memory.active_file = memstats.stat.active_file; + memory.unevictable = memstats.stat.unevictable; + memory.hierarchical_memory_limit = memstats.stat.hierarchical_memory_limit; + memory.hierarchical_memsw_limit = memstats.stat.hierarchical_memsw_limit; + memory.total_cache = memstats.stat.total_cache; + memory.total_rss = memstats.stat.total_rss; + memory.total_rss_huge = memstats.stat.total_rss_huge; + memory.total_shmem = memstats.stat.total_shmem; + memory.total_mapped_file = memstats.stat.total_mapped_file; + memory.total_dirty = memstats.stat.total_dirty; + memory.total_writeback = memstats.stat.total_writeback; + memory.total_swap = memstats.stat.total_swap; + memory.total_pgpgin = memstats.stat.total_pgpgin; + memory.total_pgpgout = memstats.stat.total_pgpgout; + memory.total_pgfault = memstats.stat.total_pgfault; + memory.total_pgmajfault = memstats.stat.total_pgmajfault; + memory.total_inactive_anon = memstats.stat.total_inactive_anon; + memory.total_active_anon = memstats.stat.total_active_anon; + memory.total_inactive_file = memstats.stat.total_inactive_file; + memory.total_active_file = memstats.stat.total_active_file; + memory.total_unevictable = memstats.stat.total_unevictable; + } + + memory + } + + fn pids_cgroup_stats(&self) -> PidsCgroupStats { + let controller: &PidController = match self.controller() { + Ok(controller) => controller, + Err(_) => return PidsCgroupStats::default(), + }; + let current = controller.get_pid_current().unwrap_or_default(); + let limit = controller + .get_pid_max() + .map(|mv| match mv { + MaxValue::Value(limit) => limit, + MaxValue::Max => 0, + }) + .unwrap_or_default(); + + PidsCgroupStats { current, limit } + } + + fn blkio_stats_v1(&self) -> Result { + let controller: &BlkIoController = self.controller()?; + let blkio = controller.blkio(); + + if blkio.io_serviced_recursive.is_empty() { + Ok(BlkioCgroupStats { + io_service_bytes_recursive: BlkioStat::from_io_services( + &blkio.throttle.io_service_bytes, + ), + io_serviced_recursive: BlkioStat::from_io_services(&blkio.throttle.io_serviced), + ..Default::default() + }) + } else { + Ok(BlkioCgroupStats { + io_service_bytes_recursive: BlkioStat::from_io_services( + &blkio.io_service_bytes_recursive, + ), + io_serviced_recursive: BlkioStat::from_io_services(&blkio.io_serviced_recursive), + io_queued_recursive: BlkioStat::from_io_services(&blkio.io_queued_recursive), + io_service_time_recursive: BlkioStat::from_io_services( + &blkio.io_service_time_recursive, + ), + io_wait_time_recursive: BlkioStat::from_io_services(&blkio.io_wait_time_recursive), + io_merged_recursive: BlkioStat::from_io_services(&blkio.io_merged_recursive), + io_time_recursive: BlkioStat::from_blk_io_data(&blkio.time_recursive), + sectors_recursive: BlkioStat::from_blk_io_data(&blkio.sectors_recursive), + }) + } + } + + fn blkio_stats_v2(&self) -> Result { + let controller: &BlkIoController = self.controller()?; + let blkio = controller.blkio(); + + Ok(BlkioCgroupStats { + io_service_bytes_recursive: BlkioStat::from_io_stats(&blkio.io_stat), + ..Default::default() + }) + } + + fn blkio_cgroup_stats(&self) -> BlkioCgroupStats { + if self.v2() { + self.blkio_stats_v2() + } else { + self.blkio_stats_v1() + } + .unwrap_or_default() + } + + fn huge_tlb_cgroup_stats(&self) -> HugeTlbCgroupStats { + let controller: &HugeTlbController = match self.controller() { + Ok(controller) => controller, + Err(_) => return HugeTlbCgroupStats::default(), + }; + + let sizes = controller.get_sizes(); + sizes + .iter() + .map(|s| { + let usage = controller.usage_in_bytes(s).unwrap_or_default(); + let max_usage = controller.max_usage_in_bytes(s).unwrap_or_default(); + let fail_cnt = controller.failcnt(s).unwrap_or_default(); + + let stat = HugeTlbStat { + usage, + max_usage, + fail_cnt, + }; + + (s.to_string(), stat) + }) + .collect() + } +} + +impl Manager for FsManager { + fn add_proc(&mut self, tgid: CgroupPid) -> Result<()> { + self.create_cgroups()?; + self.cgroup.add_task_by_tgid(tgid)?; + Ok(()) + } + + fn add_thread(&mut self, pid: CgroupPid) -> Result<()> { + self.create_cgroups()?; + + self.cgroup.add_task(pid).or_else(|err| { + // Try to add_proc with the pid when threaded cgroup is + // disabled in cgroup v2. + if err.kind() == &FsErrorKind::CgroupMode && self.v2() { + self.add_proc(pid) + } else { + Err(Error::Cgroupfs(err)) + } + }) + } + + fn pids(&self) -> Result> { + Ok(self + .controller::() + .map_err(Error::Cgroupfs)? + .tasks()) + } + + fn freeze(&self, state: FreezerState) -> Result<()> { + let controller: &FreezerController = self.controller()?; + + match state { + FreezerState::Thawed => controller.thaw()?, + FreezerState::Frozen => controller.freeze()?, + FreezerState::Freezing => return Err(Error::InvalidArgument), + } + + Ok(()) + } + + fn destroy(&mut self) -> Result<()> { + if !self.exists() { + return Ok(()); + } + + // Before deleting the cgroup, we should move processes in the + // cgroup to the root cgroup. Otherwise, we'll have a "Device or + // resource busy" error. + if self.v2() { + for tgid in self.cgroup.procs() { + // Ignore all errors as long as the cgroup is deleted. + let _ = self.cgroup.remove_task_by_tgid(tgid); + } + } else { + for pid in self.cgroup.tasks() { + // Ditto. + let _ = self.cgroup.remove_task(pid); + } + } + + self.cgroup.delete()?; + Ok(()) + } + + fn set(&mut self, resources: &LinuxResources) -> Result<()> { + if let Some(cpu) = resources.cpu() { + self.set_cpuset(cpu)?; + self.set_cpu(cpu)?; + } + + if let Some(memory) = resources.memory() { + self.set_memory(memory)?; + } + + if let Some(pid) = resources.pids() { + self.set_pids(pid)?; + } + + if let Some(blkio) = resources.block_io() { + self.set_blkio(blkio)?; + } + + if let Some(hugepage_limits) = resources.hugepage_limits() { + self.set_hugepages(hugepage_limits)?; + } + + if let Some(network) = resources.network() { + self.set_network(network)?; + } + + if let Some(devices) = resources.devices() { + self.set_devices(devices)?; + } + + Ok(()) + } + + fn cgroup_path(&self, subsystem: Option<&str>) -> Result { + if self.v2() { + return Ok(join_path(UNIFIED_MOUNTPOINT, &self.base)); + } + + let subsystem = subsystem + .ok_or_else(|| FsError::new(FsErrorKind::InvalidPath)) + .map_err(Error::Cgroupfs)?; + let path = self + .paths + .get(subsystem) + .ok_or(FsError::new(FsErrorKind::SubsystemsEmpty)) + .map_err(Error::Cgroupfs)?; + + Ok(path.clone()) + } + + fn enable_cpus_topdown(&self, cpus: &str) -> Result<()> { + if cpus.is_empty() { + return Ok(()); + } + + self.set_controller_topdown(|c: &CpuSetController| { + c.set_cpus(cpus).map_err(Error::Cgroupfs) + })?; + + Ok(()) + } + + fn stats(&self) -> CgroupStats { + CgroupStats { + cpu: self.cpu_cgroup_stats(), + memory: self.memory_cgroup_stats(), + pids: self.pids_cgroup_stats(), + blkio: self.blkio_cgroup_stats(), + hugetlb: self.huge_tlb_cgroup_stats(), + } + } + + fn paths(&self) -> &HashMap { + &self.paths + } + + fn mounts(&self) -> &HashMap { + &self.mounts + } + + fn systemd(&self) -> bool { + false + } + + fn v2(&self) -> bool { + self.cgroup.v2() + } +} + +/// Parse cgroup subsystem paths from `/proc/self/cgroup`. +fn parse_cgroup_subsystems() -> Result> { + let mut cgroup_paths = HashMap::new(); + let data = fs::read_to_string(CGROUP_PATH) + .map_err(|err| FsError::with_cause(FsErrorKind::FsError, err)) + .map_err(Error::Cgroupfs)?; + + // Expected line format: `10:memory:/user.slice` + for line in data.lines() { + let parts: Vec<&str> = line.split(':').collect(); + if parts.len() != 3 { + // Ignore corrupt lines + continue; + } + let subsystems = parts[1].split(','); + let path = parts[2]; + subsystems.for_each(|subsystem| { + cgroup_paths.insert(subsystem.to_string(), path.to_string()); + }); + } + + Ok(cgroup_paths) +} + +/// Parse cgroup mount information from `/proc/self/mountinfo`. +fn parse_cgroup_mountinfo(paths: &HashMap) -> Result> { + let mut mounts = HashMap::new(); + let data = fs::read_to_string(MOUNTINFO_PATH) + .map_err(|err| FsError::with_cause(FsErrorKind::FsError, err)) + .map_err(Error::Cgroupfs)?; + + for line in data.lines() { + let parts: Vec<&str> = line.splitn(2, " - ").collect(); + let part1: Vec<&str> = parts[0].split(' ').collect(); + let part2: Vec<&str> = parts[1].split(' ').collect(); + + if part2.len() != 3 { + continue; + } + + let fs_type = part2[0]; + if fs_type != "cgroup" && fs_type != "cgroup2" { + continue; + } + + let super_opts: Vec<&str> = part2[2].split(',').collect(); + for opt in super_opts.iter() { + // If opt matchs the one of cgroup subsystems + if paths.contains_key(*opt) { + let mountpoint = part1[4]; + mounts.insert(opt.to_string(), mountpoint.to_string()); + } + } + } + + Ok(mounts) +} + +pub(crate) fn join_path(base: &str, path: &str) -> String { + let base = Path::new(base); + base.join(path).to_string_lossy().to_string() +} + +/// Parse the value of an item from a tuple string split by whitespace. +/// +/// For example, we have a tuple string like: +/// +/// let tuple_str: &str = "system 100000\nuser 200000"; +/// +/// assert_eq!( +/// parse_value_from_tuples::(tuple_str, "user"), +/// Some(200000), +/// ); +/// assert_eq!( +/// parse_value_from_tuples::(tuple_str, "user1"), +/// None, +/// ); +fn parse_value_from_tuples(tuple_str: &str, item: &str) -> Option +where + T: FromStr, +{ + tuple_str.lines().find_map(|line| { + let mut parts = line.split_whitespace(); + let current_item = parts.next()?; + let value = parts.next()?; + if current_item != item { + return None; + } + value.parse::().ok() + }) +} + +impl BlkioStat { + fn from_io_services(io_services: &[IoService]) -> Vec { + let mut stats = Vec::new(); + + for service in io_services.iter() { + let major = service.major as u64; + let minor = service.minor as u64; + + stats.push(BlkioStat { + major, + minor, + op: "read".to_string(), + value: service.read, + }); + + stats.push(BlkioStat { + major, + minor, + op: "write".to_string(), + value: service.write, + }); + + stats.push(BlkioStat { + major, + minor, + op: "sync".to_string(), + value: service.sync, + }); + + stats.push(BlkioStat { + major, + minor, + op: "async".to_string(), + value: service.r#async, + }); + + stats.push(BlkioStat { + major, + minor, + op: "total".to_string(), + value: service.total, + }); + } + + stats + } + + fn from_io_stats(io_stats: &[IoStat]) -> Vec { + let mut stats = Vec::new(); + + for stat in io_stats.iter() { + let major = stat.major as u64; + let minor = stat.minor as u64; + + stats.push(BlkioStat { + major, + minor, + op: "read".to_string(), + value: stat.rbytes, + }); + + stats.push(BlkioStat { + major, + minor, + op: "write".to_string(), + value: stat.wbytes, + }); + + stats.push(BlkioStat { + major, + minor, + op: "rios".to_string(), + value: stat.rios, + }); + + stats.push(BlkioStat { + major, + minor, + op: "wios".to_string(), + value: stat.wios, + }); + + stats.push(BlkioStat { + major, + minor, + op: "dbytes".to_string(), + value: stat.dbytes, + }); + + stats.push(BlkioStat { + major, + minor, + op: "dios".to_string(), + value: stat.dios, + }); + } + + stats + } + + fn from_blk_io_data(blkiodata: &[BlkIoData]) -> Vec { + let op = String::new(); + + blkiodata + .iter() + .map(|item| BlkioStat { + major: item.major as u64, + minor: item.minor as u64, + op: op.clone(), + value: item.data, + }) + .collect() + } +} diff --git a/src/manager/mod.rs b/src/manager/mod.rs new file mode 100644 index 0000000..2f5005d --- /dev/null +++ b/src/manager/mod.rs @@ -0,0 +1,77 @@ +// Copyright (c) 2025 Ant Group +// +// SPDX-License-Identifier: Apache-2.0 or MIT +// + +mod error; +use std::collections::HashMap; + +pub use error::{Error, Result}; +mod fs; +pub use fs::FsManager; +mod conv; + +use oci_spec::runtime::LinuxResources; + +use crate::{CgroupPid, CgroupStats, FreezerState}; + +/// Manage cgroups designed for OCI containers. +pub trait Manager: Send + Sync { + /// Add a process specified by its tgid. + fn add_proc(&mut self, tgid: CgroupPid) -> Result<()>; + + /// Add a thread specified by its pid. + fn add_thread(&mut self, pid: CgroupPid) -> Result<()>; + + /// Get the list of pids joint to the cgroups. + fn pids(&self) -> Result>; + + /// Set the freezer cgroup to the specified state. + fn freeze(&self, state: FreezerState) -> Result<()>; + + /// Remove the cgroups. + fn destroy(&mut self) -> Result<()>; + + /// Set the resources to the cgroups. + fn set(&mut self, resources: &LinuxResources) -> Result<()>; + + /// Get the cgroup path. + /// + /// # Arguments + /// + /// - `subsystem`: cgroup subsystem, for cgroup v1 the value should not + /// be empty, while for cgroup v2 the only valid value is `None`. + fn cgroup_path(&self, subsystem: Option<&str>) -> Result; + + /// Enable CPUs, topdown from root in cgroup hierarchy, this would be + /// useful for CPU hotplug in the guest. + /// + /// The caller should update cgroup resources manually, in particular + /// cpuset, after this, in order to use the new CPUs (or avoid using + /// offline CPUs). + /// + /// # Arguments + /// + /// - `cpus`: online CPUs in the same format with `cat + /// /sys/devices/system/cpu/online`, e.g. "0-3,6-7". + fn enable_cpus_topdown(&self, cpus: &str) -> Result<()>; + + /// Get cgroup stats. + fn stats(&self) -> CgroupStats; + + /// Get the mappings of subsystems to their relative path. The full + /// path would be something like "{mountpoint}/{relative_path}". The + /// mappings of mountpoints see "mounts()". + fn paths(&self) -> &HashMap; + + /// Get the mappings of subsystems to their mountpoints. The full + /// path would be something like "{mountpoint}/{relative_path}". The + /// mappings of relative paths see "paths()". + fn mounts(&self) -> &HashMap; + + /// Indicate whether the cgroup manager is using systemd. + fn systemd(&self) -> bool; + + /// Indicate whether the cgroup manager is using cgroup v2. + fn v2(&self) -> bool; +} diff --git a/src/stats.rs b/src/stats.rs new file mode 100644 index 0000000..856cc25 --- /dev/null +++ b/src/stats.rs @@ -0,0 +1,156 @@ +// Copyright (c) 2018 Levente Kurusa +// Copyright (c) 2025 Ant Group +// +// SPDX-License-Identifier: Apache-2.0 or MIT +// + +use std::collections::HashMap; + +#[derive(Debug, Default)] +pub struct CgroupStats { + pub cpu: CpuCgroupStats, + pub memory: MemoryCgroupStats, + pub pids: PidsCgroupStats, + pub blkio: BlkioCgroupStats, + pub hugetlb: HugeTlbCgroupStats, +} + +#[derive(Debug, Default)] +pub struct CpuCgroupStats { + pub cpu_acct: Option, + pub cpu_throttling: Option, +} + +#[derive(Debug, Default)] +pub struct CpuAcctStats { + /// Usage in userspace, read from `cpuacct.stat` from the line starting + /// with `user`. Set 0 if no data. + pub user_usage: u64, + /// Usage in kernelspace, read from `cpuacct.stat` from the line + /// starting with `system`. Set 0 if no data. + pub system_usage: u64, + /// Total usage, read from `cpuacct.usage`. Set 0 if no data. + pub total_usage: u64, + /// Per-CPU usage, read from `cpuacct.usage_percpu`. + pub usage_percpu: Vec, +} + +#[derive(Debug, Default)] +pub struct CpuThrottlingStats { + /// Periods, read from `cpu.stat` from the line starting with + /// `nr_periods`. Set 0 if no data. + pub periods: u64, + /// Throttled periods, read from `cpu.stat` from the line starting with + /// `nr_throttled`. Set 0 if no data. + pub throttled_periods: u64, + /// Throttled time, read from `cpu.stat` from the line starting with + /// `throttled_time`. Set 0 if no data. + pub throttled_time: u64, +} + +#[derive(Debug, Default)] +pub struct MemoryCgroupStats { + pub memory: Option, + pub memory_swap: Option, + pub kernel_memory: Option, + + /// Use hierarchy, read from `memory.use_hierarchy` in cgroups v1. Only + /// available in cgroups v1. + pub use_hierarchy: bool, + + // The following data is read from `memory.stat`, see also + // `crate::fs::memory::MemoryStat::stat`. + pub cache: u64, + pub rss: u64, + pub rss_huge: u64, + pub shmem: u64, + pub mapped_file: u64, + pub dirty: u64, + pub writeback: u64, + pub swap: u64, + pub pgpgin: u64, + pub pgpgout: u64, + pub pgfault: u64, + pub pgmajfault: u64, + pub inactive_anon: u64, + pub active_anon: u64, + pub inactive_file: u64, + pub active_file: u64, + pub unevictable: u64, + pub hierarchical_memory_limit: i64, + pub hierarchical_memsw_limit: i64, + pub total_cache: u64, + pub total_rss: u64, + pub total_rss_huge: u64, + pub total_shmem: u64, + pub total_mapped_file: u64, + pub total_dirty: u64, + pub total_writeback: u64, + pub total_swap: u64, + pub total_pgpgin: u64, + pub total_pgpgout: u64, + pub total_pgfault: u64, + pub total_pgmajfault: u64, + pub total_inactive_anon: u64, + pub total_active_anon: u64, + pub total_inactive_file: u64, + pub total_active_file: u64, + pub total_unevictable: u64, +} + +#[derive(Debug, Default)] +pub struct MemoryStats { + /// Memory [swap] usage, read from `memory[.memsw].usage_in_bytes` in + /// cgroups v1 and `memory[.swap].current` in cgroups v2. + pub usage: u64, + /// Maximum memory [swap] usage observed by cgroups, read from + /// `memory[.memsw].max_usage_in_bytes` in cgroups v1 and + /// `memory[.swap].peak` in cgroups v2. + pub max_usage: u64, + /// Memory [swap] limit, read from `memory[.memsw].limit_in_bytes` in + /// cgroups v1 and `memory[.swap].max` in cgroups v2. + pub limit: i64, + /// Failure count, read from `memory[.memsw].failcnt`. Only available in + /// cgroups v1. + pub fail_cnt: u64, +} + +#[derive(Debug, Default)] +pub struct PidsCgroupStats { + /// Current number of processes in the cgroup, read from `pids.current`. + pub current: u64, + /// Maximum number of processes in the cgroup, read from `pids.limit`. + pub limit: i64, +} + +#[derive(Debug, Default)] +pub struct BlkioCgroupStats { + pub io_service_bytes_recursive: Vec, + pub io_serviced_recursive: Vec, + pub io_queued_recursive: Vec, + pub io_service_time_recursive: Vec, + pub io_wait_time_recursive: Vec, + pub io_merged_recursive: Vec, + pub io_time_recursive: Vec, + pub sectors_recursive: Vec, +} + +#[derive(Debug, Default)] +pub struct BlkioStat { + pub major: u64, + pub minor: u64, + pub op: String, + pub value: u64, +} + +/// A structure representing the statistics of the `hugetlb` subsystem of a +/// Cgroup. The key is the huge page size, and the value is the statistics +/// for that size. +pub type HugeTlbCgroupStats = HashMap; + +#[derive(Debug, Default)] +pub struct HugeTlbStat { + pub usage: u64, + pub max_usage: u64, + pub fail_cnt: u64, +}