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.

Author: mmodzelewski

Cargo.lock

@@ -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"

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 }

core/connectors/runtime/Cargo.toml

@@ -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");
 

core/connectors/runtime/src/main.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)
+}

core/connectors/runtime/src/stats.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>,
 }
 

core/connectors/sdk/src/api.rs

@@ -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 }

core/server-ng/Cargo.toml

@@ -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) => {

core/server-ng/src/main.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(

core/server-ng/src/responses.rs

@@ -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 }