feat: add process CPU usage cores metric

Cargo.toml

@@ -220,6 +220,9 @@ vergen-gitcl = { version = "9.1.0", features = [
 zip = { version = "8.6.0", default-features = false, features = ["deflate"] }
 anyhow = "1.0"
 
+[target.'cfg(target_os = "linux")'.dependencies]
+procfs = { version = "0.18.0", default-features = false }
+
 [dev-dependencies]
 rstest = "0.26.1"
 arrow = "59.2.0"

src/handlers/http/resource_check.rs

@@ -16,7 +16,13 @@
  *
  */
 
+#[cfg(target_os = "linux")]
+use std::path::{Path, PathBuf};
+#[cfg(target_os = "linux")]
+use std::sync::Mutex;
 use std::sync::{Arc, LazyLock, atomic::AtomicBool};
+#[cfg(target_os = "linux")]
+use std::time::Instant;
 
 use actix_web::{
     body::MessageBody,
@@ -33,31 +39,230 @@ use tokio::{
 use tracing::{info, trace, warn};
 
 use crate::analytics::{SYS_INFO, refresh_sys_info};
-use crate::metrics::{record_disk_metrics, record_process_metrics_sample};
+use crate::metrics::{
+    record_disk_metrics, record_process_cpu_usage_cores, record_process_metrics_sample,
+};
 use crate::parseable::PARSEABLE;
 
 const PROCESS_METRICS_SAMPLE_INTERVAL: Duration = Duration::from_secs(5);
+#[cfg(target_os = "linux")]
+const CGROUP_V2_CPU_MAX_FILE: &str = "cpu.max";
+#[cfg(target_os = "linux")]
+const CGROUP_V2_CPU_STAT_FILE: &str = "cpu.stat";
+#[cfg(target_os = "linux")]
+const CGROUP_V1_CPU_QUOTA_FILE: &str = "cpu.cfs_quota_us";
+#[cfg(target_os = "linux")]
+const CGROUP_V1_CPU_PERIOD_FILE: &str = "cpu.cfs_period_us";
+#[cfg(target_os = "linux")]
+const CGROUP_V1_CPU_USAGE_FILE: &str = "cpuacct.usage";
 
 static SERVER_OK: LazyLock<Arc<AtomicBool>> = LazyLock::new(|| Arc::new(AtomicBool::new(true)));
 
+#[cfg(target_os = "linux")]
+#[derive(Clone, Copy)]
+struct CgroupCpuUsageSample {
+    usage_micros: u64,
+    sampled_at: Instant,
+}
+
+#[cfg(target_os = "linux")]
+static CGROUP_CPU_USAGE_SAMPLE: LazyLock<Mutex<Option<CgroupCpuUsageSample>>> =
+    LazyLock::new(|| Mutex::new(None));
+
+#[cfg(target_os = "linux")]
+fn cpu_quota_cores(quota: &str, period: &str) -> Result<Option<f64>, ()> {
+    let quota = quota.trim();
+    if quota == "max" || quota == "-1" {
+        return Ok(None);
+    }
+
+    let quota = quota.parse::<f64>().map_err(|_| ())?;
+    let period = period.trim().parse::<f64>().map_err(|_| ())?;
+    if quota <= 0.0 || period <= 0.0 {
+        return Err(());
+    }
+
+    Ok(Some(quota / period))
+}
+
+#[cfg(target_os = "linux")]
+fn cgroup_directory(pathname: &str, root: &str, mount_point: &Path) -> Option<PathBuf> {
+    let pathname = Path::new(pathname);
+    if pathname == Path::new("/") {
+        return Some(mount_point.to_path_buf());
+    }
+    Some(mount_point.join(pathname.strip_prefix(root).ok()?))
+}
+
+#[cfg(target_os = "linux")]
+fn cgroup_cpu_limit_cores() -> Result<Option<f64>, ()> {
+    let process = procfs::process::Process::myself().map_err(|_| ())?;
+    let cgroups = process.cgroups().map_err(|_| ())?.0;
+    let mounts = process.mountinfo().map_err(|_| ())?.0;
+
+    if let (Some(cgroup), Some(mount)) = (
+        cgroups.iter().find(|group| group.hierarchy == 0),
+        mounts.iter().find(|mount| mount.fs_type == "cgroup2"),
+    ) {
+        let directory =
+            cgroup_directory(&cgroup.pathname, &mount.root, &mount.mount_point).ok_or(())?;
+        let cpu_max =
+            std::fs::read_to_string(directory.join(CGROUP_V2_CPU_MAX_FILE)).map_err(|_| ())?;
+        let mut values = cpu_max.split_whitespace();
+        return cpu_quota_cores(values.next().ok_or(())?, values.next().ok_or(())?);
+    }
+
+    let cgroup = cgroups
+        .iter()
+        .find(|group| group.controllers.iter().any(|item| item == "cpu"))
+        .ok_or(())?;
+    let mount = mounts
+        .iter()
+        .find(|mount| mount.fs_type == "cgroup" && mount.super_options.contains_key("cpu"))
+        .ok_or(())?;
+    let directory = cgroup_directory(&cgroup.pathname, &mount.root, &mount.mount_point).ok_or(())?;
+    let quota =
+        std::fs::read_to_string(directory.join(CGROUP_V1_CPU_QUOTA_FILE)).map_err(|_| ())?;
+    let period =
+        std::fs::read_to_string(directory.join(CGROUP_V1_CPU_PERIOD_FILE)).map_err(|_| ())?;
+    cpu_quota_cores(&quota, &period)
+}
+
+#[cfg(target_os = "linux")]
+fn cpu_usage_micros_from_stat(cpu_stat: &str) -> Result<u64, ()> {
+    for line in cpu_stat.lines() {
+        let mut fields = line.split_whitespace();
+        if fields.next() == Some("usage_usec") {
+            return fields.next().ok_or(())?.parse::<u64>().map_err(|_| ());
+        }
+    }
+    Err(())
+}
+
+#[cfg(target_os = "linux")]
+fn cgroup_cpu_usage_micros() -> Result<u64, ()> {
+    let process = procfs::process::Process::myself().map_err(|_| ())?;
+    let cgroups = process.cgroups().map_err(|_| ())?.0;
+    let mounts = process.mountinfo().map_err(|_| ())?.0;
+
+    if let (Some(cgroup), Some(mount)) = (
+        cgroups.iter().find(|group| group.hierarchy == 0),
+        mounts.iter().find(|mount| mount.fs_type == "cgroup2"),
+    ) {
+        let directory =
+            cgroup_directory(&cgroup.pathname, &mount.root, &mount.mount_point).ok_or(())?;
+        let cpu_stat =
+            std::fs::read_to_string(directory.join(CGROUP_V2_CPU_STAT_FILE)).map_err(|_| ())?;
+        return cpu_usage_micros_from_stat(&cpu_stat);
+    }
+
+    let cgroup = cgroups
+        .iter()
+        .find(|group| group.controllers.iter().any(|item| item == "cpuacct"))
+        .ok_or(())?;
+    let mount = mounts
+        .iter()
+        .find(|mount| mount.fs_type == "cgroup" && mount.super_options.contains_key("cpuacct"))
+        .ok_or(())?;
+    let directory = cgroup_directory(&cgroup.pathname, &mount.root, &mount.mount_point).ok_or(())?;
+    let usage_nanos = std::fs::read_to_string(directory.join(CGROUP_V1_CPU_USAGE_FILE))
+        .map_err(|_| ())?
+        .trim()
+        .parse::<u64>()
+        .map_err(|_| ())?;
+    Ok(usage_nanos / 1_000)
+}
+
+#[cfg(target_os = "linux")]
+fn cpu_usage_cores_between(
+    previous_usage_micros: u64,
+    current_usage_micros: u64,
+    elapsed_micros: u128,
+) -> Option<f64> {
+    let usage_delta = current_usage_micros.checked_sub(previous_usage_micros)?;
+    (elapsed_micros > 0).then_some(usage_delta as f64 / elapsed_micros as f64)
+}
+
+#[cfg(target_os = "linux")]
+fn cgroup_cpu_usage_cores() -> Result<Option<f64>, ()> {
+    let current = CgroupCpuUsageSample {
+        usage_micros: cgroup_cpu_usage_micros()?,
+        sampled_at: Instant::now(),
+    };
+    let mut previous = CGROUP_CPU_USAGE_SAMPLE.lock().map_err(|_| ())?;
+    let usage_cores = previous.and_then(|previous| {
+        cpu_usage_cores_between(
+            previous.usage_micros,
+            current.usage_micros,
+            current
+                .sampled_at
+                .duration_since(previous.sampled_at)
+                .as_micros(),
+        )
+    });
+    *previous = Some(current);
+    Ok(usage_cores)
+}
+
+#[cfg(not(target_os = "linux"))]
+fn cgroup_cpu_limit_cores() -> Result<Option<f64>, ()> {
+    Ok(None)
+}
+
+pub fn cpu_limit_cores() -> f64 {
+    match cgroup_cpu_limit_cores() {
+        Ok(Some(limit)) => limit,
+        Ok(None) => num_cpus::get() as f64,
+        Err(()) => 0.0,
+    }
+}
+
+pub fn cpu_usage_cores(process_cpu_usage_percent: f64) -> f64 {
+    #[cfg(target_os = "linux")]
+    if matches!(cgroup_cpu_limit_cores(), Ok(Some(_))) {
+        return match cgroup_cpu_usage_cores() {
+            Ok(Some(usage_cores)) => usage_cores,
+            Ok(None) => 0.0,
+            Err(()) => process_cpu_usage_percent / 100.0,
+        };
+    }
+
+    process_cpu_usage_percent / 100.0
+}
+
 async fn sample_process_metrics() {
     refresh_sys_info();
     let process_metrics = tokio::task::spawn_blocking(|| {
-        let sys = SYS_INFO.lock().unwrap();
-        let total_mem = if let Some(cgroup) = sys.cgroup_limits() {
-            cgroup.total_memory
-        } else {
-            sys.total_memory()
+        let process_metrics = {
+            let sys = SYS_INFO.lock().unwrap();
+            let total_mem = if let Some(cgroup) = sys.cgroup_limits() {
+                cgroup.total_memory
+            } else {
+                sys.total_memory()
+            };
+            sysinfo::get_current_pid()
+                .ok()
+                .and_then(|pid| sys.process(pid))
+                .map(|process| (process.cpu_usage() as f64, process.memory(), total_mem))
         };
-        sysinfo::get_current_pid()
-            .ok()
-            .and_then(|pid| sys.process(pid))
-            .map(|process| (process.cpu_usage() as f64, process.memory(), total_mem))
+
+        process_metrics.map(|(cpu_usage, memory_bytes, total_mem)| {
+            (
+                cpu_usage,
+                memory_bytes,
+                total_mem,
+                cpu_limit_cores(),
+                cpu_usage_cores(cpu_usage),
+            )
+        })
     })
     .await
     .unwrap();
-    if let Some((cpu_usage, memory_bytes, total_mem)) = process_metrics {
-        record_process_metrics_sample(cpu_usage, memory_bytes, total_mem);
+    if let Some((cpu_usage, memory_bytes, total_mem, cpu_limit_cores, cpu_usage_cores)) =
+        process_metrics
+    {
+        record_process_metrics_sample(cpu_usage, memory_bytes, total_mem, cpu_limit_cores);
+        record_process_cpu_usage_cores(cpu_usage_cores);
     }
 
     let staging_path = PARSEABLE.options.staging_dir().clone();
@@ -165,3 +370,42 @@ pub async fn check_resource_utilization_middleware(
     // Continue processing the request if resource utilization is within limits
     next.call(req).await
 }
+
+#[cfg(all(test, target_os = "linux"))]
+mod tests {
+    use super::{cpu_quota_cores, cpu_usage_cores_between, cpu_usage_micros_from_stat};
+
+    #[test]
+    fn parses_limited_and_unlimited_cpu_quotas() {
+        assert_eq!(cpu_quota_cores("50000", "100000"), Ok(Some(0.5)));
+        assert_eq!(cpu_quota_cores("max", "100000"), Ok(None));
+        assert_eq!(cpu_quota_cores("-1", "100000"), Ok(None));
+    }
+
+    #[test]
+    fn rejects_invalid_cpu_quotas() {
+        assert_eq!(cpu_quota_cores("invalid", "100000"), Err(()));
+        assert_eq!(cpu_quota_cores("50000", "0"), Err(()));
+    }
+
+    #[test]
+    fn parses_cgroup_v2_cpu_usage() {
+        assert_eq!(
+            cpu_usage_micros_from_stat("usage_usec 10200000\nuser_usec 8000000\n"),
+            Ok(10_200_000)
+        );
+        assert_eq!(cpu_usage_micros_from_stat("user_usec 8000000\n"), Err(()));
+    }
+
+    #[test]
+    fn calculates_cpu_usage_in_cores() {
+        assert_eq!(
+            cpu_usage_cores_between(10_000_000, 10_200_000, 5_000_000),
+            Some(0.04)
+        );
+        assert_eq!(
+            cpu_usage_cores_between(10_200_000, 10_000_000, 5_000_000),
+            None
+        );
+    }
+}