feat: export worker cpu, memory and occupancy metrics (#11629)

* feat: export worker cpu, memory and occupancy metrics

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* fix: refresh the exported memory limit with the cgroup readings

Co-Authored-By: Claude Opus 5.5 <noreply@anthropic.com>

* chore: update ee-repo-ref to e128a9b9bedd0b6a46d355b69db13bf8f650c4e6

This commit updates the EE repository reference after PR #864 was merged in windmill-ee-private.

Previous ee-repo-ref: b286a486c69e74ec0742a2aac9d4cb1161eb0c6b

New ee-repo-ref: e128a9b9bedd0b6a46d355b69db13bf8f650c4e6

Automated by sync-ee-ref workflow.

---------

Co-authored-by: Claude Opus 5.5 <noreply@anthropic.com>
Co-authored-by: windmill-internal-app[bot] <windmill-internal-app[bot]@users.noreply.github.com>

backend/ee-repo-ref.txt

@@ -1 +1 @@
-ab8476e5feedc98b727f7d83acffa1807b3b9c5b
+e128a9b9bedd0b6a46d355b69db13bf8f650c4e6

backend/tests/otel.rs

@@ -176,6 +176,10 @@ async fn test_all_metrics_e2e() {
     otel_set_worker_uptime("w-uptime", 3600.0);
     otel_set_health_status_phase("healthy");
     otel_set_health_db_unresponsive(true);
+    otel_set_worker_memory("w-res", "gpu", Some(512), Some(128), Some(1024));
+    otel_add_worker_cpu("w-res", "gpu", 1.5, 40, 7, 0.35);
+    otel_set_worker_occupancy_rate("w-res", "gpu", "5m", 0.25);
+    otel_record_job_memory_peak("python3", 100 * 1024 * 1024);
 
     // ── Histograms ──────────────────────────────────────────────────
 
@@ -189,7 +193,7 @@ async fn test_all_metrics_e2e() {
     let metrics = flush_and_get_metrics(&state);
     let names = metric_names(&metrics);
 
-    // ── Verify all 21 metric names are present ──────────────────────
+    // ── Verify all metric names are present ─────────────────────────
 
     let expected = [
         "windmill.queue.push_count",
@@ -213,6 +217,15 @@ async fn test_all_metrics_e2e() {
         "windmill.worker.uptime",
         "windmill.health.status",
         "windmill.health.db_unresponsive",
+        "windmill.worker.memory_usage",
+        "windmill.worker.windmill_memory_usage",
+        "windmill.worker.memory_limit",
+        "windmill.worker.cpu_usage",
+        "windmill.worker.cpu_periods",
+        "windmill.worker.cpu_throttled_periods",
+        "windmill.worker.cpu_throttled_time",
+        "windmill.worker.occupancy_rate",
+        "windmill.worker.job_memory_peak",
     ];
     for name in expected {
         assert!(
@@ -290,6 +303,18 @@ async fn test_all_metrics_e2e() {
         .expect("worker.busy data point with worker=worker-test-1 not found");
     assert_eq!(dp.1, 1);
 
+    let m = find_metric(&metrics, "windmill.worker.memory_limit").unwrap();
+    let values = gauge_i64_values(m);
+    let dp = values
+        .iter()
+        .find(|(attrs, _)| {
+            attrs
+                .iter()
+                .any(|kv| kv.key.as_str() == "worker_group" && kv.value.as_str() == "gpu")
+        })
+        .expect("worker.memory_limit data point with worker_group=gpu not found");
+    assert_eq!(dp.1, 1024);
+
     let m = find_metric(&metrics, "windmill.db.pool.active").unwrap();
     assert_eq!(gauge_i64_values(m)[0].1, 5);
     let m = find_metric(&metrics, "windmill.db.pool.idle").unwrap();

backend/windmill-common/src/otel_oss.rs

@@ -132,3 +132,36 @@ pub fn otel_set_health_status_phase(_phase: &str) {}
 
 #[cfg(not(feature = "private"))]
 pub fn otel_set_health_db_unresponsive(_unresponsive: bool) {}
+
+#[cfg(not(feature = "private"))]
+pub fn otel_set_worker_memory(
+    _worker: &str,
+    _worker_group: &str,
+    _usage: Option<i64>,
+    _windmill_usage: Option<i64>,
+    _limit: Option<i64>,
+) {
+}
+
+#[cfg(not(feature = "private"))]
+pub fn otel_add_worker_cpu(
+    _worker: &str,
+    _worker_group: &str,
+    _usage_secs: f64,
+    _periods: u64,
+    _throttled_periods: u64,
+    _throttled_secs: f64,
+) {
+}
+
+#[cfg(not(feature = "private"))]
+pub fn otel_set_worker_occupancy_rate(
+    _worker: &str,
+    _worker_group: &str,
+    _window: &'static str,
+    _rate: f64,
+) {
+}
+
+#[cfg(not(feature = "private"))]
+pub fn otel_record_job_memory_peak(_tag: &str, _bytes: u64) {}

backend/windmill-common/src/worker.rs

@@ -1783,6 +1783,73 @@ pub fn get_cpu_period() -> Option<i64> {
     Some(100000)
 }
 
+/// Bucket bounds in bytes, 16 MiB to 32 GiB, shared by the Prometheus and OTel job
+/// peak-memory histograms so the two exports stay comparable.
+pub const JOB_MEMORY_PEAK_BUCKETS: [f64; 12] = {
+    let mut buckets = [0.0; 12];
+    let mut i = 0;
+    while i < 12 {
+        buckets[i] = (16u64 << (20 + i)) as f64;
+        i += 1;
+    }
+    buckets
+};
+
+/// Cumulative CPU accounting of the worker's cgroup. The throttling fields stay at 0
+/// when the cgroup has no CPU controller, which is also what an unthrottled one reports.
+#[derive(Debug, Clone, Copy, Default, PartialEq)]
+pub struct CpuStat {
+    pub usage_usec: u64,
+    pub nr_periods: u64,
+    pub nr_throttled: u64,
+    pub throttled_usec: u64,
+}
+
+fn cpu_stat_field(stat: &str, key: &str) -> Option<u64> {
+    stat.lines().find_map(|line| {
+        let (k, v) = line.split_once(' ')?;
+        (k == key).then(|| v.trim().parse().ok()).flatten()
+    })
+}
+
+fn parse_cgroup_v2_cpu_stat(stat: &str) -> Option<CpuStat> {
+    Some(CpuStat {
+        usage_usec: cpu_stat_field(stat, "usage_usec")?,
+        nr_periods: cpu_stat_field(stat, "nr_periods").unwrap_or(0),
+        nr_throttled: cpu_stat_field(stat, "nr_throttled").unwrap_or(0),
+        throttled_usec: cpu_stat_field(stat, "throttled_usec").unwrap_or(0),
+    })
+}
+
+/// cgroup v1 splits the same figures over two controllers and counts time in nanoseconds.
+fn parse_cgroup_v1_cpu_stat(cpuacct_usage: &str, stat: &str) -> Option<CpuStat> {
+    Some(CpuStat {
+        usage_usec: cpuacct_usage.trim().parse::<u64>().ok()? / 1000,
+        nr_periods: cpu_stat_field(stat, "nr_periods").unwrap_or(0),
+        nr_throttled: cpu_stat_field(stat, "nr_throttled").unwrap_or(0),
+        throttled_usec: cpu_stat_field(stat, "throttled_time").unwrap_or(0) / 1000,
+    })
+}
+
+#[cfg(not(windows))]
+pub fn get_cpu_stat() -> Option<CpuStat> {
+    const V1_USAGE: &str = "/sys/fs/cgroup/cpuacct/cpuacct.usage";
+    if Path::new(V1_USAGE).exists() {
+        parse_cgroup_v1_cpu_stat(
+            &std::fs::read_to_string(V1_USAGE).ok()?,
+            &std::fs::read_to_string("/sys/fs/cgroup/cpu/cpu.stat").unwrap_or_default(),
+        )
+    } else {
+        let cgroup_path = get_cgroupv2_path()?;
+        parse_cgroup_v2_cpu_stat(&std::fs::read_to_string(format!("{cgroup_path}/cpu.stat")).ok()?)
+    }
+}
+
+#[cfg(windows)]
+pub fn get_cpu_stat() -> Option<CpuStat> {
+    None
+}
+
 /// CPUs the process is allowed to run on, ignoring any bandwidth quota — the count
 /// Go's `NumCPU` reports. `available_parallelism` cannot stand in for it: that folds
 /// the quota in, so a fraction of a CPU makes it report a single-core machine.
@@ -2919,6 +2986,32 @@ mod tests {
     use super::*;
     use std::collections::HashMap;
 
+    #[test]
+    fn cgroup_cpu_stat_is_normalized_to_microseconds() {
+        let v2 = "usage_usec 2500000\nuser_usec 2000000\nsystem_usec 500000\n\
+                  nr_periods 40\nnr_throttled 7\nthrottled_usec 350000\nnr_bursts 0\n";
+        let expected = CpuStat {
+            usage_usec: 2_500_000,
+            nr_periods: 40,
+            nr_throttled: 7,
+            throttled_usec: 350_000,
+        };
+        assert_eq!(parse_cgroup_v2_cpu_stat(v2), Some(expected));
+
+        let v1_stat = "nr_periods 40\nnr_throttled 7\nthrottled_time 350000000\n";
+        assert_eq!(
+            parse_cgroup_v1_cpu_stat("2500000000\n", v1_stat),
+            Some(expected)
+        );
+
+        // No CPU controller: usage is still reported, throttling reads as zero.
+        assert_eq!(
+            parse_cgroup_v2_cpu_stat("usage_usec 12\nuser_usec 8\nsystem_usec 4\n"),
+            Some(CpuStat { usage_usec: 12, ..Default::default() })
+        );
+        assert_eq!(parse_cgroup_v2_cpu_stat("nr_periods 40\n"), None);
+    }
+
     #[test]
     fn datatable_role_is_read_from_the_leading_comment_block() {
         let role = |code| SqlAnnotations::datatable_role(code);

backend/windmill-worker/src/handle_child.rs

@@ -1049,6 +1049,7 @@ where
                     let wm_memory_usage = get_windmill_memory_usage();
                     tracing::info!("job {job_id} on {worker_name} in {w_id} worker memory snapshot {}kB/{}kB", memory_usage.unwrap_or_default()/1024, wm_memory_usage.unwrap_or_default()/1024);
                     let occupancy = occupancy_metrics.as_mut().map(|x| x.update_occupancy_metrics());
+                    crate::resource_metrics::record_worker_resources(worker_name, memory_usage, wm_memory_usage, occupancy.as_ref());
                     if job_id != Uuid::nil() {
                         if let Err(err) = update_worker_ping_from_job(&conn, &job_id, w_id, worker_name, memory_usage, wm_memory_usage, occupancy).await {
                             tracing::error!("Unable to update worker ping for job {} in workspace {}. Error was: {:?}", job_id, w_id, err);

backend/windmill-worker/src/lib.rs

@@ -81,6 +81,7 @@ mod prepare_deps;
 mod python_executor;
 #[cfg(feature = "python")]
 mod python_versions;
+mod resource_metrics;
 pub mod result_processor;
 #[cfg(feature = "rust")]
 mod rust_executor;

backend/windmill-worker/src/resource_metrics.rs

@@ -0,0 +1,235 @@
+use std::sync::{atomic::Ordering, Mutex, OnceLock};
+
+use windmill_common::{
+    otel_oss::{
+        otel_add_worker_cpu, otel_record_job_memory_peak, otel_set_worker_memory,
+        otel_set_worker_occupancy_rate,
+    },
+    worker::{get_cpu_stat, get_memory, CpuStat, WORKER_GROUP},
+    OTEL_METRICS_ENABLED,
+};
+
+use crate::common::OccupancyResult;
+
+#[cfg(feature = "prometheus")]
+struct PromMetrics {
+    memory_usage: prometheus::IntGaugeVec,
+    windmill_memory_usage: prometheus::IntGaugeVec,
+    memory_limit: prometheus::IntGaugeVec,
+    cpu_usage: prometheus::CounterVec,
+    cpu_periods: prometheus::IntCounterVec,
+    cpu_throttled_periods: prometheus::IntCounterVec,
+    cpu_throttled_seconds: prometheus::CounterVec,
+    occupancy_rate: prometheus::GaugeVec,
+    job_memory_peak: prometheus::HistogramVec,
+}
+
+#[cfg(feature = "prometheus")]
+lazy_static::lazy_static! {
+    static ref PROM_METRICS: Option<PromMetrics> =
+        if windmill_common::METRICS_ENABLED.load(Ordering::Relaxed) {
+            const WORKER: &[&str] = &["name", "worker_group"];
+            Some(PromMetrics {
+                memory_usage: prometheus::register_int_gauge_vec!(
+                    "worker_memory_usage_bytes",
+                    "Memory used by the worker's container, page cache excluded",
+                    WORKER
+                )
+                .unwrap(),
+                windmill_memory_usage: prometheus::register_int_gauge_vec!(
+                    "worker_windmill_memory_usage_bytes",
+                    "Resident memory of the worker process itself",
+                    WORKER
+                )
+                .unwrap(),
+                memory_limit: prometheus::register_int_gauge_vec!(
+                    "worker_memory_limit_bytes",
+                    "Memory limit of the worker's container",
+                    WORKER
+                )
+                .unwrap(),
+                cpu_usage: prometheus::register_counter_vec!(
+                    "worker_cpu_usage_seconds_total",
+                    "CPU time consumed by the worker's container",
+                    WORKER
+                )
+                .unwrap(),
+                cpu_periods: prometheus::register_int_counter_vec!(
+                    "worker_cpu_periods_total",
+                    "CPU enforcement periods elapsed for the worker's container",
+                    WORKER
+                )
+                .unwrap(),
+                cpu_throttled_periods: prometheus::register_int_counter_vec!(
+                    "worker_cpu_throttled_periods_total",
+                    "CPU enforcement periods in which the worker's container was throttled",
+                    WORKER
+                )
+                .unwrap(),
+                cpu_throttled_seconds: prometheus::register_counter_vec!(
+                    "worker_cpu_throttled_seconds_total",
+                    "Time the worker's container spent throttled by its CPU limit",
+                    WORKER
+                )
+                .unwrap(),
+                occupancy_rate: prometheus::register_gauge_vec!(
+                    "worker_occupancy_rate",
+                    "Share of time the worker spent executing jobs, over the window",
+                    &["name", "worker_group", "window"]
+                )
+                .unwrap(),
+                job_memory_peak: prometheus::register_histogram_vec!(
+                    "worker_job_memory_peak_bytes",
+                    "Peak memory of a job's main process",
+                    &["tag"],
+                    windmill_common::worker::JOB_MEMORY_PEAK_BUCKETS.to_vec()
+                )
+                .unwrap(),
+            })
+        } else {
+            None
+        };
+}
+
+fn metrics_enabled() -> bool {
+    #[cfg(feature = "prometheus")]
+    if PROM_METRICS.is_some() {
+        return true;
+    }
+    OTEL_METRICS_ENABLED.load(Ordering::Relaxed)
+}
+
+// Memory and CPU are read from the cgroup, which every worker of a process shares
+// (native mode runs several). Only the first worker to report exports them, so that
+// summing the series over workers does not count one container several times.
+static CONTAINER_SERIES_OWNER: OnceLock<String> = OnceLock::new();
+static MEMORY_LIMIT: Mutex<Option<Option<i64>>> = Mutex::new(None);
+
+/// Replaces the exported memory limit, for the ping that re-reads the cgroup limits.
+pub(crate) fn set_memory_limit(limit: Option<i64>) {
+    *MEMORY_LIMIT.lock().unwrap() = Some(limit);
+}
+static LAST_CPU_STAT: Mutex<CpuStat> =
+    Mutex::new(CpuStat { usage_usec: 0, nr_periods: 0, nr_throttled: 0, throttled_usec: 0 });
+
+/// Exports the worker's resource readings. Meant for the ping ticks, which have already
+/// read memory usage and occupancy; the only extra work is the cgroup `cpu.stat` read.
+pub(crate) fn record_worker_resources(
+    worker_name: &str,
+    memory_usage: Option<i64>,
+    wm_memory_usage: Option<i64>,
+    occupancy: Option<&OccupancyResult>,
+) {
+    if !metrics_enabled() {
+        return;
+    }
+    let worker_group = WORKER_GROUP.as_str();
+
+    if let Some(occupancy) = occupancy {
+        let windows = [
+            ("total", Some(occupancy.occupancy_rate)),
+            ("15s", occupancy.occupancy_rate_15s),
+            ("5m", occupancy.occupancy_rate_5m),
+            ("30m", occupancy.occupancy_rate_30m),
+        ];
+        for (window, rate) in windows {
+            let Some(rate) = rate.filter(|r| r.is_finite()) else {
+                continue;
+            };
+            #[cfg(feature = "prometheus")]
+            if let Some(m) = PROM_METRICS.as_ref() {
+                m.occupancy_rate
+                    .with_label_values(&[worker_name, worker_group, window])
+                    .set(rate as f64);
+            }
+            otel_set_worker_occupancy_rate(worker_name, worker_group, window, rate as f64);
+        }
+    }
+
+    if CONTAINER_SERIES_OWNER.get_or_init(|| worker_name.to_string()) != worker_name {
+        return;
+    }
+
+    let memory_limit = *MEMORY_LIMIT.lock().unwrap().get_or_insert_with(get_memory);
+    #[cfg(feature = "prometheus")]
+    if let Some(m) = PROM_METRICS.as_ref() {
+        let labels = &[worker_name, worker_group];
+        if let Some(v) = memory_usage {
+            m.memory_usage.with_label_values(labels).set(v);
+        }
+        if let Some(v) = wm_memory_usage {
+            m.windmill_memory_usage.with_label_values(labels).set(v);
+        }
+        if let Some(v) = memory_limit {
+            m.memory_limit.with_label_values(labels).set(v);
+        }
+    }
+    otel_set_worker_memory(
+        worker_name,
+        worker_group,
+        memory_usage,
+        wm_memory_usage,
+        memory_limit,
+    );
+
+    let Some(stat) = get_cpu_stat() else {
+        return;
+    };
+    let delta = {
+        let mut last = LAST_CPU_STAT.lock().unwrap();
+        // A reading below the previous one means the cgroup changed under the process:
+        // restart from it rather than exporting a negative increase.
+        let delta = if stat.usage_usec < last.usage_usec {
+            CpuStat::default()
+        } else {
+            CpuStat {
+                usage_usec: stat.usage_usec - last.usage_usec,
+                nr_periods: stat.nr_periods.saturating_sub(last.nr_periods),
+                nr_throttled: stat.nr_throttled.saturating_sub(last.nr_throttled),
+                throttled_usec: stat.throttled_usec.saturating_sub(last.throttled_usec),
+            }
+        };
+        *last = stat;
+        delta
+    };
+    let usage_secs = delta.usage_usec as f64 / 1e6;
+    let throttled_secs = delta.throttled_usec as f64 / 1e6;
+    #[cfg(feature = "prometheus")]
+    if let Some(m) = PROM_METRICS.as_ref() {
+        let labels = &[worker_name, worker_group];
+        m.cpu_usage.with_label_values(labels).inc_by(usage_secs);
+        m.cpu_periods
+            .with_label_values(labels)
+            .inc_by(delta.nr_periods);
+        m.cpu_throttled_periods
+            .with_label_values(labels)
+            .inc_by(delta.nr_throttled);
+        m.cpu_throttled_seconds
+            .with_label_values(labels)
+            .inc_by(throttled_secs);
+    }
+    otel_add_worker_cpu(
+        worker_name,
+        worker_group,
+        usage_secs,
+        delta.nr_periods,
+        delta.nr_throttled,
+        throttled_secs,
+    );
+}
+
+/// `mem_peak_kb` is the job's `mem_peak`: non-positive when the executor had no process
+/// to sample or the reading failed, and those are not observations.
+pub(crate) fn record_job_memory_peak(tag: &str, mem_peak_kb: i32) {
+    if mem_peak_kb <= 0 {
+        return;
+    }
+    let bytes = mem_peak_kb as u64 * 1024;
+    #[cfg(feature = "prometheus")]
+    if let Some(m) = PROM_METRICS.as_ref() {
+        m.job_memory_peak
+            .with_label_values(&[tag])
+            .observe(bytes as f64);
+    }
+    otel_record_job_memory_peak(tag, bytes);
+}

backend/windmill-worker/src/worker.rs

@@ -5130,6 +5130,7 @@ pub async fn handle_queued_job(
             // WAC v2 job suspended while waiting for child jobs — don't complete it
             return Ok(JobOutcome::Completed);
         }
+        crate::resource_metrics::record_job_memory_peak(&cjob.tag, mem_peak);
         process_result(
             cjob,
             result.map(|x| Arc::new(x)),

backend/windmill-worker/src/worker_utils.rs

@@ -39,17 +39,26 @@ pub(crate) async fn update_worker_ping_full(
     let wm_memory_usage = get_windmill_memory_usage();
 
     let (vcpus, memory) = if read_cgroups {
-        (get_vcpus(), get_memory())
+        let memory = get_memory();
+        crate::resource_metrics::set_memory_limit(memory);
+        (get_vcpus(), memory)
     } else {
         (None, None)
     };
 
+    let occupancy = occupancy_metrics.update_occupancy_metrics();
+    crate::resource_metrics::record_worker_resources(
+        worker_name,
+        memory_usage,
+        wm_memory_usage,
+        Some(&occupancy),
+    );
     let OccupancyResult {
         occupancy_rate,
         occupancy_rate_15s,
         occupancy_rate_5m,
         occupancy_rate_30m,
-    } = occupancy_metrics.update_occupancy_metrics();
+    } = occupancy;
 
     let ping_start = std::time::Instant::now();
     if let Err(e) = (|| {