This is an automated email from the ASF dual-hosted git repository.
mmodzelewski pushed a commit to branch master
in repository https://gitbox.apache.org/repos/asf/iggy.git
The following commit(s) were added to refs/heads/master by this push:
new 2ec7d8964 feat: respect CPU and cgroup limits in server-ng and
connectors stats (#3647)
2ec7d8964 is described below
commit 2ec7d89649539416de9c5c083f07e0d78806d8cf
Author: Maciej Modzelewski <[email protected]>
AuthorDate: Sat Jul 11 07:00:40 2026 +0200
feat: respect CPU and cgroup limits in server-ng and connectors stats
(#3647)
The server-ng and connectors runtime stats endpoints reported
host-wide sysinfo numbers: a cpuset-confined or memory-capped
process showed its neighbors' CPU load and a memory total it can
never allocate, leaking host sizing on multi-tenant machines. The
legacy server already scoped these (#3615) but kept the logic
inline.
Extract that logic into a shared system_stats crate: SystemProbe
scopes total CPU usage to the allowed core set (snapshotted at
startup, before shard threads pin themselves) and memory totals
to the effective cgroup cap, falling back to host-wide values for
unconfined processes. All three binaries now probe through it.
On confined hosts the reported total_cpu_usage, total_memory and
available_memory now reflect the confinement, so dashboards and
alerts keyed to the old host-wide values will see lower numbers.
The connectors runtime also keeps its sysinfo System alive across
captures, so CPU usage reflects real deltas instead of the zero
first sample of a freshly created System on every request.
---
Cargo.lock | 13 +-
Cargo.toml | 2 +
core/connectors/runtime/Cargo.toml | 1 +
core/connectors/runtime/src/main.rs | 2 +
core/connectors/runtime/src/stats.rs | 44 ++--
core/connectors/sdk/src/api.rs | 18 ++
core/server-ng/Cargo.toml | 1 +
core/server-ng/src/main.rs | 4 +
core/server-ng/src/responses.rs | 69 +++---
core/server/Cargo.toml | 2 +-
core/server/src/main.rs | 4 +
core/server/src/shard/system/mod.rs | 1 -
core/server/src/shard/system/stats.rs | 224 +++++++-----------
core/system_stats/Cargo.toml | 31 +++
.../system => system_stats/src}/cgroup_memory.rs | 0
core/system_stats/src/lib.rs | 259 +++++++++++++++++++++
16 files changed, 462 insertions(+), 213 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/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 ed3f9053b..1ed8e9271 100644
--- a/core/connectors/runtime/src/stats.rs
+++ b/core/connectors/runtime/src/stats.rs
@@ -20,31 +20,17 @@ 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 std::sync::{Arc, Mutex, OnceLock, PoisonError};
use sysinfo::System;
+use system_stats::SystemProbe;
const VERSION: &str = env!("CARGO_PKG_VERSION");
const SEMANTIC_VERSION: SemanticVersion =
SemanticVersion::parse_const(VERSION);
-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();
+static SYSINFO: OnceLock<Mutex<System>> = OnceLock::new();
- let (cpu_usage, memory_usage) = system
- .process(sysinfo::Pid::from_u32(pid))
- .map(|p| (p.cpu_usage(), p.memory()))
- .unwrap_or((0.0, 0));
+pub async fn get_runtime_stats(context: &Arc<RuntimeContext>) ->
ConnectorRuntimeStats {
+ let system = probe_system();
let sources = context.sources.get_all().await;
let sinks = context.sinks.get_all().await;
@@ -113,12 +99,12 @@ 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,
- cpu_usage,
- total_cpu_usage,
- memory_usage,
- total_memory,
- available_memory,
+ process_id: system.process_id,
+ 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 +114,11 @@ pub async fn get_runtime_stats(context:
&Arc<RuntimeContext>) -> ConnectorRuntim
connectors,
}
}
+
+fn probe_system() -> SystemProbe {
+ let mut system = SYSINFO
+ .get_or_init(|| Mutex::new(System::new()))
+ .lock()
+ .unwrap_or_else(PoisonError::into_inner);
+ SystemProbe::capture(&mut system)
+}
diff --git a/core/connectors/sdk/src/api.rs b/core/connectors/sdk/src/api.rs
index d690d96c6..b5edeba3f 100644
--- a/core/connectors/sdk/src/api.rs
+++ b/core/connectors/sdk/src/api.rs
@@ -73,21 +73,39 @@ impl ConnectorError {
/// Runtime statistics response from `/stats` endpoint.
#[derive(Debug, Serialize, Deserialize)]
pub struct ConnectorRuntimeStats {
+ /// The version of the connectors runtime.
pub connectors_runtime_version: String,
+ /// The semantic version of the connectors runtime in the numeric format,
+ /// e.g. 1.2.3 -> 1002003 (major followed by zero-padded three-digit minor
and patch).
#[serde(skip_serializing_if = "Option::is_none")]
pub connectors_runtime_version_semver: Option<u32>,
+ /// The unique identifier of the runtime process.
pub process_id: u32,
+ /// The CPU usage of the runtime process.
pub cpu_usage: f32,
+ /// The total CPU usage of the system, scoped to the cores this process
may run on
+ /// when confined by an affinity/cpuset mask.
pub total_cpu_usage: f32,
+ /// The memory usage of the runtime process in bytes.
pub memory_usage: u64,
+ /// The total memory of the system in bytes, or the effective cgroup
memory limit when
+ /// the runtime runs inside a memory-capped cgroup (container, systemd
slice).
pub total_memory: u64,
+ /// The available memory of the system in bytes, scoped to the cgroup
limit when one applies.
pub available_memory: u64,
+ /// The elapsed time since the runtime started, in microseconds.
pub run_time: u64,
+ /// The time the runtime started, in microseconds since the UNIX epoch.
pub start_time: u64,
+ /// The number of configured source connectors, including disabled and
failed ones.
pub sources_total: u32,
+ /// The number of currently running source connectors.
pub sources_running: u32,
+ /// The number of configured sink connectors, including disabled and
failed ones.
pub sinks_total: u32,
+ /// The number of currently running sink connectors.
pub sinks_running: u32,
+ /// Per-connector statistics for every configured source and sink.
pub connectors: Vec<ConnectorStats>,
}
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/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 1fd86e0a7..8080a8194 100644
--- a/core/server-ng/src/responses.rs
+++ b/core/server-ng/src/responses.rs
@@ -87,7 +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 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
@@ -727,50 +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| {
- let sys = slot.get_or_insert_with(SysinfoSystem::new_all);
- 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: sys.global_cpu_usage(),
- 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));
- }
+ let probe = SYSINFO.with_borrow_mut(|slot| {
+ let sys = slot.get_or_insert_with(SysinfoSystem::new);
+ 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/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/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/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..a1d0114cf 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 sysinfo::System as SysinfoSystem;
+use system_stats::SystemProbe;
thread_local! {
static SYSINFO: RefCell<Option<SysinfoSystem>> = const {
RefCell::new(None) };
@@ -31,150 +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();
-
- 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();
- 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 total_cpu_usage =
- allowed_cores_cpu_usage(sys).unwrap_or_else(||
sys.global_cpu_usage());
- 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 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()
- };
-
- if let Some(process) = sys
- .processes()
- .values()
- .find(|p| p.pid() == 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(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();
- }
- }
-
- 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 probe = SYSINFO.with_borrow_mut(|slot| {
+ let sys = slot.get_or_insert_with(SysinfoSystem::new);
+ SystemProbe::capture(sys)
+ });
+
+ 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: 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()
+ };
+
+ 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)
- });
-
- 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();
}
+ (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;
+
+ 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)
}
}
-
-/// 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..eb6fd4fa6
--- /dev/null
+++ b/core/system_stats/src/lib.rs
@@ -0,0 +1,259 @@
+// 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.
+//! [`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::{Pid, Process, ProcessesToUpdate, System};
+
+mod cgroup_memory;
+
+use cgroup_memory::cgroup_available_memory;
+
+/// 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 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
+/// 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);
+}
+
+/// 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.
+///
+/// `global_cpu_usage` averages every host core, so a cpuset-confined
+/// 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.
+fn scoped_total_cpu_usage(sys: &System) -> f32 {
+ 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.
+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 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();
+ 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_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();
+
+ 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_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");
+
+ 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"
+ ),
+ }
+ }
+}
