starred/windmill
1
0
Fork 0
mirror of https://github.com/windmill-labs/windmill.git synced 2026-09-11 16:09:39 +00:00
Code Issues Packages Projects Releases Wiki Activity

perf: only write queue metrics when a tag's backlog changes (#11055)

Browse Source 
* perf: only write queue metrics when a tag's backlog changes

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
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) <noreply@anthropic.com>
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) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01Vz5TLoq492nruNSCr6LARA

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-09-10 16:09:19 +02:00
committed by GitHub
co-authored by Claude Opus 5
parent ab9efc897c
commit 9d75929247
10 changed files with 564 additions and 238 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-00e63eab76d26e148b77e932848de74e8b0943d30481465da453942e299a128f.json
Generated
-15
View File
@@ -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"
}
backend/.sqlx/query-38e12be0764d850538d3eea328aebfaa3e7ec3a7c682945f22e8cde3a63094f4.json
Generated
+15
View File
@@ -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"
}
backend/.sqlx/query-82f6674f19e8ad51a992505a46f46fc4a48172f104e9e849f755ac041c3eef92.json
Generated
-20
View File
@@ -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"
}
backend/.sqlx/query-8824b382c4e98dfa17b4aa656af3a6c1ff99973e778d71bd598a50d022da8f15.json
Generated
-15
View File
@@ -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"
}
backend/.sqlx/query-bd46b7a19f9e51de1babc3d16fd7d7b7776b90a96c6662b00d8a999b74eb5b36.json
Generated
+35
View File
@@ -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"
}
backend/.sqlx/query-d09d3a6223262ac584f045feec7e08ea1695e41d9b11538b58189925ddba4fdd.json
Generated
+32
View File
@@ -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"
}
backend/src/monitor.rs
+320 -139
View File
@@ -5106,155 +5106,211 @@ async fn vacuuming_tables(db: &Pool<Postgres>) -> error::Result<()> {
Ok(())
}
pub async fn expose_queue_metrics(db: &Pool<Postgres>) {
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<Postgres>,
queue_stats: &std::collections::HashMap<String, windmill_common::queue::QueueStat>,
) {
let sampled_ids = queue_stats
.keys()
.flat_map(|tag| {
[
format!("{QUEUE_COUNT_PREFIX}{tag}"),
format!("{QUEUE_DELAY_PREFIX}{tag}"),
]
})
.collect::<Vec<_>>();
#[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<String> = vec![];
#[allow(unused_mut, unused_variables)]
let mut otel_running_tags_to_watch: Vec<String> = 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<f64>) -> 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<Postgres>) {
// 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<Postgres>) {
.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<String> = vec![];
#[allow(unused_mut, unused_variables)]
let mut otel_running_tags_to_watch: Vec<String> = 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<Postgres>) {
backend/windmill-common/src/queue.rs
+40
View File
@@ -15,6 +15,46 @@ pub async fn get_queue_counts(db: &Pool<Postgres>) -> HashMap<String, u32> {
.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<Postgres>,
) -> crate::error::Result<HashMap<String, QueueStat>> {
// 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<Postgres>) -> HashMap<String, u32> {
sqlx::query!(
"SELECT tag AS \"tag!\", count(*) AS \"count!\" FROM v2_job_queue WHERE
backend/windmill-common/tests/queue_stats.rs
+63
View File
@@ -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<Postgres>,
tag: &str,
priority: Option<i16>,
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<Postgres>) {
// 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
);
}
frontend/src/lib/components/QueueMetricsDrawerInner.svelte
+59 -49
View File
@@ -1,5 +1,5 @@
<script lang="ts">
import { run } from 'svelte/legacy';
import { run } from 'svelte/legacy'
import 'chartjs-adapter-date-fns'
import { Line } from '$lib/components/chartjs-wrappers/chartJs'
@@ -73,51 +73,64 @@
let noMetrics = $state(false)
function fillData(
// The sampler only records a queue metric when its value moves, plus a heartbeat while a
// tag stays backlogged and a closing zero once it drains, so a gap means "unchanged". A
// series silent well past the heartbeat (which lands a monitor tick or so late) never got
// its closing zero, because no server was up when the tag drained: it reads as zero from there.
const HEARTBEAT_MS = 5 * 60 * 1000 // QUEUE_METRIC_HEARTBEAT_SECS in backend/src/monitor.rs
const STALE_AFTER_MS = 3 * HEARTBEAT_MS
// Hold each value until the next point. `'after'` starts a value at the previous point
// instead, which draws a whole backlog at the height of the zero that closes it.
const STEPPED = 'before' as const
function toPoints(
data: {
created_at: string
value: number
}[],
zero = 0
) {
// fill holes with 0
const sorted: typeof data = []
for (const el of [
...data,
{
created_at: new Date().toISOString(),
value: zero
tolerance: number
): Point[] {
const points: Point[] = []
let lastSampleTs: number | undefined
let lastValue: number | undefined
function push(x: number, y: number) {
points.push({ x, y })
lastValue = y
}
function bridge(until: number) {
if (lastSampleTs != undefined && until - lastSampleTs > STALE_AFTER_MS && lastValue !== 0) {
push(lastSampleTs + STALE_AFTER_MS, 0)
}
]) {
const last =
sorted.length > 0 ? new Date(sorted[sorted.length - 1].created_at).getTime() : undefined
const currentTs = new Date(el.created_at).getTime()
if (last && currentTs - last > 1000 * 60 * 2) {
const numElements = Math.floor((currentTs - last) / (1000 * 30))
for (let i = 1; i < numElements; i++) {
sorted.push({
created_at: new Date(last + i * (1000 * 30)).toISOString(),
value: zero
})
}
}
sorted.push(el)
}
// remove high frequency data points for similar values
const light: typeof sorted = []
for (const el of sorted) {
const last = light.length > 0 ? light[light.length - 1] : undefined
for (const el of data) {
const ts = new Date(el.created_at).getTime()
bridge(ts)
// Keep only points that move the line: a heartbeat repeats the value in force, which
// stepped drawing renders identically.
if (
!last ||
Math.abs((el.value - last.value) / last.value) > 0.1 ||
new Date(el.created_at).getTime() - new Date(last.created_at).getTime() > 1000 * 60 * 15
lastValue == undefined ||
Math.abs(el.value - lastValue) > Math.abs(lastValue) * tolerance
) {
light.push(el)
push(ts, el.value)
}
lastSampleTs = ts
}
return light
if (lastSampleTs != undefined) {
const now = Date.now()
bridge(now)
// Carry the value in force to the right edge of the chart.
push(now, lastValue!)
}
return points
}
// Delay is drawn on a log scale, which cannot plot 0, so a drained tag is pinned to 1 and
// the tooltip reads it back as 0.
function asLogSafe(points: Point[]): Point[] {
return points.map((p) => ({ x: p.x, y: p.y === 0 ? 1 : p.y }))
}
async function loadMetrics() {
@@ -145,7 +158,8 @@
label: m.id.slice(12),
backgroundColor: bgColor,
borderColor: color,
data: fillData(m.values).map((v) => ({ x: v.created_at as any, y: v.value }))
stepped: STEPPED,
data: toPoints(m.values, 0)
}
})
}
@@ -159,22 +173,18 @@
label: m.id.slice(12),
borderColor: color,
backgroundColor: bgColor,
data: fillData(m.values, 1).map((v) => ({
x: v.created_at as any,
y: v.value
}))
stepped: STEPPED,
// Delay climbs on its own while a tag stays backlogged; the sampler stores a
// new row only once it has moved by 10%, so hold the chart to the same step.
data: asLogSafe(toPoints(m.values, 0.1))
}
})
}
minDate = new Date(
Math.min(
...countData.datasets
.map((x) => x.data[0].x)
.filter((x) => x != null)
.map((d) => new Date(d).getTime())
)
)
const firstTs = [...countData.datasets, ...delayData.datasets]
.map((d) => d.data[0]?.x)
.filter((x) => x != undefined)
minDate = firstTs.length > 0 ? new Date(Math.min(...firstTs)) : new Date()
loading = false
}
@@ -185,10 +195,10 @@
run(() => {
ChartJS.defaults.color = darkMode ? '#ccc' : '#666'
});
})
run(() => {
ChartJS.defaults.borderColor = darkMode ? 'rgba(255, 255, 255, 0.1)' : 'rgba(0, 0, 0, 0.1)'
});
})
</script>
<DarkModeObserver bind:darkMode />