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 fb6d2f80c2c70d482e4d5053fa93b7b359690745 Author: Maciej Modzelewski <[email protected]> AuthorDate: Fri Jul 10 20:16:29 2026 +0200 unify probing --- core/connectors/runtime/src/stats.rs | 45 ++------- core/server-ng/src/responses.rs | 75 +++++---------- core/server/src/shard/system/stats.rs | 176 +++++++++++++++------------------- core/system_stats/src/lib.rs | 110 ++++++++++++++++++--- 4 files changed, 205 insertions(+), 201 deletions(-) diff --git a/core/connectors/runtime/src/stats.rs b/core/connectors/runtime/src/stats.rs index 455bea6c4..1ed8e9271 100644 --- a/core/connectors/runtime/src/stats.rs +++ b/core/connectors/runtime/src/stats.rs @@ -21,8 +21,8 @@ use iggy_common::{IggyTimestamp, SemanticVersion}; use iggy_connector_sdk::api::{ConnectorRuntimeStats, ConnectorStats}; use std::str::FromStr; use std::sync::{Arc, Mutex, OnceLock, PoisonError}; -use sysinfo::{Pid, ProcessesToUpdate, System}; -use system_stats::{cgroup_scoped_memory, scoped_total_cpu_usage}; +use sysinfo::System; +use system_stats::SystemProbe; const VERSION: &str = env!("CARGO_PKG_VERSION"); const SEMANTIC_VERSION: SemanticVersion = SemanticVersion::parse_const(VERSION); @@ -30,8 +30,7 @@ const SEMANTIC_VERSION: SemanticVersion = SemanticVersion::parse_const(VERSION); static SYSINFO: OnceLock<Mutex<System>> = OnceLock::new(); pub async fn get_runtime_stats(context: &Arc<RuntimeContext>) -> ConnectorRuntimeStats { - let pid = std::process::id(); - let system = probe_system(pid); + let system = probe_system(); let sources = context.sources.get_all().await; let sinks = context.sinks.get_all().await; @@ -100,7 +99,7 @@ pub async fn get_runtime_stats(context: &Arc<RuntimeContext>) -> ConnectorRuntim ConnectorRuntimeStats { connectors_runtime_version: VERSION.to_owned(), connectors_runtime_version_semver: SEMANTIC_VERSION.get_numeric_version().ok(), - process_id: pid, + process_id: system.process_id, cpu_usage: system.cpu_usage, total_cpu_usage: system.total_cpu_usage, memory_usage: system.memory_usage, @@ -116,42 +115,10 @@ pub async fn get_runtime_stats(context: &Arc<RuntimeContext>) -> ConnectorRuntim } } -struct SystemProbe { - cpu_usage: f32, - total_cpu_usage: f32, - memory_usage: u64, - total_memory: u64, - available_memory: u64, -} - -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. +fn probe_system() -> SystemProbe { let mut system = SYSINFO .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]), true); - - let mut probe = SystemProbe { - cpu_usage: 0.0, - total_cpu_usage: scoped_total_cpu_usage(&system), - memory_usage: 0, - total_memory: system.total_memory(), - available_memory: system.available_memory(), - }; - - 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) { - probe.total_memory = memory.total; - probe.available_memory = memory.available; - } - } - - probe + SystemProbe::capture(&mut system) } diff --git a/core/server-ng/src/responses.rs b/core/server-ng/src/responses.rs index 8a80338d4..8080a8194 100644 --- a/core/server-ng/src/responses.rs +++ b/core/server-ng/src/responses.rs @@ -87,8 +87,8 @@ use shard::ConnectedClientInfo; use std::cell::RefCell; use std::rc::Rc; use std::sync::{Arc, OnceLock}; -use sysinfo::{Pid, ProcessesToUpdate, System as SysinfoSystem}; -use system_stats::{cgroup_scoped_memory, scoped_total_cpu_usage}; +use sysinfo::System as SysinfoSystem; +use system_stats::SystemProbe; /// Build the `get_me` reply for the requesting connection. Identity /// (`user_id`, transport kind, peer address) comes from the per-shard @@ -728,57 +728,32 @@ impl HostIdentity { static HOST_IDENTITY: OnceLock<HostIdentity> = OnceLock::new(); 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| { - // `new()` not `new_all()`: the refreshes below cover everything read - // here; `new_all()` would keep the full host process table alive forever. + let probe = SYSINFO.with_borrow_mut(|slot| { 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); - - let mut stats = SystemStats { - process_id, - cpu_usage: 0.0, - total_cpu_usage: scoped_total_cpu_usage(sys), - memory_usage: 0, - total_memory: sys.total_memory(), - available_memory: sys.available_memory(), - run_time: 0, - start_time: 0, - read_bytes: 0, - written_bytes: 0, - threads_count: 0, - hostname: host.hostname.clone(), - os_name: host.os_name.clone(), - os_version: host.os_version.clone(), - kernel_version: host.kernel_version.clone(), - }; - - if let Some(process) = sys.process(Pid::from_u32(process_id)) { - stats.cpu_usage = process.cpu_usage(); - stats.memory_usage = process.memory(); - // sysinfo reports whole seconds; the wire fields are micros (the - // SDK decodes them via `IggyDuration` / `IggyTimestamp::from`, both - // micro-based). - stats.run_time = process.run_time().saturating_mul(1_000_000); - stats.start_time = process.start_time().saturating_mul(1_000_000); - let disk_usage = process.disk_usage(); - stats.read_bytes = disk_usage.total_read_bytes; - stats.written_bytes = disk_usage.total_written_bytes; - stats.threads_count = process - .tasks() - .map_or(0, |tasks| u32::try_from(tasks.len()).unwrap_or(u32::MAX)); - - if let Some(memory) = cgroup_scoped_memory(sys, process) { - stats.total_memory = memory.total; - stats.available_memory = memory.available; - } - } + SystemProbe::capture(sys) + }); - stats - }) + SystemStats { + process_id: probe.process_id, + cpu_usage: probe.cpu_usage, + total_cpu_usage: probe.total_cpu_usage, + memory_usage: probe.memory_usage, + total_memory: probe.total_memory, + available_memory: probe.available_memory, + // sysinfo reports whole seconds; the wire fields are micros (the + // SDK decodes them via `IggyDuration` / `IggyTimestamp::from`, both + // micro-based). + run_time: probe.run_time_secs.saturating_mul(1_000_000), + start_time: probe.start_time_secs.saturating_mul(1_000_000), + read_bytes: probe.read_bytes, + written_bytes: probe.written_bytes, + threads_count: probe.threads_count, + hostname: host.hostname.clone(), + os_name: host.os_name.clone(), + os_version: host.os_version.clone(), + kernel_version: host.kernel_version.clone(), + } } fn build_get_stream_response( diff --git a/core/server/src/shard/system/stats.rs b/core/server/src/shard/system/stats.rs index 559729757..a1d0114cf 100644 --- a/core/server/src/shard/system/stats.rs +++ b/core/server/src/shard/system/stats.rs @@ -19,8 +19,8 @@ use crate::shard::IggyShard; use crate::{SEMANTIC_VERSION, VERSION}; use iggy_common::{IggyDuration, IggyError, Stats}; use std::cell::RefCell; -use sysinfo::{Pid, ProcessesToUpdate, System as SysinfoSystem}; -use system_stats::{cgroup_scoped_memory, scoped_total_cpu_usage}; +use sysinfo::System as SysinfoSystem; +use system_stats::SystemProbe; thread_local! { static SYSINFO: RefCell<Option<SysinfoSystem>> = const { RefCell::new(None) }; @@ -30,113 +30,91 @@ impl IggyShard { pub async fn get_stats(&self) -> Result<Stats, IggyError> { assert_eq!(self.id, 0, "GetStats should only be called on shard0"); - SYSINFO.with(|sysinfo_cell| { - let mut sysinfo_opt = sysinfo_cell.borrow_mut(); - // `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(); - sys.refresh_processes(ProcessesToUpdate::Some(&[Pid::from_u32(process_id)]), true); + let probe = SYSINFO.with_borrow_mut(|slot| { + let sys = slot.get_or_insert_with(SysinfoSystem::new); + SystemProbe::capture(sys) + }); - let total_cpu_usage = scoped_total_cpu_usage(sys); - let total_memory = sys.total_memory().into(); - let available_memory = sys.available_memory().into(); - let clients_count = self.client_manager.get_clients().len() as u32; - let hostname = sysinfo::System::host_name().unwrap_or("unknown_hostname".to_string()); - let os_name = sysinfo::System::name().unwrap_or("unknown_os_name".to_string()); - let os_version = - sysinfo::System::long_os_version().unwrap_or("unknown_os_version".to_string()); - let kernel_version = - sysinfo::System::kernel_version().unwrap_or("unknown_kernel_version".to_string()); + let clients_count = self.client_manager.get_clients().len() as u32; + let hostname = sysinfo::System::host_name().unwrap_or("unknown_hostname".to_string()); + let os_name = sysinfo::System::name().unwrap_or("unknown_os_name".to_string()); + let os_version = + sysinfo::System::long_os_version().unwrap_or("unknown_os_version".to_string()); + let kernel_version = + sysinfo::System::kernel_version().unwrap_or("unknown_kernel_version".to_string()); - let mut stats = Stats { - process_id, - total_cpu_usage, - total_memory, - available_memory, - clients_count, - hostname, - os_name, - os_version, - kernel_version, - iggy_server_version: VERSION.to_owned(), - iggy_server_semver: SEMANTIC_VERSION.get_numeric_version().ok(), - ..Default::default() - }; + let mut stats = Stats { + process_id: probe.process_id, + cpu_usage: probe.cpu_usage, + total_cpu_usage: probe.total_cpu_usage, + memory_usage: probe.memory_usage.into(), + total_memory: probe.total_memory.into(), + available_memory: probe.available_memory.into(), + run_time: IggyDuration::new_from_secs(probe.run_time_secs), + start_time: IggyDuration::new_from_secs(probe.start_time_secs) + .as_micros() + .into(), + read_bytes: probe.read_bytes.into(), + written_bytes: probe.written_bytes.into(), + threads_count: probe.threads_count, + clients_count, + hostname, + os_name, + os_version, + kernel_version, + iggy_server_version: VERSION.to_owned(), + iggy_server_semver: SEMANTIC_VERSION.get_numeric_version().ok(), + ..Default::default() + }; - 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(); - stats.run_time = IggyDuration::new_from_secs(process.run_time()); - stats.start_time = IggyDuration::new_from_secs(process.start_time()) - .as_micros() - .into(); - - let disk_usage = process.disk_usage(); - stats.read_bytes = disk_usage.total_read_bytes.into(); - stats.written_bytes = disk_usage.total_written_bytes.into(); - - stats.threads_count = process.tasks().map(|t| t.len() as u32).unwrap_or(0); - - if let Some(memory) = cgroup_scoped_memory(sys, process) { - stats.total_memory = memory.total.into(); - stats.available_memory = memory.available.into(); - } - } - - let (streams_count, topics_count, partitions_count, consumer_groups_count, stream_ids) = - self.metadata.with_metadata(|m| { - let mut topics = 0u32; - let mut partitions = 0u32; - let mut cg = 0u32; - let ids: Vec<_> = m.streams.iter().map(|(k, _)| k).collect(); - for (_, stream) in m.streams.iter() { - topics += stream.topics.len() as u32; - for (_, topic) in stream.topics.iter() { - partitions += topic.partitions.len() as u32; - cg += topic.consumer_groups.len() as u32; - } + let (streams_count, topics_count, partitions_count, consumer_groups_count, stream_ids) = + self.metadata.with_metadata(|m| { + let mut topics = 0u32; + let mut partitions = 0u32; + let mut cg = 0u32; + let ids: Vec<_> = m.streams.iter().map(|(k, _)| k).collect(); + for (_, stream) in m.streams.iter() { + topics += stream.topics.len() as u32; + for (_, topic) in stream.topics.iter() { + partitions += topic.partitions.len() as u32; + cg += topic.consumer_groups.len() as u32; } - (m.streams.len() as u32, topics, partitions, cg, ids) - }); + } + (m.streams.len() as u32, topics, partitions, cg, ids) + }); - stats.streams_count = streams_count; - stats.topics_count = topics_count; - stats.partitions_count = partitions_count; - stats.consumer_groups_count = consumer_groups_count; + stats.streams_count = streams_count; + stats.topics_count = topics_count; + stats.partitions_count = partitions_count; + stats.consumer_groups_count = consumer_groups_count; - for stream_id in stream_ids { - if let Some(stream_stat) = self.metadata.get_stream_stats(stream_id) { - stats.messages_count += stream_stat.messages_count_inconsistent(); - stats.segments_count += stream_stat.segments_count_inconsistent(); - stats.messages_size_bytes += stream_stat.size_bytes_inconsistent().into(); - } + for stream_id in stream_ids { + if let Some(stream_stat) = self.metadata.get_stream_stats(stream_id) { + stats.messages_count += stream_stat.messages_count_inconsistent(); + stats.segments_count += stream_stat.segments_count_inconsistent(); + stats.messages_size_bytes += stream_stat.size_bytes_inconsistent().into(); } + } - match fs2::available_space(&self.config.system.path) { - Ok(space) => stats.free_disk_space = space.into(), - Err(err) => { - tracing::warn!( - "Failed to get available disk space for '{}': {err}", - self.config.system.path - ); - } + match fs2::available_space(&self.config.system.path) { + Ok(space) => stats.free_disk_space = space.into(), + Err(err) => { + tracing::warn!( + "Failed to get available disk space for '{}': {err}", + self.config.system.path + ); } - match fs2::total_space(&self.config.system.path) { - Ok(space) => stats.total_disk_space = space.into(), - Err(err) => { - tracing::warn!( - "Failed to get total disk space for '{}': {err}", - self.config.system.path - ); - } + } + match fs2::total_space(&self.config.system.path) { + Ok(space) => stats.total_disk_space = space.into(), + Err(err) => { + tracing::warn!( + "Failed to get total disk space for '{}': {err}", + self.config.system.path + ); } + } - Ok(stats) - }) + Ok(stats) } } diff --git a/core/system_stats/src/lib.rs b/core/system_stats/src/lib.rs index 4bda91915..eb6fd4fa6 100644 --- a/core/system_stats/src/lib.rs +++ b/core/system_stats/src/lib.rs @@ -21,29 +21,94 @@ //! confined by a cpuset or a memory-capped cgroup (systemd `AllowedCPUs=` / //! `MemoryMax=`, container limits) reports its neighbors' CPU load and a //! memory total it can never allocate. On a multi-tenant host that leaks -//! host sizing to anyone who can read a stats endpoint. These probes scope -//! the numbers to the process's allowed CPU set and effective cgroup memory -//! cap, and fall back to the host-wide values when the process is -//! unconfined. +//! host sizing to anyone who can read a stats endpoint. +//! [`SystemProbe::capture`] scopes the numbers to the process's allowed CPU +//! set and effective cgroup memory cap, and falls back to the host-wide +//! values when the process is unconfined. use cpu_allocation::allowed_cpus; use std::sync::OnceLock; -use sysinfo::{Process, System}; +use sysinfo::{Pid, Process, ProcessesToUpdate, System}; mod cgroup_memory; use cgroup_memory::cgroup_available_memory; -/// Memory totals scoped to the process's effective cgroup cap. -#[derive(Debug, Clone, Copy, PartialEq, Eq)] -pub struct CgroupMemory { - pub total: u64, - pub available: u64, +/// One sample of the calling process's resource usage plus system totals +/// scoped to what the process may actually use. +/// +/// The CPU fields are deltas over the passed `System`'s refresh history, +/// so the first capture on a fresh `System` reports zero. `run_time_secs` +/// and `start_time_secs` are whole seconds (sysinfo's granularity); +/// callers convert to their wire units. +#[derive(Debug, Clone, Copy, PartialEq)] +pub struct SystemProbe { + pub process_id: u32, + pub cpu_usage: f32, + pub total_cpu_usage: f32, + pub memory_usage: u64, + pub total_memory: u64, + pub available_memory: u64, + pub run_time_secs: u64, + pub start_time_secs: u64, + pub read_bytes: u64, + pub written_bytes: u64, + pub threads_count: u32, +} + +impl SystemProbe { + /// Refresh `sys` and sample the calling process. + /// + /// Refreshes everything it reads, so `System::new()` is enough; + /// `System::new_all()` would keep the full host process table alive + /// for no benefit. Keep `sys` alive across captures: the CPU numbers + /// are deltas since its previous refresh. + pub fn capture(sys: &mut System) -> Self { + let process_id = std::process::id(); + let pid = Pid::from_u32(process_id); + sys.refresh_cpu_all(); + sys.refresh_memory(); + sys.refresh_processes(ProcessesToUpdate::Some(&[pid]), true); + + let mut probe = Self { + process_id, + cpu_usage: 0.0, + total_cpu_usage: scoped_total_cpu_usage(sys), + memory_usage: 0, + total_memory: sys.total_memory(), + available_memory: sys.available_memory(), + run_time_secs: 0, + start_time_secs: 0, + read_bytes: 0, + written_bytes: 0, + threads_count: 0, + }; + + if let Some(process) = sys.process(pid) { + probe.cpu_usage = process.cpu_usage(); + probe.memory_usage = process.memory(); + probe.run_time_secs = process.run_time(); + probe.start_time_secs = process.start_time(); + let disk_usage = process.disk_usage(); + probe.read_bytes = disk_usage.total_read_bytes; + probe.written_bytes = disk_usage.total_written_bytes; + probe.threads_count = process + .tasks() + .map_or(0, |tasks| u32::try_from(tasks.len()).unwrap_or(u32::MAX)); + + if let Some(memory) = cgroup_scoped_memory(sys, process) { + probe.total_memory = memory.total; + probe.available_memory = memory.available; + } + } + + probe + } } static ALLOWED_CPUS: OnceLock<Vec<usize>> = OnceLock::new(); -/// Snapshot the process's allowed CPU set for [`scoped_total_cpu_usage`]. +/// Snapshot the process's allowed CPU set for the 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 @@ -53,6 +118,13 @@ pub fn capture_allowed_cpus() { ALLOWED_CPUS.get_or_init(allowed_cpus); } +/// Memory totals scoped to the process's effective cgroup cap. +#[derive(Debug, Clone, Copy, PartialEq, Eq)] +struct CgroupMemory { + total: u64, + available: u64, +} + /// Average CPU usage over the cores this process may run on, or the /// host-global average when the process is unrestricted. /// @@ -60,7 +132,7 @@ pub fn capture_allowed_cpus() { /// 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 { +fn scoped_total_cpu_usage(sys: &System) -> f32 { ALLOWED_CPUS .get() .and_then(|allowed| allowed_cores_cpu_usage(sys, allowed)) @@ -78,7 +150,7 @@ pub fn scoped_total_cpu_usage(sys: &System) -> f32 { /// `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. -pub fn cgroup_scoped_memory(sys: &System, process: &Process) -> Option<CgroupMemory> { +fn cgroup_scoped_memory(sys: &System, process: &Process) -> Option<CgroupMemory> { let limits = process .cgroup_limits() .filter(|limits| limits.total_memory < sys.total_memory())?; @@ -114,6 +186,18 @@ mod tests { use super::*; use sysinfo::{Pid, ProcessesToUpdate}; + #[test] + fn given_fresh_system_when_capturing_probe_should_sample_own_process() { + let mut sys = System::new(); + + let probe = SystemProbe::capture(&mut sys); + + assert_eq!(probe.process_id, std::process::id()); + assert!(probe.memory_usage > 0); + assert!(probe.total_memory > 0); + assert!(probe.available_memory <= probe.total_memory); + } + #[test] fn given_unrefreshed_system_when_probing_scoped_cpu_should_fall_back_to_global() { let sys = System::new();
