This is an automated email from the ASF dual-hosted git repository. mmodzelewski pushed a commit to branch connectors-scoped-stats in repository https://gitbox.apache.org/repos/asf/iggy.git
commit 26a3d921c6ac59ae9e8ade9cd3e788bf689dcbf0 Author: Maciej Modzelewski <[email protected]> AuthorDate: Fri Jul 10 19:34:33 2026 +0200 improvements --- core/connectors/runtime/src/main.rs | 2 + core/connectors/runtime/src/stats.rs | 14 +++--- core/server-ng/src/main.rs | 4 ++ core/server-ng/src/responses.rs | 8 ++-- core/server/src/main.rs | 4 ++ core/server/src/shard/system/stats.rs | 18 +++----- core/system_stats/src/lib.rs | 83 ++++++++++++++++++++++++++++++----- 7 files changed, 96 insertions(+), 37 deletions(-) diff --git a/core/connectors/runtime/src/main.rs b/core/connectors/runtime/src/main.rs index dfd149c4d..5c5ebe777 100644 --- a/core/connectors/runtime/src/main.rs +++ b/core/connectors/runtime/src/main.rs @@ -39,6 +39,7 @@ use std::{ env, sync::{Arc, atomic::AtomicU32}, }; +use system_stats::capture_allowed_cpus; use tracing::{error, info}; mod api; @@ -114,6 +115,7 @@ fn print_ascii_art(text: &str) { #[tokio::main] async fn main() -> Result<(), RuntimeError> { + capture_allowed_cpus(); Args::parse(); print_ascii_art("Iggy Connectors"); diff --git a/core/connectors/runtime/src/stats.rs b/core/connectors/runtime/src/stats.rs index e64f455d7..455bea6c4 100644 --- a/core/connectors/runtime/src/stats.rs +++ b/core/connectors/runtime/src/stats.rs @@ -124,19 +124,17 @@ struct SystemProbe { available_memory: u64, } -/// CPU and memory scoped to what this process may actually use. On a -/// shared host the runtime runs cpuset- or cgroup-confined; the host-wide -/// `sysinfo` numbers would report the neighbors' CPU load and a memory -/// total the runtime can never allocate, and they leak host sizing to -/// whoever can read `/stats`. fn probe_system(pid: u32) -> SystemProbe { + let pid = Pid::from_u32(pid); + // `new()` not `new_all()`: the refreshes below cover everything read + // here; `new_all()` would keep the full host process table alive forever. let mut system = SYSINFO - .get_or_init(|| Mutex::new(System::new_all())) + .get_or_init(|| Mutex::new(System::new())) .lock() .unwrap_or_else(PoisonError::into_inner); system.refresh_cpu_all(); system.refresh_memory(); - system.refresh_processes(ProcessesToUpdate::Some(&[Pid::from_u32(pid)]), true); + system.refresh_processes(ProcessesToUpdate::Some(&[pid]), true); let mut probe = SystemProbe { cpu_usage: 0.0, @@ -146,7 +144,7 @@ fn probe_system(pid: u32) -> SystemProbe { available_memory: system.available_memory(), }; - if let Some(process) = system.process(Pid::from_u32(pid)) { + if let Some(process) = system.process(pid) { probe.cpu_usage = process.cpu_usage(); probe.memory_usage = process.memory(); if let Some(memory) = cgroup_scoped_memory(&system, process) { diff --git a/core/server-ng/src/main.rs b/core/server-ng/src/main.rs index 4aabb4df4..a3e11fd64 100644 --- a/core/server-ng/src/main.rs +++ b/core/server-ng/src/main.rs @@ -23,9 +23,13 @@ use args::Args; use clap::Parser; use server_ng::bootstrap::{bootstrap, load_config}; use server_ng::server_error::ServerNgError; +use system_stats::capture_allowed_cpus; use tracing::{error, info}; fn main() -> Result<(), ServerNgError> { + // Before shard threads pin themselves: a pinned capture sees one core. + capture_allowed_cpus(); + let bootstrap_runtime = match server_common::create_shard_executor() { Ok(rt) => rt, Err(e) => { diff --git a/core/server-ng/src/responses.rs b/core/server-ng/src/responses.rs index c2e02279e..8a80338d4 100644 --- a/core/server-ng/src/responses.rs +++ b/core/server-ng/src/responses.rs @@ -675,9 +675,7 @@ fn build_stats_response(shard: &Rc<ServerNgShard>) -> Result<StatsResponse, Iggy /// These describe the whole process, not shard or metadata state, so any one /// shard can serve them without aggregation. The CPU fields are deltas over the /// serving thread's own [`SYSINFO`] refresh history, so they vary by serving -/// shard (a shard's first probe reports zero CPU). The `total_*` fields are -/// scoped to the process's allowed CPU set and cgroup memory cap on confined -/// hosts (see `system_stats`), host-wide otherwise. +/// shard (a shard's first probe reports zero CPU). struct SystemStats { process_id: u32, cpu_usage: f32, @@ -733,7 +731,9 @@ fn probe_system_stats() -> SystemStats { let process_id = std::process::id(); let host = HOST_IDENTITY.get_or_init(HostIdentity::probe); SYSINFO.with_borrow_mut(|slot| { - let sys = slot.get_or_insert_with(SysinfoSystem::new_all); + // `new()` not `new_all()`: the refreshes below cover everything read + // here; `new_all()` would keep the full host process table alive forever. + let sys = slot.get_or_insert_with(SysinfoSystem::new); sys.refresh_cpu_all(); sys.refresh_memory(); sys.refresh_processes(ProcessesToUpdate::Some(&[Pid::from_u32(process_id)]), true); diff --git a/core/server/src/main.rs b/core/server/src/main.rs index 2f598c75c..b5c0110f7 100644 --- a/core/server/src/main.rs +++ b/core/server/src/main.rs @@ -57,6 +57,7 @@ use std::sync::Arc; use std::sync::atomic::{AtomicBool, AtomicU32, AtomicU64, Ordering}; use std::sync::mpsc; use std::thread::JoinHandle; +use system_stats::capture_allowed_cpus; use tracing::{error, info, instrument, warn}; const COMPONENT: &str = "MAIN"; @@ -116,6 +117,9 @@ fn print_ascii_art(text: &str) { #[instrument(skip_all, name = "trace_start_server")] fn main() -> Result<(), ServerError> { + // Before shard threads pin themselves: a pinned capture sees one core. + capture_allowed_cpus(); + let rt = match compio::runtime::Runtime::new() { Ok(rt) => rt, Err(e) => { diff --git a/core/server/src/shard/system/stats.rs b/core/server/src/shard/system/stats.rs index 054c77ec5..559729757 100644 --- a/core/server/src/shard/system/stats.rs +++ b/core/server/src/shard/system/stats.rs @@ -32,14 +32,10 @@ impl IggyShard { SYSINFO.with(|sysinfo_cell| { let mut sysinfo_opt = sysinfo_cell.borrow_mut(); - - if sysinfo_opt.is_none() { - let mut sys = SysinfoSystem::new_all(); - sys.refresh_all(); - *sysinfo_opt = Some(sys); - } - - let sys = sysinfo_opt.as_mut().unwrap(); + // `new()` not `new_all()`: the refreshes below cover everything + // read here; `new_all()` would keep the full host process table + // alive forever. + let sys = sysinfo_opt.get_or_insert_with(SysinfoSystem::new); let process_id = std::process::id(); sys.refresh_cpu_all(); sys.refresh_memory(); @@ -71,11 +67,7 @@ impl IggyShard { ..Default::default() }; - if let Some(process) = sys - .processes() - .values() - .find(|p| p.pid() == Pid::from_u32(process_id)) - { + if let Some(process) = sys.process(Pid::from_u32(process_id)) { stats.process_id = process.pid().as_u32(); stats.cpu_usage = process.cpu_usage(); stats.memory_usage = process.memory().into(); diff --git a/core/system_stats/src/lib.rs b/core/system_stats/src/lib.rs index d0c89d7e9..4bda91915 100644 --- a/core/system_stats/src/lib.rs +++ b/core/system_stats/src/lib.rs @@ -27,6 +27,7 @@ //! unconfined. use cpu_allocation::allowed_cpus; +use std::sync::OnceLock; use sysinfo::{Process, System}; mod cgroup_memory; @@ -40,19 +41,40 @@ pub struct CgroupMemory { pub available: u64, } +static ALLOWED_CPUS: OnceLock<Vec<usize>> = OnceLock::new(); + +/// Snapshot the process's allowed CPU set for [`scoped_total_cpu_usage`]. +/// +/// Call once from the main thread at startup, before any thread pins +/// itself to a core: `sched_getaffinity` reports the calling thread's +/// mask, so a capture from a pinned shard thread would record that single +/// core as the whole process's set. +pub fn capture_allowed_cpus() { + ALLOWED_CPUS.get_or_init(allowed_cpus); +} + /// Average CPU usage over the cores this process may run on, or the /// host-global average when the process is unrestricted. /// /// `global_cpu_usage` averages every host core, so a cpuset-confined -/// process would report its neighbors' load. +/// process would report its neighbors' load. The allowed set is the +/// [`capture_allowed_cpus`] boot snapshot; without one this stays on the +/// host-global average. pub fn scoped_total_cpu_usage(sys: &System) -> f32 { - allowed_cores_cpu_usage(sys).unwrap_or_else(|| sys.global_cpu_usage()) + ALLOWED_CPUS + .get() + .and_then(|allowed| allowed_cores_cpu_usage(sys, allowed)) + .unwrap_or_else(|| sys.global_cpu_usage()) } /// Memory totals scoped to the process's memory cgroup. `None` when no /// ancestor caps memory below the host total; callers keep the host /// numbers then. /// +/// `process` must be the calling process: the `available` refinement +/// walks `/proc/self/cgroup`, so a foreign process would silently get +/// self's reclaim numbers. +/// /// `available` adds reclaimable file cache back rather than taking the /// kernel's `limit - current`, which trends toward zero on a cache-heavy /// workload long before real OOM pressure. @@ -69,18 +91,17 @@ pub fn cgroup_scoped_memory(sys: &System, process: &Process) -> Option<CgroupMem }) } -/// `None` when the process is unrestricted (or an allowed core is missing -/// from `sys.cpus()`), where the host-global number is already the right -/// one. -fn allowed_cores_cpu_usage(sys: &System) -> Option<f32> { +/// `None` when the allowed set is not a strict subset of the host's cores +/// (or an allowed core is missing from `sys.cpus()`), where the +/// host-global number is already the right one. +fn allowed_cores_cpu_usage(sys: &System, allowed: &[usize]) -> Option<f32> { let cpus = sys.cpus(); - let allowed = allowed_cpus(); if allowed.is_empty() || allowed.len() >= cpus.len() { return None; } let mut total_usage = 0.0f32; - for cpu_id in &allowed { + for cpu_id in allowed { let name = format!("cpu{cpu_id}"); total_usage += cpus.iter().find(|cpu| cpu.name() == name)?.cpu_usage(); } @@ -100,17 +121,55 @@ mod tests { assert_eq!(scoped_total_cpu_usage(&sys), sys.global_cpu_usage()); } + // Linux-only: the by-name lookup relies on the kernel's cpu0..cpuN + // naming, which is also the only platform where confinement exists. + #[cfg(target_os = "linux")] + #[test] + fn given_allowed_core_subset_when_probing_scoped_cpu_should_average_those_cores() { + let mut sys = System::new(); + sys.refresh_cpu_all(); + std::thread::sleep(sysinfo::MINIMUM_CPU_UPDATE_INTERVAL); + sys.refresh_cpu_all(); + + let cpus = sys.cpus(); + if cpus.len() < 2 { + return; + } + + let allowed: Vec<usize> = (0..cpus.len() - 1).collect(); + let expected = allowed + .iter() + .map(|cpu_id| { + cpus.iter() + .find(|cpu| cpu.name() == format!("cpu{cpu_id}")) + .expect("linux names cores cpu0..cpuN") + .cpu_usage() + }) + .sum::<f32>() + / allowed.len() as f32; + + assert_eq!(allowed_cores_cpu_usage(&sys, &allowed), Some(expected)); + } + #[test] - fn given_own_process_when_probing_cgroup_memory_should_stay_within_host_total() { + fn given_own_process_when_probing_cgroup_memory_should_reflect_confinement() { let mut sys = System::new(); sys.refresh_memory(); let pid = Pid::from_u32(std::process::id()); sys.refresh_processes(ProcessesToUpdate::Some(&[pid]), true); let process = sys.process(pid).expect("own process must be visible"); - if let Some(memory) = cgroup_scoped_memory(&sys, process) { - assert!(memory.total < sys.total_memory()); - assert!(memory.available <= memory.total); + match cgroup_scoped_memory(&sys, process) { + Some(memory) => { + assert!(memory.total < sys.total_memory()); + assert!(memory.available <= memory.total); + } + None => assert!( + process + .cgroup_limits() + .is_none_or(|limits| limits.total_memory >= sys.total_memory()), + "None must mean no ancestor caps memory below the host total" + ), } } }
