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

feat: live queue status per tag and bounded queue metric charts (#11067)

Browse Source 
* feat: live per-tag queue status and bounded charts in the queues drawer

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011fYMqh7R8FnePmbYzirJXZ

* fix: drop stale chart failures and test the queue metrics series query

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011fYMqh7R8FnePmbYzirJXZ

* fix: name the queue status refresh, skip overlapping polls, soften the no-worker warning

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011fYMqh7R8FnePmbYzirJXZ

* feat: draw a stuck tag's queue delay exactly as it climbs (#11071)

* feat: store a stuck tag's queue delay as its head's wait start

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011fYMqh7R8FnePmbYzirJXZ

* fix: keep a climb's top inside a slot and stamp held delays exactly

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011fYMqh7R8FnePmbYzirJXZ

* fix: redraw a climb as soon as its head leaves, and document the lookup slack

Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_011fYMqh7R8FnePmbYzirJXZ

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 5 (1M context) <noreply@anthropic.com>
This commit is contained in:
Ruben Fiszel
2026-09-10 18:40:54 +00:00
committed by GitHub
co-authored by Claude Opus 5
parent 08d876aebf
commit 569adb85c1
22 changed files with 1459 additions and 384 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-2742245bc03290120a97b21c441cb56825e9fd552a7aeddfb8a372540c19b863.json
Generated
+73
View File
@@ -0,0 +1,73 @@
{
"db_name": "PostgreSQL",
"query": "WITH slots AS (\n SELECT id, slot, min(t) AS first, max(t) AS last, max(v) AS peak,\n (min(ARRAY[t, v]))[2] AS first_value, (max(ARRAY[t, v]))[2] AS last_value,\n (max(ARRAY[t, climbing]))[2] = 1 AS last_climbing,\n COALESCE(bool_and(climbing = 1) AND max(since) - min(since) < $4, false) AS ramp,\n max(ARRAY[t, since]) FILTER (WHERE climbing = 1) AS last_climb\n FROM (\n SELECT id, t,\n CASE jsonb_typeof(value)\n WHEN 'number' THEN value::double precision\n WHEN 'object' THEN t - (value->>'since')::double precision\n END AS v,\n (value->>'since')::double precision AS since,\n (jsonb_typeof(value) = 'object')::int::double precision AS climbing,\n greatest(floor((t - $1::double precision) / $2::double precision), -1)::int\n AS slot\n FROM (\n SELECT id, value, EXTRACT(EPOCH FROM created_at)::double precision AS t\n FROM metrics\n WHERE id LIKE 'queue_%'\n AND created_at > to_timestamp($1::double precision - $3::double precision)\n ) m\n ) s\n WHERE v IS NOT NULL\n GROUP BY id, slot\n )\n SELECT id AS \"id!\", slot AS \"slot!\", first AS \"first!\", last AS \"last!\",\n greatest(peak, CASE WHEN last_climb[1] < last THEN (\n SELECT EXTRACT(EPOCH FROM min(n.created_at))::double precision\n FROM metrics n\n WHERE n.id = slots.id AND n.id LIKE 'queue_%'\n AND n.created_at > to_timestamp(last_climb[1] + 0.001)\n AND n.created_at <= to_timestamp(last + 0.001)\n ) - last_climb[2] END) AS \"peak!\",\n first_value AS \"first_value!\", last_value AS \"last_value!\",\n last_climbing AS \"last_climbing!\", ramp AS \"ramp!\"\n FROM slots\n ORDER BY id, slot",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id!",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "slot!",
"type_info": "Int4"
},
{
"ordinal": 2,
"name": "first!",
"type_info": "Float8"
},
{
"ordinal": 3,
"name": "last!",
"type_info": "Float8"
},
{
"ordinal": 4,
"name": "peak!",
"type_info": "Float8"
},
{
"ordinal": 5,
"name": "first_value!",
"type_info": "Float8"
},
{
"ordinal": 6,
"name": "last_value!",
"type_info": "Float8"
},
{
"ordinal": 7,
"name": "last_climbing!",
"type_info": "Bool"
},
{
"ordinal": 8,
"name": "ramp!",
"type_info": "Bool"
}
],
"parameters": {
"Left": [
"Float8",
"Float8",
"Float8",
"Float8"
]
},
"nullable": [
false,
null,
null,
null,
null,
null,
null,
null,
null
]
},
"hash": "2742245bc03290120a97b21c441cb56825e9fd552a7aeddfb8a372540c19b863"
}
backend/.sqlx/query-38e12be0764d850538d3eea328aebfaa3e7ec3a7c682945f22e8cde3a63094f4.json
Generated
-15
View File
@@ -1,15 +0,0 @@
{
"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-3bd816e986ef2d2a193e51c985b61c04b71f464021e4b884bf21cc7c25f6a753.json
Generated
+20
View File
@@ -0,0 +1,20 @@
{
"db_name": "PostgreSQL",
"query": "SELECT EXTRACT(EPOCH FROM now())::double precision AS \"now!\"",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "now!",
"type_info": "Float8"
}
],
"parameters": {
"Left": []
},
"nullable": [
null
]
},
"hash": "3bd816e986ef2d2a193e51c985b61c04b71f464021e4b884bf21cc7c25f6a753"
}
backend/.sqlx/query-44dd7a66ecc9564ad5727970b5f60a1717eda8924999741be522ffb74cc173fa.json
Generated
-26
View File
@@ -1,26 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "WITH queue_metrics as (\n SELECT id, value, created_at\n FROM metrics\n WHERE id LIKE 'queue_%'\n AND created_at > now() - interval '14 day'\n )\n SELECT id, array_agg(json_build_object('value', value, 'created_at', created_at) ORDER BY created_at ASC) as \"values!\"\n FROM queue_metrics\n GROUP BY id\n ORDER BY id ASC",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "values!",
"type_info": "JsonArray"
}
],
"parameters": {
"Left": []
},
"nullable": [
false,
null
]
},
"hash": "44dd7a66ecc9564ad5727970b5f60a1717eda8924999741be522ffb74cc173fa"
}
backend/.sqlx/query-58f99e0d5877f403cde04459e4a425efd77aa4c0118f62ada2caa0794b0d738b.json
Generated
+41
View File
@@ -0,0 +1,41 @@
{
"db_name": "PostgreSQL",
"query": "SELECT COALESCE(c.id, r.id) AS \"id!\", r.value AS \"value?\",\n EXTRACT(EPOCH FROM r.created_at)::double precision AS \"at?\",\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": "at?",
"type_info": "Float8"
},
{
"ordinal": 3,
"name": "age?",
"type_info": "Float8"
}
],
"parameters": {
"Left": [
"TextArray",
"Float8"
]
},
"nullable": [
true,
true,
true,
true
]
},
"hash": "58f99e0d5877f403cde04459e4a425efd77aa4c0118f62ada2caa0794b0d738b"
}
backend/.sqlx/query-7745127eb4a4be2b67427a708e8e5bf2973af84cf315047e586071438cd5e438.json
Generated
+34
View File
@@ -0,0 +1,34 @@
{
"db_name": "PostgreSQL",
"query": "WITH running AS (\n SELECT tag, count(*) AS n FROM v2_job_queue WHERE running = true GROUP BY tag\n )\n SELECT t.tag AS \"tag!\", COALESCE(r.n, 0) AS \"running!\",\n (SELECT count(*) FROM worker_ping w\n WHERE w.ping_at > now() - interval '1 minute' AND w.custom_tags @> ARRAY[t.tag]\n ) AS \"workers!\"\n FROM (SELECT tag::text FROM running UNION SELECT unnest($1::text[])) t(tag)\n LEFT JOIN running r ON r.tag = t.tag\n ORDER BY t.tag",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "tag!",
"type_info": "Text"
},
{
"ordinal": 1,
"name": "running!",
"type_info": "Int8"
},
{
"ordinal": 2,
"name": "workers!",
"type_info": "Int8"
}
],
"parameters": {
"Left": [
"TextArray"
]
},
"nullable": [
null,
null,
null
]
},
"hash": "7745127eb4a4be2b67427a708e8e5bf2973af84cf315047e586071438cd5e438"
}
backend/.sqlx/query-7af0fd3d8dd1d949ce11b190a4fa6b56c84904aeed791b2a0879b36add6bf9b5.json
Generated
+38
View File
@@ -0,0 +1,38 @@
{
"db_name": "PostgreSQL",
"query": "SELECT tag AS \"tag!\", count AS \"count!\",\n EXTRACT(EPOCH FROM now() - head)::double precision AS \"delay!\",\n EXTRACT(EPOCH FROM head)::double precision AS \"head_since!\"\n FROM (\n SELECT tag, sum(n)::bigint AS count,\n (array_agg(head ORDER BY priority DESC NULLS LAST))[1] AS head\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\n ) t",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "tag!",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "count!",
"type_info": "Int8"
},
{
"ordinal": 2,
"name": "delay!",
"type_info": "Float8"
},
{
"ordinal": 3,
"name": "head_since!",
"type_info": "Float8"
}
],
"parameters": {
"Left": []
},
"nullable": [
false,
null,
null,
null
]
},
"hash": "7af0fd3d8dd1d949ce11b190a4fa6b56c84904aeed791b2a0879b36add6bf9b5"
}
backend/.sqlx/query-a498f752e169b711c0cb26ac167b2e554a7fdfc1055389fd65f330469e329d16.json
Generated
+26
View File
@@ -0,0 +1,26 @@
{
"db_name": "PostgreSQL",
"query": "WITH queue_metrics as (\n SELECT id, created_at,\n CASE WHEN jsonb_typeof(value) = 'object'\n THEN to_jsonb(EXTRACT(EPOCH FROM created_at) - (value->>'since')::numeric)\n ELSE value\n END AS value\n FROM metrics\n WHERE id LIKE 'queue_%'\n AND created_at > now() - interval '14 day'\n )\n SELECT id, array_agg(json_build_object('value', value, 'created_at', created_at) ORDER BY created_at ASC) as \"values!\"\n FROM queue_metrics\n GROUP BY id\n ORDER BY id ASC",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "id",
"type_info": "Varchar"
},
{
"ordinal": 1,
"name": "values!",
"type_info": "JsonArray"
}
],
"parameters": {
"Left": []
},
"nullable": [
false,
null
]
},
"hash": "a498f752e169b711c0cb26ac167b2e554a7fdfc1055389fd65f330469e329d16"
}
backend/.sqlx/query-bd46b7a19f9e51de1babc3d16fd7d7b7776b90a96c6662b00d8a999b74eb5b36.json
Generated
-35
View File
@@ -1,35 +0,0 @@
{
"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
@@ -1,32 +0,0 @@
{
"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/.sqlx/query-d4ce900b8e60b530c2ea57c082edbc8c12c71869c4e45219dad4fb7198aaa1e3.json
Generated
+16
View File
@@ -0,0 +1,16 @@
{
"db_name": "PostgreSQL",
"query": "INSERT INTO metrics (id, value)\n SELECT id, COALESCE(to_jsonb(EXTRACT(EPOCH FROM now())::double precision - held_head), value)\n FROM unnest($1::text[], $2::jsonb[], $3::double precision[]) AS u(id, value, held_head)",
"describe": {
"columns": [],
"parameters": {
"Left": [
"TextArray",
"JsonbArray",
"Float8Array"
]
},
"nullable": []
},
"hash": "d4ce900b8e60b530c2ea57c082edbc8c12c71869c4e45219dad4fb7198aaa1e3"
}
backend/src/monitor.rs
+162 -66
View File
@@ -109,6 +109,10 @@ use windmill_common::{
APP_WORKSPACED_ROUTE_SETTING, HTTP_ROUTE_WORKSPACED_ROUTE,
HTTP_ROUTE_WORKSPACED_ROUTE_SETTING,
},
queue_metrics::{
QueueSample, QUEUE_COUNT_PREFIX, QUEUE_DELAY_PREFIX, QUEUE_DELAY_SAME_HEAD_SECS,
QUEUE_METRIC_HEARTBEAT_SECS, QUEUE_METRIC_STALE_SECS,
},
};
#[cfg(feature = "parquet")]
use windmill_object_store::reload_object_store_setting;
@@ -5110,31 +5114,20 @@ async fn vacuuming_tables(db: &Pool<Postgres>) -> error::Result<()> {
/// 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.
/// A held delay hovers while the head keeps changing, so an exact-value comparison would rarely
/// dedup it. Only a move the chart would actually render is stored.
const QUEUE_DELAY_TOLERANCE: f64 = 0.1;
const QUEUE_COUNT_PREFIX: &str = "queue_count_";
const QUEUE_DELAY_PREFIX: &str = "queue_delay_";
/// Append the queue metrics the drawer at `GET /workers/queue_metrics` charts, skipping any
/// sample that repeats what is already stored.
/// Append the queue metrics the drawer at `GET /workers/queue_metrics_series` 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
/// the whole 14-day retention. Each metric is written when the value it draws 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.
/// mean "unchanged since the last row", which is what the chart interpolates. A delay whose
/// head job stays put is stored as that job's wait start, which the chart draws climbing, so it
/// never moves away from what is stored either.
async fn save_queue_metrics(
db: &Pool<Postgres>,
queue_stats: &std::collections::HashMap<String, windmill_common::queue::QueueStat>,
@@ -5150,11 +5143,12 @@ async fn save_queue_metrics(
.collect::<Vec<_>>();
// 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.
// recently enough to still be believed backlogged. Bounding the lookup by the stale 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 r.created_at)::double precision AS \"at?\",
EXTRACT(EPOCH FROM now() - r.created_at)::double precision AS \"age?\"
FROM unnest($1::text[]) AS c(id)
FULL JOIN (
@@ -5164,7 +5158,7 @@ async fn save_queue_metrics(
ORDER BY id, created_at DESC
) r ON r.id = c.id",
&sampled_ids[..],
QUEUE_METRIC_LOOKBACK_SECS,
QUEUE_METRIC_STALE_SECS,
)
.fetch_all(db)
.await
@@ -5178,6 +5172,8 @@ async fn save_queue_metrics(
let mut ids = vec![];
let mut values = vec![];
// The wait start of the head of each held delay, whose value the INSERT computes.
let mut held_heads: Vec<Option<f64>> = vec![];
for row in last_samples {
let Some((prefix, tag)) = [QUEUE_COUNT_PREFIX, QUEUE_DELAY_PREFIX]
.into_iter()
@@ -5185,8 +5181,14 @@ async fn save_queue_metrics(
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);
// A stored value that cannot be read cannot be compared, so the next reading is kept.
let last = row
.value
.as_ref()
.and_then(QueueSample::parse)
.zip(row.at)
.zip(row.age)
.map(|((sample, at), age)| (sample, at, age));
let stat = queue_stats.get(tag);
let current = stat.map(|stat| {
if prefix == QUEUE_COUNT_PREFIX {
@@ -5196,23 +5198,42 @@ async fn save_queue_metrics(
}
});
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),
});
let next_delay = stat
.filter(|_| prefix == QUEUE_DELAY_PREFIX)
.map(|stat| delay_sample(last.map(|(sample, at, _)| sample.head_since(at)), stat));
let drawn_now = last.map(|(sample, at, age)| (sample.value_at(at + age), age));
let redraws = last
.zip(next_delay)
.is_some_and(|((sample, ..), next)| redraws(sample, next));
if should_store(prefix, drawn_now, current, redraws) {
let (value, held_head) = match (stat, next_delay) {
(None, _) => (serde_json::json!(0), None),
(Some(stat), None) => (serde_json::json!(stat.count), None),
(Some(stat), Some(QueueSample::Held(_))) => {
(serde_json::Value::Null, Some(stat.head_since))
}
(Some(_), Some(climbing)) => (climbing.to_json(), None),
};
ids.push(row.id);
values.push(value);
held_heads.push(held_head);
}
}
if ids.is_empty() {
return;
}
// A held delay is computed from this statement's `now()`, the row's `created_at` too, so
// `created_at - value` is exactly its head's wait start. That is how the next sample tells
// whether the same job is still at the head, within `QUEUE_DELAY_SAME_HEAD_SECS`, which the
// time between reading the queue and this INSERT could otherwise exceed on a busy database.
if let Err(e) = sqlx::query!(
"INSERT INTO metrics (id, value) SELECT * FROM unnest($1::text[], $2::jsonb[])",
"INSERT INTO metrics (id, value)
SELECT id, COALESCE(to_jsonb(EXTRACT(EPOCH FROM now())::double precision - held_head), value)
FROM unnest($1::text[], $2::jsonb[], $3::double precision[]) AS u(id, value, held_head)",
&ids[..],
&values[..],
&held_heads[..] as &[Option<f64>],
)
.execute(db)
.await
@@ -5221,10 +5242,38 @@ async fn save_queue_metrics(
}
}
/// What to store for a delay reading, given when the head job of the last stored sample started
/// waiting. The same job still at the head keeps the delay climbing from its wait start, which
/// the chart draws exactly. A head that changed means a moving queue, whose delay hovers and is
/// held; so is a first sample, which cannot tell yet and must not draw a climb that never was.
fn delay_sample(
last_head_since: Option<f64>,
stat: &windmill_common::queue::QueueStat,
) -> QueueSample {
match last_head_since {
Some(since) if (since - stat.head_since).abs() < QUEUE_DELAY_SAME_HEAD_SECS => {
QueueSample::Climbing { since: stat.head_since }
}
_ => QueueSample::Held(stat.delay),
}
}
/// Whether the next delay sample is drawn differently from the last one even at the same value:
/// a climb whose head left would otherwise go on climbing from the old head, and a held delay
/// whose head stayed would stay flat while the wait grows.
fn redraws(last: QueueSample, next: QueueSample) -> bool {
matches!(last, QueueSample::Climbing { .. }) != matches!(next, QueueSample::Climbing { .. })
}
/// 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 {
/// the value it draws now and how many seconds ago it was written. `current` is `None` once
/// the tag has no backlog left; `redraws` is set when the reading must be drawn differently.
fn should_store(
prefix: &str,
last: Option<(f64, f64)>,
current: Option<f64>,
redraws: bool,
) -> 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.
@@ -5239,11 +5288,12 @@ fn should_store(prefix: &str, last: Option<(f64, f64)>, current: Option<f64>) ->
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
}
&& (redraws
|| if prefix == QUEUE_COUNT_PREFIX {
last_value != current
} else {
(current - last_value).abs() > last_value.abs() * QUEUE_DELAY_TOLERANCE
})
}
#[cfg(test)]
@@ -5255,36 +5305,33 @@ mod queue_metric_sampling {
#[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)
));
assert!(!should_store(QUEUE_COUNT_PREFIX, held, Some(3.0), false));
let due = Some((3.0, QUEUE_METRIC_HEARTBEAT_SECS));
assert!(should_store(QUEUE_COUNT_PREFIX, due, Some(3.0), false));
// A held delay hovers, so only a move past the tolerance counts as a change.
let delay = Some((100.0, RECENT));
assert!(!should_store(QUEUE_DELAY_PREFIX, delay, Some(105.0), false));
assert!(should_store(QUEUE_DELAY_PREFIX, delay, Some(120.0), false));
}
#[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((3.0, RECENT)),
None,
false
));
assert!(!should_store(
QUEUE_COUNT_PREFIX,
Some((0.0, QUEUE_METRIC_LOOKBACK_SECS)),
None
Some((0.0, RECENT)),
None,
false
));
assert!(!should_store(QUEUE_COUNT_PREFIX, None, None));
// Including once the heartbeat is due: a tag that is gone stays silent.
let gone = Some((0.0, QUEUE_METRIC_STALE_SECS));
assert!(!should_store(QUEUE_COUNT_PREFIX, gone, None, false));
assert!(!should_store(QUEUE_COUNT_PREFIX, None, None, false));
}
#[test]
@@ -5292,15 +5339,64 @@ mod queue_metric_sampling {
assert!(!should_store(
QUEUE_COUNT_PREFIX,
Some((3.0, 1.0)),
Some(9.0)
Some(9.0),
false
));
assert!(should_store(
QUEUE_COUNT_PREFIX,
Some((3.0, RECENT)),
Some(9.0)
Some(9.0),
false
));
// A tag that has just backed up is recorded at once.
assert!(should_store(QUEUE_COUNT_PREFIX, None, Some(9.0)));
assert!(should_store(QUEUE_COUNT_PREFIX, None, Some(9.0), false));
}
#[test]
fn a_delay_climbs_while_the_same_job_stays_at_the_head() {
let stat = windmill_common::queue::QueueStat { count: 3, delay: 330.0, head_since: 1000.0 };
// A held sample written at 1320 saw the same head: it switches to climbing at once,
// although the delay has not moved past the tolerance yet.
let first = QueueSample::Held(320.0);
let climbing = delay_sample(Some(first.head_since(1320.0)), &stat);
assert_eq!(climbing, QueueSample::Climbing { since: 1000.0 });
assert!(redraws(first, climbing));
let drawn = Some((first.value_at(1330.0), RECENT));
assert!(should_store(
QUEUE_DELAY_PREFIX,
drawn,
Some(stat.delay),
true
));
// Stored climbing, it draws the delay exactly: nothing more until the heartbeat.
let drawn = Some((climbing.value_at(1600.0), RECENT));
assert!(!should_store(QUEUE_DELAY_PREFIX, drawn, Some(600.0), false));
assert_eq!(delay_sample(None, &stat), QueueSample::Held(330.0));
}
#[test]
fn a_climb_whose_head_left_is_held_even_within_the_tolerance() {
// The head waiting since 0 left at 3600 for one queued at 100: 3500s is within 10% of
// the 3600s the climb draws, but kept, the climb would go on from the old head.
let moved =
windmill_common::queue::QueueStat { count: 2, delay: 3500.0, head_since: 100.0 };
let climbing = QueueSample::Climbing { since: 0.0 };
let next = delay_sample(Some(climbing.head_since(3000.0)), &moved);
assert_eq!(next, QueueSample::Held(3500.0));
assert!(redraws(climbing, next));
let drawn = Some((climbing.value_at(3600.0), RECENT));
assert!(!should_store(
QUEUE_DELAY_PREFIX,
drawn,
Some(moved.delay),
false
));
assert!(should_store(
QUEUE_DELAY_PREFIX,
drawn,
Some(moved.delay),
true
));
}
}
backend/windmill-api-workers/src/lib.rs
+92 -1
View File
@@ -19,6 +19,7 @@ use windmill_common::{
db::UserDB,
error::JsonResult,
jobs::{HIDE_WORKERS_FOR_NON_ADMINS, TAGS_ARE_SENSITIVE},
queue_metrics::{read_queue_metrics_series, QueueMetricsSeries},
utils::{paginate, Pagination},
worker::{ALL_TAGS, CUSTOM_TAGS_PER_WORKSPACE, DEFAULT_TAGS, DEFAULT_TAGS_PER_WORKSPACE},
workspaces::workspace_with_fork_ancestors,
@@ -38,6 +39,8 @@ pub fn global_service() -> Router {
)
.route("/get_default_tags", get(get_default_tags))
.route("/queue_metrics", get(get_queue_metrics))
.route("/queue_metrics_series", get(get_queue_metrics_series))
.route("/queue_status", get(get_queue_status))
.route("/queue_counts", get(get_queue_counts))
.route("/queue_running_counts", get(get_queue_running_counts))
.route(
@@ -270,10 +273,16 @@ async fn get_queue_metrics(
) -> JsonResult<Vec<QueueMetric>> {
require_devops_role(&db, &authed).await?;
// The API declares every `value` a number, so a climbing delay, stored as its head's wait
// start, is returned as the delay at the time of its sample.
let queue_metrics = sqlx::query_as!(
QueueMetric,
"WITH queue_metrics as (
SELECT id, value, created_at
SELECT id, created_at,
CASE WHEN jsonb_typeof(value) = 'object'
THEN to_jsonb(EXTRACT(EPOCH FROM created_at) - (value->>'since')::numeric)
ELSE value
END AS value
FROM metrics
WHERE id LIKE 'queue_%'
AND created_at > now() - interval '14 day'
@@ -289,6 +298,88 @@ async fn get_queue_metrics(
Ok(Json(queue_metrics))
}
#[derive(Deserialize)]
struct QueueMetricsSeriesQuery {
window_secs: Option<i64>,
}
const QUEUE_METRICS_DEFAULT_WINDOW_SECS: i64 = 24 * 3600;
/// Retention of queue metrics, past which there is nothing left to read.
const QUEUE_METRICS_MAX_WINDOW_SECS: i64 = 14 * 24 * 3600;
async fn get_queue_metrics_series(
authed: ApiAuthed,
Extension(db): Extension<DB>,
Query(query): Query<QueueMetricsSeriesQuery>,
) -> JsonResult<QueueMetricsSeries> {
require_devops_role(&db, &authed).await?;
let window = query
.window_secs
.unwrap_or(QUEUE_METRICS_DEFAULT_WINDOW_SECS)
.clamp(60, QUEUE_METRICS_MAX_WINDOW_SECS);
Ok(Json(read_queue_metrics_series(&db, window as f64).await?))
}
#[derive(Serialize)]
struct QueueTagStatus {
tag: String,
/// Jobs due for more than 3 seconds that no worker has picked up.
waiting: u32,
/// How long the job the next pull would take has been waiting, in seconds.
#[serde(skip_serializing_if = "Option::is_none")]
delay: Option<f64>,
running: i64,
/// Workers that pinged in the last minute and pull this tag.
workers: i64,
}
/// Every tag with jobs waiting or running, read live from the queue. A backlog on a tag no live
/// worker pulls waits for one to start: a worker group scaling up from zero, or none at all for a
/// tag nobody serves.
async fn get_queue_status(
authed: ApiAuthed,
Extension(db): Extension<DB>,
) -> JsonResult<Vec<QueueTagStatus>> {
require_devops_role(&db, &authed).await?;
let backlog = windmill_common::queue::get_queue_stats(&db).await?;
let backlog_tags = backlog.keys().cloned().collect::<Vec<_>>();
// A job's tag is resolved before it is queued (per-workspace and dedicated worker tags
// included), and the pull matches it exactly against the worker's tags, so containment is
// exact here too.
let rows = sqlx::query!(
"WITH running AS (
SELECT tag, count(*) AS n FROM v2_job_queue WHERE running = true GROUP BY tag
)
SELECT t.tag AS \"tag!\", COALESCE(r.n, 0) AS \"running!\",
(SELECT count(*) FROM worker_ping w
WHERE w.ping_at > now() - interval '1 minute' AND w.custom_tags @> ARRAY[t.tag]
) AS \"workers!\"
FROM (SELECT tag::text FROM running UNION SELECT unnest($1::text[])) t(tag)
LEFT JOIN running r ON r.tag = t.tag
ORDER BY t.tag",
&backlog_tags[..],
)
.fetch_all(&db)
.await?;
Ok(Json(
rows.into_iter()
.map(|row| {
let stat = backlog.get(&row.tag);
QueueTagStatus {
waiting: stat.map_or(0, |s| s.count),
delay: stat.map(|s| s.delay),
running: row.running,
workers: row.workers,
tag: row.tag,
}
})
.collect(),
))
}
async fn get_queue_counts(
authed: ApiAuthed,
Extension(db): Extension<DB>,
backend/windmill-api/openapi.yaml
+92
View File
@@ -21834,6 +21834,98 @@ paths:
- id
- values
/workers/queue_metrics_series:
get:
summary: get the queue metrics of a time window, as a bounded line per tag
operationId: getQueueMetricsSeries
tags:
- worker
parameters:
- name: window_secs
in: query
required: false
description: how far back to read, in seconds (defaults to one day, capped at the 14-day retention)
schema:
type: integer
responses:
"200":
description: jobs waiting and queue delay per tag, as the vertices of lines joined by straight segments
content:
application/json:
schema:
type: object
properties:
from:
type: integer
description: start of the window, in epoch milliseconds
to:
type: integer
description: end of the window, in epoch milliseconds
tags:
type: array
items:
type: object
properties:
tag:
type: string
count:
type: array
description: "[epoch ms, jobs waiting more than 3 seconds] vertices"
items:
type: array
items:
type: number
delay:
type: array
description: "[epoch ms, seconds the next job has waited] vertices"
items:
type: array
items:
type: number
required:
- tag
- count
- delay
required:
- from
- to
- tags
/workers/queue_status:
get:
summary: get the live queue status of every tag with jobs waiting or running
operationId: getQueueStatus
tags:
- worker
responses:
"200":
description: queue status per tag
content:
application/json:
schema:
type: array
items:
type: object
properties:
tag:
type: string
waiting:
type: integer
description: jobs due for more than 3 seconds that no worker has picked up
delay:
type: number
description: seconds the job the next pull would take has been waiting, absent when none is
running:
type: integer
workers:
type: integer
description: workers that pinged in the last minute and pull this tag
required:
- tag
- waiting
- running
- workers
/workers/queue_counts:
get:
summary: get counts of jobs waiting for an executor per tag
backend/windmill-common/src/lib.rs
+1
View File
@@ -111,6 +111,7 @@ pub use pipeline_advanced_ee as pipeline_advanced;
pub use pipeline_advanced_oss as pipeline_advanced;
pub mod query_builders;
pub mod queue;
pub mod queue_metrics;
pub mod result_stream;
pub mod runnable_settings;
pub mod schedule;
backend/windmill-common/src/queue.rs
+21 -10
View File
@@ -20,6 +20,8 @@ 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,
/// When that job started waiting (its `scheduled_for`), in epoch seconds.
pub head_since: f64,
}
/// Same backlog as [`get_queue_counts`], plus the delay of the job at the head of each
@@ -36,22 +38,31 @@ pub async fn get_queue_stats(
// 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!\"
"SELECT tag AS \"tag!\", count AS \"count!\",
EXTRACT(EPOCH FROM now() - head)::double precision AS \"delay!\",
EXTRACT(EPOCH FROM head)::double precision AS \"head_since!\"
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",
SELECT tag, sum(n)::bigint AS count,
(array_agg(head ORDER BY priority DESC NULLS LAST))[1] AS head
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
) t",
)
.fetch_all(db)
.await?;
Ok(rows
.into_iter()
.map(|x| (x.tag, QueueStat { count: x.count as u32, delay: x.delay }))
.map(|x| {
(
x.tag,
QueueStat { count: x.count as u32, delay: x.delay, head_since: x.head_since },
)
})
.collect())
}
backend/windmill-common/src/queue_metrics.rs
+482
View File
@@ -0,0 +1,482 @@
//! The queue metrics the monitor samples into `metrics` (`queue_count_{tag}` and
//! `queue_delay_{tag}`), and how a stored series is drawn back.
//!
//! A stored value is a number, held until the next sample, or, for a delay, `{"since": <epoch
//! seconds>}`: the job at the head of the queue has been waiting since then and was still there
//! when sampled, so the delay climbs one second per second until the next sample. Besides
//! [`QueueSample`], the SQL in [`read_queue_metrics_series`] and in `GET /workers/queue_metrics`
//! decodes both shapes.
use std::collections::BTreeMap;
use serde::Serialize;
use sqlx::{Pool, Postgres};
pub const QUEUE_COUNT_PREFIX: &str = "queue_count_";
pub const QUEUE_DELAY_PREFIX: &str = "queue_delay_";
/// A backlogged tag whose value has not moved is re-sampled only this often. A longer heartbeat
/// writes fewer rows, but keeps a tag whose drain was never recorded (no server was up when it
/// drained) drawn as backlogged for longer.
pub const QUEUE_METRIC_HEARTBEAT_SECS: f64 = 5.0 * 60.0;
/// A series silent for longer than this has drained: the sampler stops looking for it, so no
/// closing zero will come, and it is drawn as zero from there. Heartbeats land up to a monitor
/// tick and a sampling slot late, so this must stay well above their real spacing.
pub const QUEUE_METRIC_STALE_SECS: f64 = 3.0 * QUEUE_METRIC_HEARTBEAT_SECS;
/// Heads that started waiting within this of each other are one wait: jobs queued together
/// leave the head one after another without the delay dropping.
pub const QUEUE_DELAY_SAME_HEAD_SECS: f64 = 1.0;
/// Slots a series is split into, whatever the window. A slot draws at most four vertices, and a
/// climb one more at each slot boundary it crosses, so a line stays under about 600 points
/// however many rows the window holds.
const QUEUE_METRICS_SERIES_SLOTS: f64 = 120.0;
/// A stored sample, as it is drawn from the moment it was written until the next one.
#[derive(Debug, Clone, Copy, PartialEq)]
pub enum QueueSample {
/// A count, or a delay while the head keeps changing, which hovers.
Held(f64),
/// A delay while the job that started waiting at `since` (epoch seconds) stays at the head.
Climbing { since: f64 },
}
impl QueueSample {
pub fn parse(value: &serde_json::Value) -> Option<Self> {
match value.get("since") {
Some(since) => since.as_f64().map(|since| Self::Climbing { since }),
None => value.as_f64().map(Self::Held),
}
}
pub fn to_json(self) -> serde_json::Value {
match self {
Self::Held(value) => serde_json::json!(value),
Self::Climbing { since } => serde_json::json!({ "since": since }),
}
}
/// Its value at `t`, in epoch seconds.
pub fn value_at(self, t: f64) -> f64 {
match self {
Self::Held(value) => value,
Self::Climbing { since } => t - since,
}
}
/// When the job at the head of a delay sample written at `at` started waiting.
pub fn head_since(self, at: f64) -> f64 {
match self {
Self::Held(delay) => at - delay,
Self::Climbing { since } => since,
}
}
}
#[derive(Serialize)]
pub struct QueueMetricsSeries {
/// The window drawn, in epoch milliseconds.
pub from: i64,
pub to: i64,
pub tags: Vec<QueueTagSeries>,
}
#[derive(Serialize)]
pub struct QueueTagSeries {
pub tag: String,
/// Vertices `[epoch ms, value]` of a line joined by straight segments.
pub count: Vec<(i64, f64)>,
pub delay: Vec<(i64, f64)>,
}
/// The queue metrics of the last `window_secs`, each series aggregated per slot by the database
/// and drawn by [`render_series`], so the size is bounded by the number of tags rather than by
/// how many rows they wrote.
///
/// Reads the metrics of every workspace's tags: a caller exposing the result MUST restrict it to
/// devops users, as `GET /workers/queue_metrics_series` does.
pub async fn read_queue_metrics_series(
db: &Pool<Postgres>,
window_secs: f64,
) -> crate::error::Result<QueueMetricsSeries> {
let to = sqlx::query_scalar!("SELECT EXTRACT(EPOCH FROM now())::double precision AS \"now!\"")
.fetch_one(db)
.await?;
let from = to - window_secs;
let slot_secs = window_secs / QUEUE_METRICS_SERIES_SLOTS;
// Slot -1 holds the samples written before the window, of which only the last is used: it
// sets the value in force at the left edge. A series silent for longer than the stale window
// reads as zero, so nothing older can matter. Arrays compare element by element, so
// `max(ARRAY[t, v])` is the slot's latest sample, found without sorting every row. `v` is a
// sample's value when it was written: for a climbing delay, how long its head had waited.
//
// A climb keeps rising until the next sample, so when that sample lands in the same slot
// (the tag drained, or its head moved), the climb's top is higher than any `v`. Looking the
// next sample up for the slot's last climb, rather than ordering every row, keeps the pass a
// plain aggregate; an earlier climb in the same slot still shows up to its last heartbeat.
// `t` round-trips through `to_timestamp` to within a microsecond either way, so both bounds
// carry a millisecond of slack, far less than two distinct samples of a series are apart:
// without it the climbing sample can match itself, or the one at `last` fall outside.
let rows = sqlx::query!(
"WITH slots AS (
SELECT id, slot, min(t) AS first, max(t) AS last, max(v) AS peak,
(min(ARRAY[t, v]))[2] AS first_value, (max(ARRAY[t, v]))[2] AS last_value,
(max(ARRAY[t, climbing]))[2] = 1 AS last_climbing,
COALESCE(bool_and(climbing = 1) AND max(since) - min(since) < $4, false) AS ramp,
max(ARRAY[t, since]) FILTER (WHERE climbing = 1) AS last_climb
FROM (
SELECT id, t,
CASE jsonb_typeof(value)
WHEN 'number' THEN value::double precision
WHEN 'object' THEN t - (value->>'since')::double precision
END AS v,
(value->>'since')::double precision AS since,
(jsonb_typeof(value) = 'object')::int::double precision AS climbing,
greatest(floor((t - $1::double precision) / $2::double precision), -1)::int
AS slot
FROM (
SELECT id, value, EXTRACT(EPOCH FROM created_at)::double precision AS t
FROM metrics
WHERE id LIKE 'queue_%'
AND created_at > to_timestamp($1::double precision - $3::double precision)
) m
) s
WHERE v IS NOT NULL
GROUP BY id, slot
)
SELECT id AS \"id!\", slot AS \"slot!\", first AS \"first!\", last AS \"last!\",
greatest(peak, CASE WHEN last_climb[1] < last THEN (
SELECT EXTRACT(EPOCH FROM min(n.created_at))::double precision
FROM metrics n
WHERE n.id = slots.id AND n.id LIKE 'queue_%'
AND n.created_at > to_timestamp(last_climb[1] + 0.001)
AND n.created_at <= to_timestamp(last + 0.001)
) - last_climb[2] END) AS \"peak!\",
first_value AS \"first_value!\", last_value AS \"last_value!\",
last_climbing AS \"last_climbing!\", ramp AS \"ramp!\"
FROM slots
ORDER BY id, slot",
from,
slot_secs,
QUEUE_METRIC_STALE_SECS,
QUEUE_DELAY_SAME_HEAD_SECS,
)
.fetch_all(db)
.await?;
#[derive(Default)]
struct Stored {
carried: Option<MetricSlot>,
slots: Vec<MetricSlot>,
}
// [count, delay] per tag.
let mut stored: BTreeMap<String, [Stored; 2]> = BTreeMap::new();
for row in rows {
let (series, tag) = if let Some(tag) = row.id.strip_prefix(QUEUE_COUNT_PREFIX) {
(0, tag)
} else if let Some(tag) = row.id.strip_prefix(QUEUE_DELAY_PREFIX) {
(1, tag)
} else {
continue;
};
let series = &mut stored.entry(tag.to_string()).or_default()[series];
let slot = MetricSlot {
first: row.first,
last: row.last,
peak: row.peak,
first_value: row.first_value,
last_value: row.last_value,
last_climbing: row.last_climbing,
ramp: row.ramp,
};
if row.slot < 0 {
series.carried = Some(slot);
} else {
series.slots.push(slot);
}
}
let tags = stored
.into_iter()
.map(|(tag, [count, delay])| {
let draw =
|s: &Stored| render_series(s.carried.as_ref(), &s.slots, from, to, slot_secs);
QueueTagSeries { count: draw(&count), delay: draw(&delay), tag }
})
// A tag that drained before the window has nothing to draw in it.
.filter(|s| s.count.iter().chain(&s.delay).any(|(_, v)| *v != 0.0))
.collect();
Ok(QueueMetricsSeries {
from: (from * 1000.0).round() as i64,
to: (to * 1000.0).round() as i64,
tags,
})
}
/// The stored samples of one series that fall in one time slot.
#[derive(Debug, Clone, Copy)]
pub struct MetricSlot {
/// When the first and the last sample of the slot were written, in epoch seconds.
pub first: f64,
pub last: f64,
/// The highest value the series drew over the slot, a climb that ends inside it included.
pub peak: f64,
pub first_value: f64,
/// The value of the last sample, which holds (or climbs, for a climbing delay) until the next.
pub last_value: f64,
pub last_climbing: bool,
/// Every sample of the slot climbs from the same head, so the slot is one exact ramp.
pub ramp: bool,
}
/// Draw a stored series over `[from, to]` (epoch seconds), split into slots of `slot_secs`, as
/// the vertices of a line joined by straight segments, each `(epoch ms, value)`.
///
/// A sample holds its value, or a climbing delay keeps climbing, until the next sample or until
/// the series has been silent for [`QUEUE_METRIC_STALE_SECS`]. `carried` is the slot before
/// `from`, whose last sample sets the left edge. A slot draws its peak across the span of its
/// samples, so a spike shorter than a slot still shows at full height, unless it is a single
/// climb, drawn exactly. A climb gets a vertex at every slot boundary it crosses: the delay axis
/// is logarithmic, so one straight segment across many slots would misplace it.
pub fn render_series(
carried: Option<&MetricSlot>,
slots: &[MetricSlot],
from: f64,
to: f64,
slot_secs: f64,
) -> Vec<(i64, f64)> {
let mut line = Line { points: vec![], from, slot_secs };
let mut held = carried
.map(Held::after)
.filter(|h| from - h.at <= QUEUE_METRIC_STALE_SECS);
if let Some(h) = held {
line.push(from, h.value_at(from));
}
for slot in slots {
let entering = line.advance(&mut held, slot.first);
line.push(slot.first, entering);
if slot.ramp {
line.push(slot.first, slot.first_value);
} else {
line.push(slot.first, slot.peak);
line.push(slot.last, slot.peak);
}
line.push(slot.last, slot.last_value);
held = Some(Held::after(slot));
}
if !line.points.is_empty() {
let value = line.advance(&mut held, to);
line.push(to, value);
}
line.points
}
/// The last sample drawn: when it was written, its value then, and whether it climbs from there.
#[derive(Clone, Copy)]
struct Held {
at: f64,
value: f64,
climbing: bool,
}
impl Held {
fn after(slot: &MetricSlot) -> Self {
Self { at: slot.last, value: slot.last_value, climbing: slot.last_climbing }
}
fn value_at(self, t: f64) -> f64 {
if self.climbing {
self.value + (t - self.at)
} else {
self.value
}
}
}
struct Line {
points: Vec<(i64, f64)>,
from: f64,
slot_secs: f64,
}
impl Line {
/// The value `held` has at `t`, drawing the climb that leads there and, when the series went
/// silent for too long first, its drop to zero, after which it is forgotten.
fn advance(&mut self, held: &mut Option<Held>, t: f64) -> f64 {
let Some(h) = *held else {
return 0.0;
};
let stale_at = h.at + QUEUE_METRIC_STALE_SECS;
if h.climbing {
let end = t.min(stale_at);
let start = h.at.max(self.from);
let mut boundary = self.from
+ ((start - self.from) / self.slot_secs).floor() * self.slot_secs
+ self.slot_secs;
while boundary < end {
self.push(boundary, h.value_at(boundary));
boundary += self.slot_secs;
}
}
if t <= stale_at {
return h.value_at(t);
}
self.push(stale_at, h.value_at(stale_at));
self.push(stale_at, 0.0);
*held = None;
0.0
}
fn push(&mut self, t: f64, value: f64) {
let point = ((t * 1000.0).round() as i64, value);
match self.points.as_mut_slice() {
[.., last] if *last == point => {}
// A horizontal run only needs its two ends.
[.., a, b] if a.1 == value && b.1 == value => b.0 = point.0,
_ => self.points.push(point),
}
}
}
#[cfg(test)]
mod tests {
use super::*;
const FROM: f64 = 1_000_000.0;
const TO: f64 = FROM + 3600.0;
const SLOT: f64 = 30.0;
fn held(first: f64, last: f64, peak: f64, last_value: f64) -> MetricSlot {
MetricSlot {
first: FROM + first,
last: FROM + last,
peak,
first_value: peak,
last_value,
last_climbing: false,
ramp: false,
}
}
/// A slot whose samples all climb from a head that started waiting 30s before `FROM`.
fn climbing(first: f64, last: f64) -> MetricSlot {
MetricSlot {
first: FROM + first,
last: FROM + last,
peak: last + 30.0,
first_value: first + 30.0,
last_value: last + 30.0,
last_climbing: true,
ramp: true,
}
}
fn at(secs: f64, value: f64) -> (i64, f64) {
(((FROM + secs) * 1000.0) as i64, value)
}
#[test]
fn a_value_holds_until_the_next_sample_and_a_drain_drops_where_it_was_written() {
let line = render_series(
None,
&[
held(60.0, 60.0, 3.0, 3.0),
held(600.0, 600.0, 2.0, 2.0),
held(900.0, 900.0, 0.0, 0.0),
],
FROM,
TO,
SLOT,
);
assert_eq!(
line,
vec![
at(60.0, 0.0),
at(60.0, 3.0),
at(600.0, 3.0),
at(600.0, 2.0),
at(900.0, 2.0),
at(900.0, 0.0),
at(3600.0, 0.0),
]
);
}
#[test]
fn a_series_silent_past_the_stale_window_drops_to_zero() {
let line = render_series(None, &[held(60.0, 60.0, 3.0, 3.0)], FROM, TO, SLOT);
let dropped = 60.0 + QUEUE_METRIC_STALE_SECS;
assert_eq!(
line,
vec![
at(60.0, 0.0),
at(60.0, 3.0),
at(dropped, 3.0),
at(dropped, 0.0),
at(3600.0, 0.0)
]
);
}
#[test]
fn a_slot_draws_its_peak_then_continues_from_its_last_sample() {
// Samples at 60 (5), 70 (9), 80 (4) collapsed into one slot.
let line = render_series(
Some(&held(-30.0, -30.0, 2.0, 2.0)),
&[held(60.0, 80.0, 9.0, 4.0)],
FROM,
FROM + 120.0,
SLOT,
);
assert_eq!(
line,
vec![
at(0.0, 2.0),
at(60.0, 2.0),
at(60.0, 9.0),
at(80.0, 9.0),
at(80.0, 4.0),
at(120.0, 4.0)
]
);
}
#[test]
fn a_climbing_delay_is_drawn_exactly_up_to_its_drain() {
// 300s slots: one holds two climbing samples, and heartbeats follow until the drain.
let line = render_series(
None,
&[
climbing(60.0, 120.0),
climbing(360.0, 360.0),
climbing(660.0, 660.0),
held(900.0, 900.0, 0.0, 0.0),
],
FROM,
TO,
300.0,
);
assert_eq!(
line,
vec![
at(60.0, 0.0),
// The slot is one climb, not its peak held across it.
at(60.0, 90.0),
at(120.0, 150.0),
// A vertex at each slot boundary the climb crosses.
at(300.0, 330.0),
at(360.0, 390.0),
at(600.0, 630.0),
at(660.0, 690.0),
// Still climbing right up to the closing zero.
at(900.0, 930.0),
at(900.0, 0.0),
at(3600.0, 0.0),
]
);
}
}
backend/windmill-common/tests/queue_metrics_series.rs
+136
View File
@@ -0,0 +1,136 @@
use serde_json::json;
use sqlx::{Pool, Postgres};
use windmill_common::queue_metrics::{read_queue_metrics_series, QUEUE_METRIC_STALE_SECS};
const WINDOW: f64 = 3600.0;
/// Store a sample written `at` seconds after the start of a `WINDOW` ending now.
async fn sample(db: &Pool<Postgres>, id: &str, value: serde_json::Value, at: f64) {
sqlx::query(
"INSERT INTO metrics (id, value, created_at) VALUES ($1, $2, now() - make_interval(secs => $3))",
)
.bind(id)
.bind(value)
.bind(WINDOW - at)
.execute(db)
.await
.expect("failed to store a metric sample");
}
/// The database hands the renderer the last sample before the window, which sets the left edge,
/// and for each slot its peak and its latest value, which the line continues from.
#[sqlx::test(migrations = "../migrations")]
async fn a_series_starts_from_the_sample_before_the_window_and_keeps_each_slot_peak(
db: Pool<Postgres>,
) {
// Before the window: 1, then 2, which is what is in force at the left edge.
sample(&db, "queue_count_t", json!(1), -120.0).await;
sample(&db, "queue_count_t", json!(2), -60.0).await;
// Three samples inside one 30s slot: the line rises to their peak, then drops to the last.
sample(&db, "queue_count_t", json!(5), 605.0).await;
sample(&db, "queue_count_t", json!(9), 612.0).await;
sample(&db, "queue_count_t", json!(4), 620.0).await;
// Drained before the window: nothing left to draw.
sample(&db, "queue_count_gone", json!(3), -300.0).await;
sample(&db, "queue_count_gone", json!(0), -200.0).await;
let series = read_queue_metrics_series(&db, WINDOW).await.unwrap();
assert_eq!(
series.tags.len(),
1,
"a tag drained before the window is left out"
);
let tag = &series.tags[0];
assert_eq!(tag.tag, "t");
assert!(tag.delay.is_empty());
let stale = 620.0 + QUEUE_METRIC_STALE_SECS;
let expected = [
(0.0, 2.0),
(605.0, 2.0),
(605.0, 9.0),
(620.0, 9.0),
(620.0, 4.0),
(stale, 4.0),
(stale, 0.0),
(WINDOW, 0.0),
];
assert_eq!(tag.count.len(), expected.len(), "vertices: {:?}", tag.count);
for ((ms, value), (at, expected_value)) in tag.count.iter().zip(expected) {
let secs = (*ms - series.from) as f64 / 1000.0;
// Samples are stored a few milliseconds before the window is read.
assert!(
(secs - at).abs() < 2.0 && *value == expected_value,
"expected ({at}, {expected_value}), got ({secs}, {value}) in {:?}",
tag.count
);
}
}
/// A delay stored as its head's wait start is drawn as that wait, growing a second per second,
/// right up to the zero that closes it.
#[sqlx::test(migrations = "../migrations")]
async fn a_climbing_delay_is_drawn_as_the_wait_of_its_head(db: Pool<Postgres>) {
let now: f64 = sqlx::query_scalar("SELECT EXTRACT(EPOCH FROM now())::double precision")
.fetch_one(&db)
.await
.unwrap();
// The head started waiting 30s before the window; heartbeats restate it until the drain.
let head = json!({ "since": now - WINDOW - 30.0 });
for at in [60.0, 360.0, 660.0] {
sample(&db, "queue_delay_t", head.clone(), at).await;
}
sample(&db, "queue_delay_t", json!(0), 900.0).await;
let series = read_queue_metrics_series(&db, WINDOW).await.unwrap();
let points = series.tags[0]
.delay
.iter()
.map(|(ms, value)| ((*ms - series.from) as f64 / 1000.0, *value))
.collect::<Vec<_>>();
let climb = points
.iter()
.filter(|(_, value)| *value > 0.0)
.collect::<Vec<_>>();
assert!(
climb.len() > 4,
"the climb has vertices along the way: {points:?}"
);
for (at, value) in &climb {
assert!(
(value - (at + 30.0)).abs() < 2.0,
"off the climb at {at}: {points:?}"
);
}
let (first, _) = climb[0];
let (top, _) = climb[climb.len() - 1];
assert!(
(first - 60.0).abs() < 2.0 && (top - 900.0).abs() < 2.0,
"{points:?}"
);
}
/// A climb that drains inside its slot keeps its top, which no stored value holds: it is reached
/// at the next sample.
#[sqlx::test(migrations = "../migrations")]
async fn a_climb_that_drains_inside_its_slot_keeps_its_top(db: Pool<Postgres>) {
let now: f64 = sqlx::query_scalar("SELECT EXTRACT(EPOCH FROM now())::double precision")
.fetch_one(&db)
.await
.unwrap();
// All in the 30s slot starting at 600: held at 5s, then climbing from a head queued at 597,
// which is still there when the tag drains at 627, 30s into its wait.
sample(&db, "queue_delay_t", json!(5), 602.0).await;
sample(&db, "queue_delay_t", json!({ "since": now - WINDOW + 597.0 }), 610.0).await;
sample(&db, "queue_delay_t", json!(0), 627.0).await;
let series = read_queue_metrics_series(&db, WINDOW).await.unwrap();
let top = series.tags[0]
.delay
.iter()
.map(|(_, value)| *value)
.fold(0.0, f64::max);
assert!((top - 30.0).abs() < 2.0, "{:?}", series.tags[0].delay);
}
backend/windmill-common/tests/queue_stats.rs
+6
View File
@@ -45,6 +45,10 @@ async fn queue_stats_report_the_delay_of_the_job_pulled_next(db: Pool<Postgres>)
queue_job(&db, "unprioritized", None, 50.0, false).await;
let stats = get_queue_stats(&db).await.unwrap();
let now: f64 = sqlx::query_scalar("SELECT EXTRACT(EPOCH FROM now())::double precision")
.fetch_one(&db)
.await
.unwrap();
let mixed = &stats["mixed"];
assert_eq!(mixed.count, 4);
@@ -53,6 +57,8 @@ async fn queue_stats_report_the_delay_of_the_job_pulled_next(db: Pool<Postgres>)
"expected the oldest job of the highest priority, got a delay of {}",
mixed.delay
);
// The same job's wait start, which the delay is measured from.
assert!((mixed.head_since + mixed.delay - now).abs() < 5.0);
let unprioritized = &stats["unprioritized"];
assert_eq!(unprioritized.count, 2);
assert!(
frontend/src/lib/components/QueueMetricsDrawer.svelte
+6 -1
View File
@@ -1,6 +1,7 @@
<script lang="ts">
import { Drawer, DrawerContent } from './common'
import QueueMetricsDrawerInner from './QueueMetricsDrawerInner.svelte'
import QueueStatusTable from './QueueStatusTable.svelte'
import QueueAlerts from './QueueAlerts.svelte'
import WorkspaceFairnessEvents from './WorkspaceFairnessEvents.svelte'
import { enterpriseLicense } from '$lib/stores'
@@ -17,12 +18,16 @@
on:close={drawer.closeDrawer}
documentationLink="https://www.windmill.dev/docs/core_concepts/worker_groups#queue-metrics"
>
<QueueAlerts />
<QueueStatusTable />
<div class="py-8"></div>
<QueueMetricsDrawerInner />
<div class="py-8"></div>
<QueueAlerts />
{#if $enterpriseLicense}
<div class="py-8"></div>
<WorkspaceFairnessEvents />
frontend/src/lib/components/QueueMetricsDrawerInner.svelte
+98 -198
View File
@@ -16,15 +16,17 @@
LogarithmicScale,
TimeScale,
type ChartData,
type ChartOptions,
type Point
} from 'chart.js'
import { WorkerService } from '$lib/gen'
import { resource } from 'runed'
import Skeleton from './common/skeleton/Skeleton.svelte'
import DarkModeObserver from './DarkModeObserver.svelte'
import Alert from './common/alert/Alert.svelte'
import { Section } from './common'
let loading: boolean = $state(true)
import ToggleButtonGroup from './common/toggleButton-v2/ToggleButtonGroup.svelte'
import ToggleButton from './common/toggleButton-v2/ToggleButton.svelte'
const colorTuples = [
['#7EB26D', 'rgba(126, 178, 109, 0.2)'],
@@ -49,11 +51,6 @@
['#6BBA70', 'rgba(107, 186, 112, 0.2)']
]
function getColors(labels: string[]) {
const colors = labels.map((_, i) => colorTuples[i % colorTuples.length])
return Object.fromEntries(colors.map((c, i) => [labels[i], c]))
}
ChartJS.register(
Title,
Tooltip,
@@ -66,130 +63,89 @@
LogarithmicScale
)
let countData: ChartData<'line', Point[], undefined> | undefined = $state(undefined)
let delayData: ChartData<'line', Point[], undefined> | undefined = $state(undefined)
const WINDOWS = {
'1h': { secs: 3600, label: 'hour' },
'24h': { secs: 24 * 3600, label: '24 hours' },
'7d': { secs: 7 * 24 * 3600, label: '7 days' },
'14d': { secs: 14 * 24 * 3600, label: '14 days' }
}
let windowKey: keyof typeof WINDOWS = $state('24h')
let minDate = $state(new Date())
let noMetrics = $state(false)
// 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
}[],
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 metrics = resource(
() => windowKey,
async (key, _, { signal }) => {
try {
return await WorkerService.getQueueMetricsSeries({ windowSecs: WINDOWS[key].secs })
} finally {
// A slower answer for a window no longer selected, success or failure, must not
// replace the current one: `resource` drops the abort error thrown in its place.
signal.throwIfAborted()
}
}
)
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 (
lastValue == undefined ||
Math.abs(el.value - lastValue) > Math.abs(lastValue) * tolerance
) {
push(ts, el.value)
function datasets(
kind: 'count' | 'delay',
toPoint: (vertex: number[]) => Point
): ChartData<'line', Point[], undefined> {
const tags = metrics.current?.tags ?? []
return {
datasets: tags
.map((t, i) => ({ t, colors: colorTuples[i % colorTuples.length] }))
.filter(({ t }) => t[kind].length > 0)
.map(({ t, colors: [color, bgColor] }) => ({
label: t.tag,
borderColor: color,
backgroundColor: bgColor,
data: t[kind].map(toPoint)
}))
}
}
// The server draws each line (a value holds until the next sample, and a spike keeps its
// height when a slot aggregates many samples), so its vertices are joined as they are.
const countData = $derived(datasets('count', ([x, y]) => ({ x, y })))
// 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.
const delayData = $derived(datasets('delay', ([x, y]) => ({ x, y: y === 0 ? 1 : y })))
function chartOptions(title: string, y: ChartOptions<'line'>['scales']): ChartOptions<'line'> {
return {
animation: false,
elements: { point: { radius: 0, hoverRadius: 4 } },
interaction: { mode: 'nearest', axis: 'x', intersect: false },
plugins: { title: { display: true, text: title } },
scales: {
x: { type: 'time', min: metrics.current?.from, max: metrics.current?.to },
...y
}
lastSampleTs = ts
}
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 }))
}
const countOptions = $derived(
chartOptions('Number of delayed jobs per tag (> 3s)', {
y: { title: { display: true, text: 'count' } }
})
)
async function loadMetrics() {
loading = true
let metrics = await WorkerService.getQueueMetrics()
if (metrics.length == 0) {
noMetrics = true
loading = false
return
const delayOptions = $derived.by(() => {
const options = chartOptions('Queue delay per tag (> 3s)', {
y: {
type: 'logarithmic',
title: { display: true, text: 'delay (s)' },
ticks: { callback: (value) => (value === 1 ? '0' : value) }
}
})
options.plugins!.tooltip = {
callbacks: {
label: (context) => {
const y = (context.raw as Point).y
return `${context.dataset.label}: ${y === 1 ? 0 : y}`
}
}
}
const labels = metrics
.map((m) => m.id.slice(12))
.filter((v, i, a) => a.indexOf(v) === i)
.sort()
const labelColors = getColors(labels)
countData = {
datasets: metrics
.filter((m) => m.id.startsWith('queue_count_'))
.map((m) => {
const [color, bgColor] = labelColors[m.id.slice(12)]
return {
label: m.id.slice(12),
backgroundColor: bgColor,
borderColor: color,
stepped: STEPPED,
data: toPoints(m.values, 0)
}
})
}
delayData = {
datasets: metrics
.filter((m) => m.id.startsWith('queue_delay_'))
.map((m) => {
const [color, bgColor] = labelColors[m.id.slice(12)]
return {
label: m.id.slice(12),
borderColor: color,
backgroundColor: bgColor,
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))
}
})
}
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
}
loadMetrics()
return options
})
let darkMode = $state(false)
@@ -204,87 +160,31 @@
<DarkModeObserver bind:darkMode />
<Section label="Queue metrics">
{#if loading}
{#snippet action()}
<ToggleButtonGroup bind:selected={windowKey} noWFull>
{#snippet children({ item })}
{#each Object.keys(WINDOWS) as key (key)}
<ToggleButton value={key} label={key} size="sm" {item} />
{/each}
{/snippet}
</ToggleButtonGroup>
{/snippet}
{#if metrics.error}
<Alert type="error" title="Failed to load the queue metrics">{metrics.error.message}</Alert>
{:else if metrics.current === undefined}
<Skeleton layout={[[20]]} />
{:else if noMetrics}
<p class="text-secondary">No jobs delayed by more than 3 seconds in the last 14 days</p>
{:else if metrics.current.tags.length === 0}
<p class="text-secondary text-xs">
No jobs delayed by more than 3 seconds in the last {WINDOWS[windowKey].label}
</p>
{:else}
<div class="flex flex-col gap-4">
{#if countData}
<Line
data={countData}
options={{
animation: false,
plugins: {
title: {
display: true,
text: 'Number of delayed jobs per tag (> 3s)'
}
},
scales: {
x: {
type: 'time',
min: minDate.toISOString(),
max: new Date().toISOString()
},
y: {
title: {
display: true,
text: 'count'
}
}
}
}}
/>
{/if}
{#if delayData}
<Line
data={delayData}
options={{
animation: false,
plugins: {
title: {
display: true,
text: 'Queue delay per tag (> 3s)'
},
tooltip: {
callbacks: {
label: function (context) {
// @ts-ignore
if (context.raw.y === 1) {
return context.dataset.label + ': 0'
} else {
// @ts-ignore
return context.dataset.label + ': ' + context.raw.y
}
}
}
}
},
scales: {
x: {
type: 'time',
min: minDate.toISOString(),
max: new Date().toISOString()
},
y: {
type: 'logarithmic',
title: {
display: true,
text: 'delay (s)'
},
ticks: {
callback: (value, _) => (value === 1 ? '0' : value)
}
}
}
}}
/>
{/if}
<Line data={countData} options={countOptions} />
<Line data={delayData} options={delayOptions} />
<Alert title="Info">
Only tags for jobs that have been delayed by more than 3 seconds in the last 14 days are
included in the graph.
Only tags with jobs delayed by more than 3 seconds in this window are included. At wide
windows a line shows the highest value of each time slot, so short spikes stay visible.
</Alert>
</div>
{/if}
frontend/src/lib/components/QueueStatusTable.svelte
+115
View File
@@ -0,0 +1,115 @@
<script lang="ts">
import { WorkerService, type GetQueueStatusResponse } from '$lib/gen'
import { RefreshCw, TriangleAlert } from 'lucide-svelte'
import { twMerge } from 'tailwind-merge'
import { Alert, Button, Section, Skeleton } from './common'
import DataTable from './table/DataTable.svelte'
import Head from './table/Head.svelte'
import Cell from './table/Cell.svelte'
import { msToReadableTime } from '$lib/utils'
import { onMount } from 'svelte'
const REFRESH_MS = 10_000
let status = $state<GetQueueStatusResponse>()
let loading = $state(false)
let error = $state<string>()
async function load() {
loading = true
try {
status = await WorkerService.getQueueStatus()
error = undefined
} catch (e) {
error = e instanceof Error ? e.message : String(e)
} finally {
loading = false
}
}
onMount(() => {
load()
// A tick that lands while the last read is still running is skipped: the endpoint is slowest
// on the backlogged instance this table is opened for, and reads must not pile up.
const interval = setInterval(() => {
if (!loading) load()
}, REFRESH_MS)
return () => clearInterval(interval)
})
type TagStatus = GetQueueStatusResponse[number]
// A backlog no worker currently pulls leads the table: it waits until one starts.
function unserved(s: TagStatus) {
return s.waiting > 0 && s.workers === 0
}
const rows = $derived(
[...(status ?? [])].sort(
(a, b) =>
Number(unserved(b)) - Number(unserved(a)) ||
(b.delay ?? -1) - (a.delay ?? -1) ||
b.running - a.running ||
a.tag.localeCompare(b.tag)
)
)
</script>
<Section
label="Queue status"
tooltip="Waiting counts jobs due for more than 3 seconds that no worker has picked up. Next job's wait is how long the job the next pull would take has been waiting. Workers counts the workers that pinged in the last minute and pull the tag."
>
{#snippet action()}
<Button
variant="subtle"
unifiedSize="sm"
startIcon={{ icon: RefreshCw, classes: twMerge(loading ? 'animate-spin' : '') }}
iconOnly
title="Refresh queue status"
onclick={load}
disabled={loading}
/>
{/snippet}
{#if error}
<Alert type="error" title="Failed to load the queue status">{error}</Alert>
{:else if status === undefined}
<Skeleton layout={[[6]]} />
{:else if rows.length === 0}
<p class="text-secondary text-xs">No jobs are waiting or running.</p>
{:else}
<DataTable size="sm" noBorder={false} rounded={true}>
<Head>
<tr>
<Cell head first>Tag</Cell>
<Cell head numeric>Waiting</Cell>
<Cell head numeric>Next job's wait</Cell>
<Cell head numeric>Running</Cell>
<Cell head last>Workers</Cell>
</tr>
</Head>
<tbody>
{#each rows as s (s.tag)}
<tr class="border-b last:border-b-0">
<Cell first class="text-xs font-mono text-primary">{s.tag}</Cell>
<Cell numeric class="text-xs text-primary">{s.waiting}</Cell>
<Cell numeric class="text-xs text-primary">
{s.delay === undefined ? '-' : msToReadableTime(s.delay * 1000, 0)}
</Cell>
<Cell numeric class="text-xs text-primary">{s.running}</Cell>
<Cell last class="text-xs text-primary">
{#if unserved(s)}
<span class="inline-flex items-center gap-1 text-yellow-600 dark:text-yellow-400">
<TriangleAlert size={14} />
No worker currently pulls this tag
</span>
{:else}
{s.workers}
{/if}
</Cell>
</tr>
{/each}
</tbody>
</DataTable>
{/if}
</Section>