You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@iggy.apache.org by ma...@apache.org on 2026/07/10 14:20:19 UTC
(iggy) branch connectors-scoped-stats created (now fbd018fb9)
This is an automated email from the ASF dual-hosted git repository.
mmodzelewski pushed a change to branch connectors-scoped-stats
in repository https://gitbox.apache.org/repos/asf/iggy.git
at fbd018fb9 feat: respect CPU and cgroup limits in server-ng and connectors stats
This branch includes the following new commits:
new fbd018fb9 feat: respect CPU and cgroup limits in server-ng and connectors stats
The 1 revisions listed above as "new" are entirely new to this
repository and will be described in separate emails. The revisions
listed as "add" were already present in the repository and have only
been added to this reference.
(iggy) 01/01: feat: respect CPU and cgroup limits in server-ng and connectors stats
Posted by ma...@apache.org.
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 fbd018fb9e0bfd8d399118af727787e0689b47da
Author: Maciej Modzelewski <ma...@modzelewski.dev>
AuthorDate: Fri Jul 10 16:09:02 2026 +0200
feat: respect CPU and cgroup limits in server-ng and connectors stats
#3615 scoped the legacy server's stats to the process's allowed
CPU set and cgroup memory cap, but server-ng and the connectors
runtime kept reporting host-wide sysinfo numbers: a confined
instance showed its neighbors' CPU load and a memory total it
can never allocate, leaking host sizing to anyone who can read
the stats endpoint.
Extract the scoping into a shared system_stats crate
(scoped_total_cpu_usage, cgroup_scoped_memory) and use it from
the legacy server, server-ng and the connectors runtime.
The connectors runtime also built a fresh sysinfo System on
every call, so its CPU fields (deltas over the previous refresh)
were always zero. Keep one process-wide System so readings are
real from the second probe on.
---
Cargo.lock | 13 ++-
Cargo.toml | 2 +
core/connectors/runtime/Cargo.toml | 1 +
core/connectors/runtime/src/stats.rs | 81 +++++++++-----
core/server-ng/Cargo.toml | 1 +
core/server-ng/src/responses.rs | 12 ++-
core/server/Cargo.toml | 2 +-
core/server/src/shard/system/mod.rs | 1 -
core/server/src/shard/system/stats.rs | 40 +------
core/system_stats/Cargo.toml | 31 ++++++
.../system => system_stats/src}/cgroup_memory.rs | 0
core/system_stats/src/lib.rs | 116 +++++++++++++++++++++
12 files changed, 236 insertions(+), 64 deletions(-)
diff --git a/Cargo.lock b/Cargo.lock
index e5ed364e9..c75598082 100644
--- a/Cargo.lock
+++ b/Cargo.lock
@@ -6751,6 +6751,7 @@ dependencies = [
"serde_yaml_ng",
"strum 0.28.0",
"sysinfo 0.39.5",
+ "system_stats",
"tempfile",
"thiserror 2.0.18",
"tokio",
@@ -11698,7 +11699,6 @@ dependencies = [
"clap",
"compio",
"configs",
- "cpu_allocation",
"ctrlc",
"cyper",
"cyper-axum",
@@ -11737,6 +11737,7 @@ dependencies = [
"socket2 0.6.4",
"strum 0.28.0",
"sysinfo 0.39.5",
+ "system_stats",
"tempfile",
"thiserror 2.0.18",
"tokio",
@@ -11817,6 +11818,7 @@ dependencies = [
"socket2 0.6.4",
"strum 0.28.0",
"sysinfo 0.39.5",
+ "system_stats",
"tempfile",
"thiserror 2.0.18",
"tokio",
@@ -12759,6 +12761,15 @@ dependencies = [
"libc",
]
+[[package]]
+name = "system_stats"
+version = "0.1.0"
+dependencies = [
+ "cpu_allocation",
+ "sysinfo 0.39.5",
+ "tempfile",
+]
+
[[package]]
name = "tagptr"
version = "0.2.0"
diff --git a/Cargo.toml b/Cargo.toml
index 5990f44e7..0c31e253f 100644
--- a/Cargo.toml
+++ b/Cargo.toml
@@ -64,6 +64,7 @@ members = [
"core/shard",
"core/shard_allocator",
"core/simulator",
+ "core/system_stats",
"core/tools",
"examples/rust",
]
@@ -305,6 +306,7 @@ strum = { version = "0.28.0", features = ["derive"] }
strum_macros = "0.28.0"
syn = { version = "2", features = ["full", "extra-traits"] }
sysinfo = "0.39.5"
+system_stats = { path = "core/system_stats" }
tempfile = "3.27.0"
terminal_size = { version = "0.4.4" }
test-case = "3.3.1"
diff --git a/core/connectors/runtime/Cargo.toml b/core/connectors/runtime/Cargo.toml
index e193c5325..648300999 100644
--- a/core/connectors/runtime/Cargo.toml
+++ b/core/connectors/runtime/Cargo.toml
@@ -70,6 +70,7 @@ serde_with = { workspace = true }
serde_yaml_ng = { workspace = true }
strum = { workspace = true }
sysinfo = { workspace = true }
+system_stats = { workspace = true }
thiserror = { workspace = true }
tokio = { workspace = true }
toml = { workspace = true }
diff --git a/core/connectors/runtime/src/stats.rs b/core/connectors/runtime/src/stats.rs
index ed3f9053b..f6a586e38 100644
--- a/core/connectors/runtime/src/stats.rs
+++ b/core/connectors/runtime/src/stats.rs
@@ -20,31 +20,22 @@ use crate::metrics::ConnectorType;
use iggy_common::{IggyTimestamp, SemanticVersion};
use iggy_connector_sdk::api::{ConnectorRuntimeStats, ConnectorStats};
use std::str::FromStr;
-use std::sync::Arc;
-use sysinfo::System;
+use std::sync::{Arc, Mutex, OnceLock, PoisonError};
+use sysinfo::{Pid, ProcessesToUpdate, System};
+use system_stats::{cgroup_scoped_memory, scoped_total_cpu_usage};
const VERSION: &str = env!("CARGO_PKG_VERSION");
const SEMANTIC_VERSION: SemanticVersion = SemanticVersion::parse_const(VERSION);
+// CPU readings are deltas over the previous refresh, so the sampled `System`
+// must outlive a single call: a freshly created one reports zero for every
+// CPU field. The first call still reports zeros; the periodic pollers behind
+// `/stats` get real deltas from the second call on.
+static SYSINFO: OnceLock<Mutex<System>> = OnceLock::new();
+
pub async fn get_runtime_stats(context: &Arc<RuntimeContext>) -> ConnectorRuntimeStats {
let pid = std::process::id();
-
- let mut system = System::new_all();
- system.refresh_cpu_all();
- system.refresh_memory();
- system.refresh_processes(
- sysinfo::ProcessesToUpdate::Some(&[sysinfo::Pid::from_u32(pid)]),
- true,
- );
-
- let total_cpu_usage = system.global_cpu_usage();
- let total_memory = system.total_memory();
- let available_memory = system.available_memory();
-
- let (cpu_usage, memory_usage) = system
- .process(sysinfo::Pid::from_u32(pid))
- .map(|p| (p.cpu_usage(), p.memory()))
- .unwrap_or((0.0, 0));
+ let system = probe_system(pid);
let sources = context.sources.get_all().await;
let sinks = context.sinks.get_all().await;
@@ -114,11 +105,11 @@ pub async fn get_runtime_stats(context: &Arc<RuntimeContext>) -> ConnectorRuntim
connectors_runtime_version: VERSION.to_owned(),
connectors_runtime_version_semver: SEMANTIC_VERSION.get_numeric_version().ok(),
process_id: pid,
- cpu_usage,
- total_cpu_usage,
- memory_usage,
- total_memory,
- available_memory,
+ cpu_usage: system.cpu_usage,
+ total_cpu_usage: system.total_cpu_usage,
+ memory_usage: system.memory_usage,
+ total_memory: system.total_memory,
+ available_memory: system.available_memory,
run_time,
start_time: start,
sources_total,
@@ -128,3 +119,45 @@ pub async fn get_runtime_stats(context: &Arc<RuntimeContext>) -> ConnectorRuntim
connectors,
}
}
+
+struct SystemProbe {
+ cpu_usage: f32,
+ total_cpu_usage: f32,
+ memory_usage: u64,
+ total_memory: u64,
+ 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 mut system = SYSINFO
+ .get_or_init(|| Mutex::new(System::new_all()))
+ .lock()
+ .unwrap_or_else(PoisonError::into_inner);
+ system.refresh_cpu_all();
+ system.refresh_memory();
+ system.refresh_processes(ProcessesToUpdate::Some(&[Pid::from_u32(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::from_u32(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
+}
diff --git a/core/server-ng/Cargo.toml b/core/server-ng/Cargo.toml
index 26f397d00..91523299b 100644
--- a/core/server-ng/Cargo.toml
+++ b/core/server-ng/Cargo.toml
@@ -154,6 +154,7 @@ slab = { workspace = true }
socket2 = { workspace = true }
strum = { workspace = true }
sysinfo = { workspace = true }
+system_stats = { workspace = true }
tempfile = { workspace = true }
thiserror = { workspace = true }
tokio = { workspace = true }
diff --git a/core/server-ng/src/responses.rs b/core/server-ng/src/responses.rs
index 1fd86e0a7..c2e02279e 100644
--- a/core/server-ng/src/responses.rs
+++ b/core/server-ng/src/responses.rs
@@ -88,6 +88,7 @@ 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};
/// Build the `get_me` reply for the requesting connection. Identity
/// (`user_id`, transport kind, peer address) comes from the per-shard
@@ -674,7 +675,9 @@ 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).
+/// 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.
struct SystemStats {
process_id: u32,
cpu_usage: f32,
@@ -738,7 +741,7 @@ fn probe_system_stats() -> SystemStats {
let mut stats = SystemStats {
process_id,
cpu_usage: 0.0,
- total_cpu_usage: sys.global_cpu_usage(),
+ total_cpu_usage: scoped_total_cpu_usage(sys),
memory_usage: 0,
total_memory: sys.total_memory(),
available_memory: sys.available_memory(),
@@ -767,6 +770,11 @@ fn probe_system_stats() -> SystemStats {
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;
+ }
}
stats
diff --git a/core/server/Cargo.toml b/core/server/Cargo.toml
index 037bd30b9..f40538611 100644
--- a/core/server/Cargo.toml
+++ b/core/server/Cargo.toml
@@ -48,7 +48,6 @@ chrono = { workspace = true }
clap = { workspace = true }
compio = { workspace = true }
configs = { workspace = true }
-cpu_allocation = { workspace = true }
ctrlc = { workspace = true }
cyper = { workspace = true }
cyper-axum = { workspace = true }
@@ -87,6 +86,7 @@ slab = { workspace = true }
socket2 = { workspace = true }
strum = { workspace = true }
sysinfo = { workspace = true }
+system_stats = { workspace = true }
tempfile = { workspace = true }
thiserror = { workspace = true }
tokio = { workspace = true }
diff --git a/core/server/src/shard/system/mod.rs b/core/server/src/shard/system/mod.rs
index 907d2c8ab..f326ea08e 100644
--- a/core/server/src/shard/system/mod.rs
+++ b/core/server/src/shard/system/mod.rs
@@ -15,7 +15,6 @@
// specific language governing permissions and limitations
// under the License.
-mod cgroup_memory;
pub mod clients;
pub mod cluster;
pub mod consumer_groups;
diff --git a/core/server/src/shard/system/stats.rs b/core/server/src/shard/system/stats.rs
index 6da51fce0..054c77ec5 100644
--- a/core/server/src/shard/system/stats.rs
+++ b/core/server/src/shard/system/stats.rs
@@ -16,12 +16,11 @@
// under the License.
use crate::shard::IggyShard;
-use crate::shard::system::cgroup_memory::cgroup_available_memory;
use crate::{SEMANTIC_VERSION, VERSION};
-use cpu_allocation::allowed_cpus;
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};
thread_local! {
static SYSINFO: RefCell<Option<SysinfoSystem>> = const { RefCell::new(None) };
@@ -46,8 +45,7 @@ impl IggyShard {
sys.refresh_memory();
sys.refresh_processes(ProcessesToUpdate::Some(&[Pid::from_u32(process_id)]), true);
- let total_cpu_usage =
- allowed_cores_cpu_usage(sys).unwrap_or_else(|| sys.global_cpu_usage());
+ 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;
@@ -92,15 +90,9 @@ impl IggyShard {
stats.threads_count = process.tasks().map(|t| t.len() as u32).unwrap_or(0);
- if let Some(limits) = process
- .cgroup_limits()
- .filter(|limits| limits.total_memory < sys.total_memory())
- {
- stats.total_memory = limits.total_memory.into();
- stats.available_memory = cgroup_available_memory(sys.total_memory())
- .unwrap_or(limits.free_memory)
- .min(limits.total_memory)
- .into();
+ if let Some(memory) = cgroup_scoped_memory(sys, process) {
+ stats.total_memory = memory.total.into();
+ stats.available_memory = memory.available.into();
}
}
@@ -156,25 +148,3 @@ impl IggyShard {
})
}
}
-
-/// Average usage over the cores this process may run on.
-///
-/// `global_cpu_usage` averages every host core, so a cpuset-confined
-/// instance would report its neighbors' load. `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: &SysinfoSystem) -> 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 {
- let name = format!("cpu{cpu_id}");
- total_usage += cpus.iter().find(|cpu| cpu.name() == name)?.cpu_usage();
- }
-
- Some(total_usage / allowed.len() as f32)
-}
diff --git a/core/system_stats/Cargo.toml b/core/system_stats/Cargo.toml
new file mode 100644
index 000000000..c3f596057
--- /dev/null
+++ b/core/system_stats/Cargo.toml
@@ -0,0 +1,31 @@
+# Licensed to the Apache Software Foundation (ASF) under one
+# or more contributor license agreements. See the NOTICE file
+# distributed with this work for additional information
+# regarding copyright ownership. The ASF licenses this file
+# to you under the Apache License, Version 2.0 (the
+# "License"); you may not use this file except in compliance
+# with the License. You may obtain a copy of the License at
+#
+# http://www.apache.org/licenses/LICENSE-2.0
+#
+# Unless required by applicable law or agreed to in writing,
+# software distributed under the License is distributed on an
+# "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+# KIND, either express or implied. See the License for the
+# specific language governing permissions and limitations
+# under the License.
+
+[package]
+name = "system_stats"
+version = "0.1.0"
+description = "System stats scoped to the resources a process may actually use: CPU usage over the allowed core set and cgroup-capped memory totals, shared by the stats endpoints of the servers and the connectors runtime."
+edition = "2024"
+license = "Apache-2.0"
+publish = false
+
+[dependencies]
+cpu_allocation = { workspace = true }
+sysinfo = { workspace = true }
+
+[dev-dependencies]
+tempfile = { workspace = true }
diff --git a/core/server/src/shard/system/cgroup_memory.rs b/core/system_stats/src/cgroup_memory.rs
similarity index 100%
rename from core/server/src/shard/system/cgroup_memory.rs
rename to core/system_stats/src/cgroup_memory.rs
diff --git a/core/system_stats/src/lib.rs b/core/system_stats/src/lib.rs
new file mode 100644
index 000000000..d0c89d7e9
--- /dev/null
+++ b/core/system_stats/src/lib.rs
@@ -0,0 +1,116 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Stats scoped to the resources a process may actually use.
+//!
+//! `sysinfo`'s host-wide numbers describe the whole machine, so a process
+//! 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.
+
+use cpu_allocation::allowed_cpus;
+use sysinfo::{Process, 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,
+}
+
+/// 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.
+pub fn scoped_total_cpu_usage(sys: &System) -> f32 {
+ allowed_cores_cpu_usage(sys).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.
+///
+/// `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> {
+ let limits = process
+ .cgroup_limits()
+ .filter(|limits| limits.total_memory < sys.total_memory())?;
+ let available = cgroup_available_memory(sys.total_memory())
+ .unwrap_or(limits.free_memory)
+ .min(limits.total_memory);
+ Some(CgroupMemory {
+ total: limits.total_memory,
+ available,
+ })
+}
+
+/// `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> {
+ 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 {
+ let name = format!("cpu{cpu_id}");
+ total_usage += cpus.iter().find(|cpu| cpu.name() == name)?.cpu_usage();
+ }
+
+ Some(total_usage / allowed.len() as f32)
+}
+
+#[cfg(test)]
+mod tests {
+ use super::*;
+ use sysinfo::{Pid, ProcessesToUpdate};
+
+ #[test]
+ fn given_unrefreshed_system_when_probing_scoped_cpu_should_fall_back_to_global() {
+ let sys = System::new();
+
+ assert_eq!(scoped_total_cpu_usage(&sys), sys.global_cpu_usage());
+ }
+
+ #[test]
+ fn given_own_process_when_probing_cgroup_memory_should_stay_within_host_total() {
+ 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);
+ }
+ }
+}