diff --git a/backend/.sqlx/query-2742245bc03290120a97b21c441cb56825e9fd552a7aeddfb8a372540c19b863.json b/backend/.sqlx/query-2742245bc03290120a97b21c441cb56825e9fd552a7aeddfb8a372540c19b863.json new file mode 100644 index 0000000000..5f3d8300eb --- /dev/null +++ b/backend/.sqlx/query-2742245bc03290120a97b21c441cb56825e9fd552a7aeddfb8a372540c19b863.json @@ -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" +} diff --git a/backend/.sqlx/query-38e12be0764d850538d3eea328aebfaa3e7ec3a7c682945f22e8cde3a63094f4.json b/backend/.sqlx/query-38e12be0764d850538d3eea328aebfaa3e7ec3a7c682945f22e8cde3a63094f4.json deleted file mode 100644 index fe9c2f27e3..0000000000 --- a/backend/.sqlx/query-38e12be0764d850538d3eea328aebfaa3e7ec3a7c682945f22e8cde3a63094f4.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-3bd816e986ef2d2a193e51c985b61c04b71f464021e4b884bf21cc7c25f6a753.json b/backend/.sqlx/query-3bd816e986ef2d2a193e51c985b61c04b71f464021e4b884bf21cc7c25f6a753.json new file mode 100644 index 0000000000..843484dd86 --- /dev/null +++ b/backend/.sqlx/query-3bd816e986ef2d2a193e51c985b61c04b71f464021e4b884bf21cc7c25f6a753.json @@ -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" +} diff --git a/backend/.sqlx/query-44dd7a66ecc9564ad5727970b5f60a1717eda8924999741be522ffb74cc173fa.json b/backend/.sqlx/query-44dd7a66ecc9564ad5727970b5f60a1717eda8924999741be522ffb74cc173fa.json deleted file mode 100644 index 9f97b97a88..0000000000 --- a/backend/.sqlx/query-44dd7a66ecc9564ad5727970b5f60a1717eda8924999741be522ffb74cc173fa.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-58f99e0d5877f403cde04459e4a425efd77aa4c0118f62ada2caa0794b0d738b.json b/backend/.sqlx/query-58f99e0d5877f403cde04459e4a425efd77aa4c0118f62ada2caa0794b0d738b.json new file mode 100644 index 0000000000..f0d9aabea3 --- /dev/null +++ b/backend/.sqlx/query-58f99e0d5877f403cde04459e4a425efd77aa4c0118f62ada2caa0794b0d738b.json @@ -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" +} diff --git a/backend/.sqlx/query-7745127eb4a4be2b67427a708e8e5bf2973af84cf315047e586071438cd5e438.json b/backend/.sqlx/query-7745127eb4a4be2b67427a708e8e5bf2973af84cf315047e586071438cd5e438.json new file mode 100644 index 0000000000..8eac288a1b --- /dev/null +++ b/backend/.sqlx/query-7745127eb4a4be2b67427a708e8e5bf2973af84cf315047e586071438cd5e438.json @@ -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" +} diff --git a/backend/.sqlx/query-7af0fd3d8dd1d949ce11b190a4fa6b56c84904aeed791b2a0879b36add6bf9b5.json b/backend/.sqlx/query-7af0fd3d8dd1d949ce11b190a4fa6b56c84904aeed791b2a0879b36add6bf9b5.json new file mode 100644 index 0000000000..ae8cc698b9 --- /dev/null +++ b/backend/.sqlx/query-7af0fd3d8dd1d949ce11b190a4fa6b56c84904aeed791b2a0879b36add6bf9b5.json @@ -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" +} diff --git a/backend/.sqlx/query-a498f752e169b711c0cb26ac167b2e554a7fdfc1055389fd65f330469e329d16.json b/backend/.sqlx/query-a498f752e169b711c0cb26ac167b2e554a7fdfc1055389fd65f330469e329d16.json new file mode 100644 index 0000000000..8c6dd010af --- /dev/null +++ b/backend/.sqlx/query-a498f752e169b711c0cb26ac167b2e554a7fdfc1055389fd65f330469e329d16.json @@ -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" +} diff --git a/backend/.sqlx/query-bd46b7a19f9e51de1babc3d16fd7d7b7776b90a96c6662b00d8a999b74eb5b36.json b/backend/.sqlx/query-bd46b7a19f9e51de1babc3d16fd7d7b7776b90a96c6662b00d8a999b74eb5b36.json deleted file mode 100644 index 5dce7b09cc..0000000000 --- a/backend/.sqlx/query-bd46b7a19f9e51de1babc3d16fd7d7b7776b90a96c6662b00d8a999b74eb5b36.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-d09d3a6223262ac584f045feec7e08ea1695e41d9b11538b58189925ddba4fdd.json b/backend/.sqlx/query-d09d3a6223262ac584f045feec7e08ea1695e41d9b11538b58189925ddba4fdd.json deleted file mode 100644 index ad7d1bf057..0000000000 --- a/backend/.sqlx/query-d09d3a6223262ac584f045feec7e08ea1695e41d9b11538b58189925ddba4fdd.json +++ /dev/null @@ -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" -} diff --git a/backend/.sqlx/query-d4ce900b8e60b530c2ea57c082edbc8c12c71869c4e45219dad4fb7198aaa1e3.json b/backend/.sqlx/query-d4ce900b8e60b530c2ea57c082edbc8c12c71869c4e45219dad4fb7198aaa1e3.json new file mode 100644 index 0000000000..3f73c2171e --- /dev/null +++ b/backend/.sqlx/query-d4ce900b8e60b530c2ea57c082edbc8c12c71869c4e45219dad4fb7198aaa1e3.json @@ -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" +} diff --git a/backend/src/monitor.rs b/backend/src/monitor.rs index de46f0b9c8..9f8605c17a 100644 --- a/backend/src/monitor.rs +++ b/backend/src/monitor.rs @@ -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) -> 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, queue_stats: &std::collections::HashMap, @@ -5150,11 +5143,12 @@ async fn save_queue_metrics( .collect::>(); // 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> = 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], ) .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, + 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) -> 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, + 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) -> 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 + )); } } diff --git a/backend/windmill-api-workers/src/lib.rs b/backend/windmill-api-workers/src/lib.rs index 6ee55ed003..07bef364c8 100644 --- a/backend/windmill-api-workers/src/lib.rs +++ b/backend/windmill-api-workers/src/lib.rs @@ -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> { 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, +} + +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, + Query(query): Query, +) -> JsonResult { + 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, + 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, +) -> JsonResult> { + require_devops_role(&db, &authed).await?; + + let backlog = windmill_common::queue::get_queue_stats(&db).await?; + let backlog_tags = backlog.keys().cloned().collect::>(); + // 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, diff --git a/backend/windmill-api/openapi.yaml b/backend/windmill-api/openapi.yaml index b16d8bd87b..63adf55f51 100644 --- a/backend/windmill-api/openapi.yaml +++ b/backend/windmill-api/openapi.yaml @@ -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 diff --git a/backend/windmill-common/src/lib.rs b/backend/windmill-common/src/lib.rs index 7fb5e85a69..1639b4605d 100644 --- a/backend/windmill-common/src/lib.rs +++ b/backend/windmill-common/src/lib.rs @@ -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; diff --git a/backend/windmill-common/src/queue.rs b/backend/windmill-common/src/queue.rs index 7166b3d3cd..2008e1ce93 100644 --- a/backend/windmill-common/src/queue.rs +++ b/backend/windmill-common/src/queue.rs @@ -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()) } diff --git a/backend/windmill-common/src/queue_metrics.rs b/backend/windmill-common/src/queue_metrics.rs new file mode 100644 index 0000000000..f5affd7346 --- /dev/null +++ b/backend/windmill-common/src/queue_metrics.rs @@ -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": }`: 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 { + 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, +} + +#[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, + window_secs: f64, +) -> crate::error::Result { + 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, + slots: Vec, + } + // [count, delay] per tag. + let mut stored: BTreeMap = 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, 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), + ] + ); + } +} diff --git a/backend/windmill-common/tests/queue_metrics_series.rs b/backend/windmill-common/tests/queue_metrics_series.rs new file mode 100644 index 0000000000..f6bc7b4056 --- /dev/null +++ b/backend/windmill-common/tests/queue_metrics_series.rs @@ -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, 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, +) { + // 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) { + 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::>(); + + let climb = points + .iter() + .filter(|(_, value)| *value > 0.0) + .collect::>(); + 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) { + 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); +} diff --git a/backend/windmill-common/tests/queue_stats.rs b/backend/windmill-common/tests/queue_stats.rs index 2b7a80a59d..419ad98d56 100644 --- a/backend/windmill-common/tests/queue_stats.rs +++ b/backend/windmill-common/tests/queue_stats.rs @@ -45,6 +45,10 @@ async fn queue_stats_report_the_delay_of_the_job_pulled_next(db: Pool) 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) "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!( diff --git a/frontend/src/lib/components/QueueMetricsDrawer.svelte b/frontend/src/lib/components/QueueMetricsDrawer.svelte index 35aea3eebb..98d67bf444 100644 --- a/frontend/src/lib/components/QueueMetricsDrawer.svelte +++ b/frontend/src/lib/components/QueueMetricsDrawer.svelte @@ -1,6 +1,7 @@ + +
+ {#snippet action()} +