From 5d90f96863b64e9bc88ed35429f8068a6a788d1e Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Thu, 8 Oct 2026 22:17:44 +0200 Subject: [PATCH] feat: export worker cpu, memory and occupancy metrics (#11629) * feat: export worker cpu, memory and occupancy metrics Co-Authored-By: Claude Opus 5.5 * fix: refresh the exported memory limit with the cgroup readings Co-Authored-By: Claude Opus 5.5 * 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 Co-authored-by: windmill-internal-app[bot] --- backend/ee-repo-ref.txt | 2 +- backend/tests/otel.rs | 27 +- backend/windmill-common/src/otel_oss.rs | 33 +++ backend/windmill-common/src/worker.rs | 93 +++++++ backend/windmill-worker/src/handle_child.rs | 1 + backend/windmill-worker/src/lib.rs | 1 + .../windmill-worker/src/resource_metrics.rs | 235 ++++++++++++++++++ backend/windmill-worker/src/worker.rs | 1 + backend/windmill-worker/src/worker_utils.rs | 13 +- 9 files changed, 402 insertions(+), 4 deletions(-) create mode 100644 backend/windmill-worker/src/resource_metrics.rs diff --git a/backend/ee-repo-ref.txt b/backend/ee-repo-ref.txt index 2de431e127..d292c22c2e 100644 --- a/backend/ee-repo-ref.txt +++ b/backend/ee-repo-ref.txt @@ -1 +1 @@ -ab8476e5feedc98b727f7d83acffa1807b3b9c5b +e128a9b9bedd0b6a46d355b69db13bf8f650c4e6 diff --git a/backend/tests/otel.rs b/backend/tests/otel.rs index f1e042037f..30585c08a1 100644 --- a/backend/tests/otel.rs +++ b/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(); diff --git a/backend/windmill-common/src/otel_oss.rs b/backend/windmill-common/src/otel_oss.rs index bd0a5202df..95e635dd8f 100644 --- a/backend/windmill-common/src/otel_oss.rs +++ b/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, + _windmill_usage: Option, + _limit: Option, +) { +} + +#[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) {} diff --git a/backend/windmill-common/src/worker.rs b/backend/windmill-common/src/worker.rs index 52fa2da905..017c244670 100644 --- a/backend/windmill-common/src/worker.rs +++ b/backend/windmill-common/src/worker.rs @@ -1783,6 +1783,73 @@ pub fn get_cpu_period() -> Option { 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 { + 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 { + 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 { + Some(CpuStat { + usage_usec: cpuacct_usage.trim().parse::().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 { + 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 { + 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); diff --git a/backend/windmill-worker/src/handle_child.rs b/backend/windmill-worker/src/handle_child.rs index c25ad82a83..84fd547641 100644 --- a/backend/windmill-worker/src/handle_child.rs +++ b/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); diff --git a/backend/windmill-worker/src/lib.rs b/backend/windmill-worker/src/lib.rs index 6059936af6..aa1d9ab209 100644 --- a/backend/windmill-worker/src/lib.rs +++ b/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; diff --git a/backend/windmill-worker/src/resource_metrics.rs b/backend/windmill-worker/src/resource_metrics.rs new file mode 100644 index 0000000000..53c7dc5b4d --- /dev/null +++ b/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 = + 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 = OnceLock::new(); +static MEMORY_LIMIT: Mutex>> = 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) { + *MEMORY_LIMIT.lock().unwrap() = Some(limit); +} +static LAST_CPU_STAT: Mutex = + 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, + wm_memory_usage: Option, + 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); +} diff --git a/backend/windmill-worker/src/worker.rs b/backend/windmill-worker/src/worker.rs index d0d5d8b4e2..5b5df51b30 100644 --- a/backend/windmill-worker/src/worker.rs +++ b/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)), diff --git a/backend/windmill-worker/src/worker_utils.rs b/backend/windmill-worker/src/worker_utils.rs index cabf004f4d..c634c13b6a 100644 --- a/backend/windmill-worker/src/worker_utils.rs +++ b/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) = (|| {