From 9d75929247ea2ec39286971fcbebf95d886194f3 Mon Sep 17 00:00:00 2001 From: Ruben Fiszel Date: Thu, 10 Sep 2026 16:09:19 +0200 Subject: [PATCH] perf: only write queue metrics when a tag's backlog changes (#11055) * perf: only write queue metrics when a tag's backlog changes Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Vz5TLoq492nruNSCr6LARA * fix: hold queue metric steps until the next sample and skip failed reads Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Vz5TLoq492nruNSCr6LARA * fix: keep running-count gauges on a failed backlog read and widen stale margins Co-Authored-By: Claude Opus 5 (1M context) Claude-Session: https://claude.ai/code/session_01Vz5TLoq492nruNSCr6LARA --------- Co-authored-by: Claude Opus 5 (1M context) --- ...de74e8b0943d30481465da453942e299a128f.json | 15 - ...ebfaa3e7ec3a7c682945f22e8cde3a63094f4.json | 15 + ...46fc4a48172f104e9e849f755ac041c3eef92.json | 20 - ...3a6c1ff99973e778d71bd598a50d022da8f15.json | 15 - ...7d7b7776b90a96c6662b00d8a999b74eb5b36.json | 35 ++ ...e08ea1695e41d9b11538b58189925ddba4fdd.json | 32 ++ backend/src/monitor.rs | 459 ++++++++++++------ backend/windmill-common/src/queue.rs | 40 ++ backend/windmill-common/tests/queue_stats.rs | 63 +++ .../components/QueueMetricsDrawerInner.svelte | 108 +++-- 10 files changed, 564 insertions(+), 238 deletions(-) delete mode 100644 backend/.sqlx/query-00e63eab76d26e148b77e932848de74e8b0943d30481465da453942e299a128f.json create mode 100644 backend/.sqlx/query-38e12be0764d850538d3eea328aebfaa3e7ec3a7c682945f22e8cde3a63094f4.json delete mode 100644 backend/.sqlx/query-82f6674f19e8ad51a992505a46f46fc4a48172f104e9e849f755ac041c3eef92.json delete mode 100644 backend/.sqlx/query-8824b382c4e98dfa17b4aa656af3a6c1ff99973e778d71bd598a50d022da8f15.json create mode 100644 backend/.sqlx/query-bd46b7a19f9e51de1babc3d16fd7d7b7776b90a96c6662b00d8a999b74eb5b36.json create mode 100644 backend/.sqlx/query-d09d3a6223262ac584f045feec7e08ea1695e41d9b11538b58189925ddba4fdd.json create mode 100644 backend/windmill-common/tests/queue_stats.rs diff --git a/backend/.sqlx/query-00e63eab76d26e148b77e932848de74e8b0943d30481465da453942e299a128f.json b/backend/.sqlx/query-00e63eab76d26e148b77e932848de74e8b0943d30481465da453942e299a128f.json deleted file mode 100644 index 9c1f8d00a7..0000000000 --- a/backend/.sqlx/query-00e63eab76d26e148b77e932848de74e8b0943d30481465da453942e299a128f.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO metrics (id, value)\n VALUES ($1, to_jsonb((\n SELECT EXTRACT(EPOCH FROM now() - scheduled_for)\n FROM v2_job_queue\n WHERE tag = $2 AND running = false AND scheduled_for <= now() - ('3 seconds')::interval\n ORDER BY priority DESC NULLS LAST, scheduled_for LIMIT 1\n )))", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Text" - ] - }, - "nullable": [] - }, - "hash": "00e63eab76d26e148b77e932848de74e8b0943d30481465da453942e299a128f" -} diff --git a/backend/.sqlx/query-38e12be0764d850538d3eea328aebfaa3e7ec3a7c682945f22e8cde3a63094f4.json b/backend/.sqlx/query-38e12be0764d850538d3eea328aebfaa3e7ec3a7c682945f22e8cde3a63094f4.json new file mode 100644 index 0000000000..fe9c2f27e3 --- /dev/null +++ b/backend/.sqlx/query-38e12be0764d850538d3eea328aebfaa3e7ec3a7c682945f22e8cde3a63094f4.json @@ -0,0 +1,15 @@ +{ + "db_name": "PostgreSQL", + "query": "INSERT INTO metrics (id, value) SELECT * FROM unnest($1::text[], $2::jsonb[])", + "describe": { + "columns": [], + "parameters": { + "Left": [ + "TextArray", + "JsonbArray" + ] + }, + "nullable": [] + }, + "hash": "38e12be0764d850538d3eea328aebfaa3e7ec3a7c682945f22e8cde3a63094f4" +} diff --git a/backend/.sqlx/query-82f6674f19e8ad51a992505a46f46fc4a48172f104e9e849f755ac041c3eef92.json b/backend/.sqlx/query-82f6674f19e8ad51a992505a46f46fc4a48172f104e9e849f755ac041c3eef92.json deleted file mode 100644 index f98f3fdeea..0000000000 --- a/backend/.sqlx/query-82f6674f19e8ad51a992505a46f46fc4a48172f104e9e849f755ac041c3eef92.json +++ /dev/null @@ -1,20 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "SELECT created_at FROM metrics WHERE id LIKE 'queue_count_%' ORDER BY created_at DESC LIMIT 1", - "describe": { - "columns": [ - { - "ordinal": 0, - "name": "created_at", - "type_info": "Timestamptz" - } - ], - "parameters": { - "Left": [] - }, - "nullable": [ - false - ] - }, - "hash": "82f6674f19e8ad51a992505a46f46fc4a48172f104e9e849f755ac041c3eef92" -} diff --git a/backend/.sqlx/query-8824b382c4e98dfa17b4aa656af3a6c1ff99973e778d71bd598a50d022da8f15.json b/backend/.sqlx/query-8824b382c4e98dfa17b4aa656af3a6c1ff99973e778d71bd598a50d022da8f15.json deleted file mode 100644 index 124a6e36c5..0000000000 --- a/backend/.sqlx/query-8824b382c4e98dfa17b4aa656af3a6c1ff99973e778d71bd598a50d022da8f15.json +++ /dev/null @@ -1,15 +0,0 @@ -{ - "db_name": "PostgreSQL", - "query": "INSERT INTO metrics (id, value) VALUES ($1, $2)", - "describe": { - "columns": [], - "parameters": { - "Left": [ - "Varchar", - "Jsonb" - ] - }, - "nullable": [] - }, - "hash": "8824b382c4e98dfa17b4aa656af3a6c1ff99973e778d71bd598a50d022da8f15" -} diff --git a/backend/.sqlx/query-bd46b7a19f9e51de1babc3d16fd7d7b7776b90a96c6662b00d8a999b74eb5b36.json b/backend/.sqlx/query-bd46b7a19f9e51de1babc3d16fd7d7b7776b90a96c6662b00d8a999b74eb5b36.json new file mode 100644 index 0000000000..5dce7b09cc --- /dev/null +++ b/backend/.sqlx/query-bd46b7a19f9e51de1babc3d16fd7d7b7776b90a96c6662b00d8a999b74eb5b36.json @@ -0,0 +1,35 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT COALESCE(c.id, r.id) AS \"id!\", r.value AS \"value?\",\n EXTRACT(EPOCH FROM now() - r.created_at)::double precision AS \"age?\"\n FROM unnest($1::text[]) AS c(id)\n FULL JOIN (\n SELECT DISTINCT ON (id) id, value, created_at\n FROM metrics\n WHERE id LIKE 'queue_%' AND created_at > now() - make_interval(secs => $2)\n ORDER BY id, created_at DESC\n ) r ON r.id = c.id", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "id!", + "type_info": "Text" + }, + { + "ordinal": 1, + "name": "value?", + "type_info": "Jsonb" + }, + { + "ordinal": 2, + "name": "age?", + "type_info": "Float8" + } + ], + "parameters": { + "Left": [ + "TextArray", + "Float8" + ] + }, + "nullable": [ + true, + true, + true + ] + }, + "hash": "bd46b7a19f9e51de1babc3d16fd7d7b7776b90a96c6662b00d8a999b74eb5b36" +} diff --git a/backend/.sqlx/query-d09d3a6223262ac584f045feec7e08ea1695e41d9b11538b58189925ddba4fdd.json b/backend/.sqlx/query-d09d3a6223262ac584f045feec7e08ea1695e41d9b11538b58189925ddba4fdd.json new file mode 100644 index 0000000000..ad7d1bf057 --- /dev/null +++ b/backend/.sqlx/query-d09d3a6223262ac584f045feec7e08ea1695e41d9b11538b58189925ddba4fdd.json @@ -0,0 +1,32 @@ +{ + "db_name": "PostgreSQL", + "query": "SELECT tag AS \"tag!\", sum(n)::bigint AS \"count!\",\n EXTRACT(EPOCH FROM now() - (array_agg(head ORDER BY priority DESC NULLS LAST))[1])\n ::double precision AS \"delay!\"\n FROM (\n SELECT tag, priority, count(*) AS n, min(scheduled_for) AS head\n FROM v2_job_queue WHERE\n scheduled_for <= now() - ('3 seconds')::interval AND running = false\n GROUP BY tag, priority\n ) g\n GROUP BY tag", + "describe": { + "columns": [ + { + "ordinal": 0, + "name": "tag!", + "type_info": "Varchar" + }, + { + "ordinal": 1, + "name": "count!", + "type_info": "Int8" + }, + { + "ordinal": 2, + "name": "delay!", + "type_info": "Float8" + } + ], + "parameters": { + "Left": [] + }, + "nullable": [ + false, + null, + null + ] + }, + "hash": "d09d3a6223262ac584f045feec7e08ea1695e41d9b11538b58189925ddba4fdd" +} diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index 867d2c8520..de46f0b9c8 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -5106,155 +5106,211 @@ async fn vacuuming_tables(db: &Pool) -> error::Result<()> { Ok(()) } -pub async fn expose_queue_metrics(db: &Pool) { - let last_check = sqlx::query_scalar!( - "SELECT created_at FROM metrics WHERE id LIKE 'queue_count_%' ORDER BY created_at DESC LIMIT 1" - ) - .fetch_optional(db) - .await - .unwrap_or(Some(chrono::Utc::now())); +/// Shortest spacing between two stored samples of the same queue metric, so a tag whose +/// value moves on every monitor round still writes at most one row per interval. Also how +/// often each server samples the queue when no Prometheus or OTel gauge needs it sooner. +const QUEUE_METRIC_MIN_INTERVAL_SECS: f64 = 25.0; +/// A backlogged tag whose value has not moved is re-sampled only this often. The queue metrics +/// drawer (`QueueMetricsDrawerInner.svelte`, which must be changed with it) treats a series +/// silent for longer as drained. That is how a tag whose drop to zero was never recorded +/// (no server was up when it drained) stops being drawn as backlogged. A longer heartbeat +/// writes fewer rows but keeps such a stale line up for longer. +const QUEUE_METRIC_HEARTBEAT_SECS: f64 = 5.0 * 60.0; +/// How far back the last-sample lookup reaches. Must exceed the real spacing of heartbeats, +/// which land up to a monitor tick and a sampling slot late, so a tag that is still believed +/// backlogged is always found and can be given its closing zero. +const QUEUE_METRIC_LOOKBACK_SECS: f64 = 3.0 * QUEUE_METRIC_HEARTBEAT_SECS; +/// Queue delay climbs on its own for as long as a tag stays backlogged, so an exact-value +/// comparison would never dedup it. Only a move the chart would actually render is stored. +const QUEUE_DELAY_TOLERANCE: f64 = 0.1; - let metrics_enabled = METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed); - let save_metrics = last_check - .map(|last_check| chrono::Utc::now() - last_check > chrono::Duration::seconds(25)) - .unwrap_or(true); +const QUEUE_COUNT_PREFIX: &str = "queue_count_"; +const QUEUE_DELAY_PREFIX: &str = "queue_delay_"; - if metrics_enabled || save_metrics || OTEL_METRICS_ENABLED.load(Ordering::Relaxed) { - let queue_counts = windmill_common::queue::get_queue_counts(db).await; +/// Append the queue metrics the drawer at `GET /workers/queue_metrics` charts, skipping any +/// sample that repeats what is already stored. +/// +/// Only tags with a backlog appear in `queue_stats`, and an arbitrary `?tag=` nobody serves +/// stays backlogged forever, so writing every round would repeat the same pair of rows for +/// the whole 14-day retention. Each metric is written when its value moves, once per +/// heartbeat while it holds, and once more (as a zero) when the tag drains. Gaps therefore +/// mean "unchanged since the last row", which is what the chart interpolates. +async fn save_queue_metrics( + db: &Pool, + queue_stats: &std::collections::HashMap, +) { + let sampled_ids = queue_stats + .keys() + .flat_map(|tag| { + [ + format!("{QUEUE_COUNT_PREFIX}{tag}"), + format!("{QUEUE_DELAY_PREFIX}{tag}"), + ] + }) + .collect::>(); - #[cfg(feature = "prometheus")] - if metrics_enabled { - for q in QUEUE_COUNT_TAGS.read().await.iter() { - if queue_counts.get(q).is_none() { - (*QUEUE_COUNT).with_label_values(&[q]).set(0); - } - } + // Last stored sample of every metric that either has a backlog now or was written + // recently enough to still be believed backlogged. Bounding the lookup by the window keeps + // it cheap at any `metrics` size; a per-id `ORDER BY created_at DESC LIMIT 1` does not, + // since the planner may serve it from `metrics_sort_idx` and walk the whole table. + let last_samples = match sqlx::query!( + "SELECT COALESCE(c.id, r.id) AS \"id!\", r.value AS \"value?\", + EXTRACT(EPOCH FROM now() - r.created_at)::double precision AS \"age?\" + FROM unnest($1::text[]) AS c(id) + FULL JOIN ( + SELECT DISTINCT ON (id) id, value, created_at + FROM metrics + WHERE id LIKE 'queue_%' AND created_at > now() - make_interval(secs => $2) + ORDER BY id, created_at DESC + ) r ON r.id = c.id", + &sampled_ids[..], + QUEUE_METRIC_LOOKBACK_SECS, + ) + .fetch_all(db) + .await + { + Ok(rows) => rows, + Err(e) => { + tracing::error!("Failed to read last queue metrics samples: {e:#}"); + return; } + }; - let otel_enabled = OTEL_METRICS_ENABLED.load(Ordering::Relaxed); - - if otel_enabled { - for q in OTEL_QUEUE_COUNT_TAGS.read().await.iter() { - if queue_counts.get(q).is_none() { - otel_set_queue_count(q, 0); - } + let mut ids = vec![]; + let mut values = vec![]; + for row in last_samples { + let Some((prefix, tag)) = [QUEUE_COUNT_PREFIX, QUEUE_DELAY_PREFIX] + .into_iter() + .find_map(|p| row.id.strip_prefix(p).map(|tag| (p, tag))) + else { + continue; + }; + // A stored value that is not a number cannot be compared, so the next reading is kept. + let last = row.value.as_ref().and_then(|v| v.as_f64()).zip(row.age); + let stat = queue_stats.get(tag); + let current = stat.map(|stat| { + if prefix == QUEUE_COUNT_PREFIX { + stat.count as f64 + } else { + stat.delay } - } + }); - #[allow(unused_mut)] - let mut tags_to_watch = vec![]; - #[allow(unused_mut)] - let mut otel_tags_to_watch = vec![]; - for q in queue_counts { - let count = q.1; - let tag = q.0; - - #[cfg(feature = "prometheus")] - if metrics_enabled { - let metric = (*QUEUE_COUNT).with_label_values(&[&tag]); - metric.set(count as i64); - tags_to_watch.push(tag.to_string()); - } - - if otel_enabled { - otel_tags_to_watch.push(tag.to_string()); - } - otel_set_queue_count(&tag, count as i64); - - // save queue_count and delay metrics per tag - if save_metrics { - sqlx::query!( - "INSERT INTO metrics (id, value) VALUES ($1, $2)", - format!("queue_count_{}", tag), - serde_json::json!(count) - ) - .execute(db) - .await - .ok(); - if count > 0 { - sqlx::query!( - "INSERT INTO metrics (id, value) - VALUES ($1, to_jsonb(( - SELECT EXTRACT(EPOCH FROM now() - scheduled_for) - FROM v2_job_queue - WHERE tag = $2 AND running = false AND scheduled_for <= now() - ('3 seconds')::interval - ORDER BY priority DESC NULLS LAST, scheduled_for LIMIT 1 - )))", - format!("queue_delay_{}", tag), - tag - ) - .execute(db) - .await - .ok(); - } - } - } - if metrics_enabled { - let mut w = QUEUE_COUNT_TAGS.write().await; - *w = tags_to_watch; - } - if otel_enabled { - let mut w = OTEL_QUEUE_COUNT_TAGS.write().await; - *w = otel_tags_to_watch; - } - - // Single DB query for running counts, shared by Prometheus and OTel - let otel_running = otel_enabled; - #[cfg(feature = "prometheus")] - let need_running_counts = metrics_enabled || otel_running; - #[cfg(not(feature = "prometheus"))] - let need_running_counts = otel_running; - - if need_running_counts { - let queue_running_counts = windmill_common::queue::get_queue_running_counts(db).await; - - #[cfg(feature = "prometheus")] - if metrics_enabled { - for q in QUEUE_RUNNING_COUNT_TAGS.read().await.iter() { - if queue_running_counts.get(q).is_none() { - (*QUEUE_RUNNING_COUNT).with_label_values(&[q]).set(0); - } - } - } - - if otel_running { - for q in OTEL_QUEUE_RUNNING_COUNT_TAGS.read().await.iter() { - if queue_running_counts.get(q).is_none() { - otel_set_queue_running_count(q, 0); - } - } - } - - #[allow(unused_mut, unused_variables)] - let mut running_tags_to_watch: Vec = vec![]; - #[allow(unused_mut, unused_variables)] - let mut otel_running_tags_to_watch: Vec = vec![]; - for (tag, count) in &queue_running_counts { - #[cfg(feature = "prometheus")] - if metrics_enabled { - let metric = (*QUEUE_RUNNING_COUNT).with_label_values(&[tag]); - metric.set(*count as i64); - running_tags_to_watch.push(tag.to_string()); - } - - if otel_running { - otel_set_queue_running_count(tag, *count as i64); - otel_running_tags_to_watch.push(tag.to_string()); - } - } - - #[cfg(feature = "prometheus")] - if metrics_enabled { - let mut w = QUEUE_RUNNING_COUNT_TAGS.write().await; - *w = running_tags_to_watch; - } - if otel_running { - let mut w = OTEL_QUEUE_RUNNING_COUNT_TAGS.write().await; - *w = otel_running_tags_to_watch; - } + if should_store(prefix, last, current) { + values.push(match (stat, prefix) { + (None, _) => serde_json::json!(0), + (Some(stat), QUEUE_COUNT_PREFIX) => serde_json::json!(stat.count), + (Some(stat), _) => serde_json::json!(stat.delay), + }); + ids.push(row.id); } } + if ids.is_empty() { + return; + } + if let Err(e) = sqlx::query!( + "INSERT INTO metrics (id, value) SELECT * FROM unnest($1::text[], $2::jsonb[])", + &ids[..], + &values[..], + ) + .execute(db) + .await + { + tracing::error!("Failed to save queue metrics: {e:#}"); + } +} + +/// Whether a reading deserves a row of its own, given the last one stored for that metric: +/// its value and how many seconds ago it was written. `current` is `None` once the tag has +/// no backlog left. +fn should_store(prefix: &str, last: Option<(f64, f64)>, current: Option) -> bool { + let Some((last_value, age)) = last else { + // Nothing comparable within the lookback window: a tag that just backed up needs a + // first sample, one that was already gone needs nothing. + return current.is_some(); + }; + let Some(current) = current else { + // The tag drained. One zero pins where the line drops; after that the metric matches + // and goes quiet, then falls out of the lookback window entirely. + return last_value != 0.0; + }; + if age >= QUEUE_METRIC_HEARTBEAT_SECS { + return true; + } + age >= QUEUE_METRIC_MIN_INTERVAL_SECS + && if prefix == QUEUE_COUNT_PREFIX { + last_value != current + } else { + (current - last_value).abs() > last_value.abs() * QUEUE_DELAY_TOLERANCE + } +} + +#[cfg(test)] +mod queue_metric_sampling { + use super::*; + + const RECENT: f64 = QUEUE_METRIC_MIN_INTERVAL_SECS + 1.0; + + #[test] + fn a_holding_backlog_writes_only_on_the_heartbeat() { + let held = Some((3.0, RECENT)); + assert!(!should_store(QUEUE_COUNT_PREFIX, held, Some(3.0))); + assert!(should_store( + QUEUE_COUNT_PREFIX, + Some((3.0, QUEUE_METRIC_HEARTBEAT_SECS)), + Some(3.0) + )); + // Delay climbs on its own, so only a move past the tolerance counts as a change. + assert!(!should_store( + QUEUE_DELAY_PREFIX, + Some((100.0, RECENT)), + Some(105.0) + )); + assert!(should_store( + QUEUE_DELAY_PREFIX, + Some((100.0, RECENT)), + Some(120.0) + )); + } + + #[test] + fn a_drained_tag_writes_one_zero_then_stops() { + assert!(should_store(QUEUE_COUNT_PREFIX, Some((3.0, RECENT)), None)); + assert!(!should_store(QUEUE_COUNT_PREFIX, Some((0.0, RECENT)), None)); + // Including once the heartbeat is due: a tag that is gone stays silent. + assert!(!should_store( + QUEUE_COUNT_PREFIX, + Some((0.0, QUEUE_METRIC_LOOKBACK_SECS)), + None + )); + assert!(!should_store(QUEUE_COUNT_PREFIX, None, None)); + } + + #[test] + fn a_change_waits_for_the_minimum_interval() { + assert!(!should_store( + QUEUE_COUNT_PREFIX, + Some((3.0, 1.0)), + Some(9.0) + )); + assert!(should_store( + QUEUE_COUNT_PREFIX, + Some((3.0, RECENT)), + Some(9.0) + )); + // A tag that has just backed up is recorded at once. + assert!(should_store(QUEUE_COUNT_PREFIX, None, Some(9.0))); + } +} + +/// When this server last sampled the queue into `metrics`, in Unix milliseconds. It only paces +/// how often the queue is scanned for that; whether a sample earns a row is decided from what +/// is already stored. Servers sampling in the same instant can each write it, and the duplicate +/// draws the same. +static LAST_QUEUE_SAMPLE_MS: std::sync::atomic::AtomicI64 = std::sync::atomic::AtomicI64::new(0); + +pub async fn expose_queue_metrics(db: &Pool) { // clean queue metrics older than 14 days sqlx::query!( "DELETE FROM metrics WHERE id LIKE 'queue_%' AND created_at < NOW() - INTERVAL '14 day'" @@ -5262,6 +5318,131 @@ pub async fn expose_queue_metrics(db: &Pool) { .execute(db) .await .ok(); + + let metrics_enabled = METRICS_ENABLED.load(std::sync::atomic::Ordering::Relaxed); + let otel_enabled = OTEL_METRICS_ENABLED.load(Ordering::Relaxed); + let now_ms = chrono::Utc::now().timestamp_millis(); + let save_metrics = now_ms - LAST_QUEUE_SAMPLE_MS.load(Ordering::Relaxed) + >= (QUEUE_METRIC_MIN_INTERVAL_SECS * 1000.0) as i64; + if !(metrics_enabled || otel_enabled || save_metrics) { + return; + } + + // Single DB query for running counts, shared by Prometheus and OTel. It runs ahead of the + // backlog read below, which gives up on the rest of the round when it fails. + let otel_running = otel_enabled; + #[cfg(feature = "prometheus")] + let need_running_counts = metrics_enabled || otel_running; + #[cfg(not(feature = "prometheus"))] + let need_running_counts = otel_running; + + if need_running_counts { + let queue_running_counts = windmill_common::queue::get_queue_running_counts(db).await; + + #[cfg(feature = "prometheus")] + if metrics_enabled { + for q in QUEUE_RUNNING_COUNT_TAGS.read().await.iter() { + if queue_running_counts.get(q).is_none() { + (*QUEUE_RUNNING_COUNT).with_label_values(&[q]).set(0); + } + } + } + + if otel_running { + for q in OTEL_QUEUE_RUNNING_COUNT_TAGS.read().await.iter() { + if queue_running_counts.get(q).is_none() { + otel_set_queue_running_count(q, 0); + } + } + } + + #[allow(unused_mut, unused_variables)] + let mut running_tags_to_watch: Vec = vec![]; + #[allow(unused_mut, unused_variables)] + let mut otel_running_tags_to_watch: Vec = vec![]; + for (tag, count) in &queue_running_counts { + #[cfg(feature = "prometheus")] + if metrics_enabled { + let metric = (*QUEUE_RUNNING_COUNT).with_label_values(&[tag]); + metric.set(*count as i64); + running_tags_to_watch.push(tag.to_string()); + } + + if otel_running { + otel_set_queue_running_count(tag, *count as i64); + otel_running_tags_to_watch.push(tag.to_string()); + } + } + + #[cfg(feature = "prometheus")] + if metrics_enabled { + let mut w = QUEUE_RUNNING_COUNT_TAGS.write().await; + *w = running_tags_to_watch; + } + if otel_running { + let mut w = OTEL_QUEUE_RUNNING_COUNT_TAGS.write().await; + *w = otel_running_tags_to_watch; + } + } + + let queue_stats = match windmill_common::queue::get_queue_stats(db).await { + Ok(queue_stats) => queue_stats, + Err(e) => { + tracing::error!("Failed to read queue stats: {e:#}"); + return; + } + }; + + #[cfg(feature = "prometheus")] + if metrics_enabled { + for q in QUEUE_COUNT_TAGS.read().await.iter() { + if queue_stats.get(q).is_none() { + (*QUEUE_COUNT).with_label_values(&[q]).set(0); + } + } + } + + if otel_enabled { + for q in OTEL_QUEUE_COUNT_TAGS.read().await.iter() { + if queue_stats.get(q).is_none() { + otel_set_queue_count(q, 0); + } + } + } + + #[allow(unused_mut)] + let mut tags_to_watch = vec![]; + #[allow(unused_mut)] + let mut otel_tags_to_watch = vec![]; + for (tag, stat) in queue_stats.iter() { + let count = stat.count; + + #[cfg(feature = "prometheus")] + if metrics_enabled { + let metric = (*QUEUE_COUNT).with_label_values(&[tag]); + metric.set(count as i64); + tags_to_watch.push(tag.to_string()); + } + + if otel_enabled { + otel_tags_to_watch.push(tag.to_string()); + } + otel_set_queue_count(tag, count as i64); + } + + if save_metrics { + LAST_QUEUE_SAMPLE_MS.store(now_ms, Ordering::Relaxed); + save_queue_metrics(db, &queue_stats).await; + } + + if metrics_enabled { + let mut w = QUEUE_COUNT_TAGS.write().await; + *w = tags_to_watch; + } + if otel_enabled { + let mut w = OTEL_QUEUE_COUNT_TAGS.write().await; + *w = otel_tags_to_watch; + } } pub async fn reload_smtp_config(db: &Pool) { diff --git a/backend/windmill-common/src/queue.rs b/backend/windmill-common/src/queue.rs index 2656ce5af5..7166b3d3cd 100644 --- a/backend/windmill-common/src/queue.rs +++ b/backend/windmill-common/src/queue.rs @@ -15,6 +15,46 @@ pub async fn get_queue_counts(db: &Pool) -> HashMap { .unwrap_or_else(|| HashMap::new()) } +/// Backlog of a single tag: jobs waiting more than 3 seconds past their `scheduled_for`. +pub struct QueueStat { + pub count: u32, + /// How long the job that would be picked up next has already been waiting, in seconds. + pub delay: f64, +} + +/// Same backlog as [`get_queue_counts`], plus the delay of the job at the head of each +/// tag's queue. The head is picked with the same ordering the worker pull uses, so the +/// delay reported is the one a worker is about to observe. +/// +/// Reads the queue of every workspace: a caller exposing the result MUST restrict it to +/// devops users, as `GET /workers/queue_counts` does. Unlike [`get_queue_counts`], a failed +/// read is an error rather than an empty map, which would read as every backlog draining. +pub async fn get_queue_stats( + db: &Pool, +) -> crate::error::Result> { + // Grouping by (tag, priority) first finds every head in the same single pass as the + // count. A per-tag `ORDER BY ... LIMIT 1` walks `queue_sort_v2`, whose `tag` column comes + // last, through every other tag's backlog queued ahead of it. + let rows = sqlx::query!( + "SELECT tag AS \"tag!\", sum(n)::bigint AS \"count!\", + EXTRACT(EPOCH FROM now() - (array_agg(head ORDER BY priority DESC NULLS LAST))[1]) + ::double precision AS \"delay!\" + FROM ( + SELECT tag, priority, count(*) AS n, min(scheduled_for) AS head + FROM v2_job_queue WHERE + scheduled_for <= now() - ('3 seconds')::interval AND running = false + GROUP BY tag, priority + ) g + GROUP BY tag", + ) + .fetch_all(db) + .await?; + Ok(rows + .into_iter() + .map(|x| (x.tag, QueueStat { count: x.count as u32, delay: x.delay })) + .collect()) +} + pub async fn get_queue_running_counts(db: &Pool) -> HashMap { sqlx::query!( "SELECT tag AS \"tag!\", count(*) AS \"count!\" FROM v2_job_queue WHERE diff --git a/backend/windmill-common/tests/queue_stats.rs b/backend/windmill-common/tests/queue_stats.rs new file mode 100644 index 0000000000..2b7a80a59d --- /dev/null +++ b/backend/windmill-common/tests/queue_stats.rs @@ -0,0 +1,63 @@ +use sqlx::{Pool, Postgres}; +use windmill_common::queue::get_queue_stats; + +const WORKSPACE: &str = "test-workspace"; + +async fn queue_job( + db: &Pool, + tag: &str, + priority: Option, + waited_secs: f64, + running: bool, +) { + sqlx::query( + "WITH job AS ( + INSERT INTO v2_job (id, workspace_id, tag) VALUES (gen_random_uuid(), $1, $2) + RETURNING id + ) + INSERT INTO v2_job_queue (id, workspace_id, tag, priority, running, scheduled_for) + SELECT id, $1, $2, $3, $4, now() - make_interval(secs => $5) FROM job", + ) + .bind(WORKSPACE) + .bind(tag) + .bind(priority) + .bind(running) + .bind(waited_secs) + .execute(db) + .await + .expect("failed to queue job"); +} + +/// The delay reported for a tag is that of the job the worker pull takes first, ordered +/// `priority DESC NULLS LAST, scheduled_for`, not simply the oldest one waiting. Running jobs +/// and jobs less than 3 seconds past due are not part of the backlog at all. +#[sqlx::test(migrations = "../migrations", fixtures("base"))] +async fn queue_stats_report_the_delay_of_the_job_pulled_next(db: Pool) { + // The oldest job has no priority, so every prioritized job runs before it. + queue_job(&db, "mixed", None, 900.0, false).await; + queue_job(&db, "mixed", Some(1), 600.0, false).await; + queue_job(&db, "mixed", Some(5), 300.0, false).await; + queue_job(&db, "mixed", Some(5), 100.0, false).await; + // Highest priority, but not backlog: already running, or not yet 3 seconds past due. + queue_job(&db, "mixed", Some(9), 1200.0, true).await; + queue_job(&db, "mixed", Some(9), 1.0, false).await; + queue_job(&db, "unprioritized", None, 500.0, false).await; + queue_job(&db, "unprioritized", None, 50.0, false).await; + + let stats = get_queue_stats(&db).await.unwrap(); + + let mixed = &stats["mixed"]; + assert_eq!(mixed.count, 4); + assert!( + (mixed.delay - 300.0).abs() < 5.0, + "expected the oldest job of the highest priority, got a delay of {}", + mixed.delay + ); + let unprioritized = &stats["unprioritized"]; + assert_eq!(unprioritized.count, 2); + assert!( + (unprioritized.delay - 500.0).abs() < 5.0, + "expected the oldest job, got a delay of {}", + unprioritized.delay + ); +} diff --git a/frontend/src/lib/components/QueueMetricsDrawerInner.svelte b/frontend/src/lib/components/QueueMetricsDrawerInner.svelte index 3b25513b4d..d9df9f57c2 100644 --- a/frontend/src/lib/components/QueueMetricsDrawerInner.svelte +++ b/frontend/src/lib/components/QueueMetricsDrawerInner.svelte @@ -1,5 +1,5 @@