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

fix(alerts): count time-window completions when classifying gates

Browse Source 
The saturation check read only concurrency_counter.job_uuids, but the
limiter admits on running jobs plus those that ended inside
concurrency_time_window_s, so a gate held shut purely by recent
completions read as free and its backlog raised the same false capacity
alert one completion later.

Emitter change lives in windmill-ee-private (see ee-repo-ref.txt bump).
Adds the regression test for that case and refreshes the query cache.
This commit is contained in:
Ruben Fiszel
2026-07-20 05:07:11 +00:00
parent 4b3225e5f3
commit 39c0e7c4aa
4 changed files with 79 additions and 39 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/.sqlx/query-978ff21db190a357799539a6da07e2047df060d61b77f23f730d4a02e7181040.json
Generated
-35
View File
@@ -1,35 +0,0 @@
{
"db_name": "PostgreSQL",
"query": "\n WITH waiting AS (\n SELECT ck.key AS concurrency_key, cs.concurrent_limit,\n COUNT(*) AS waiting_count, MIN(q.scheduled_for) AS oldest_job\n FROM v2_job_queue q\n LEFT JOIN concurrency_key ck ON ck.job_id = q.id\n LEFT JOIN runnable_settings rs ON rs.hash = q.runnable_settings_handle\n LEFT JOIN concurrency_settings cs ON cs.hash = rs.concurrency_settings\n WHERE q.tag = $1\n AND q.scheduled_for <= NOW() - $2::interval\n AND q.running = false\n GROUP BY ck.key, cs.concurrent_limit\n ),\n classified AS (\n SELECT w.waiting_count, w.oldest_job,\n w.concurrent_limit > 0 AND (\n SELECT COUNT(*) FROM jsonb_object_keys(cc.job_uuids)\n ) >= w.concurrent_limit AS gated\n FROM waiting w\n LEFT JOIN concurrency_counter cc ON cc.concurrency_id = w.concurrency_key\n )\n SELECT COALESCE(SUM(waiting_count) FILTER (WHERE gated IS NOT TRUE), 0)::bigint AS count,\n COALESCE(SUM(waiting_count) FILTER (WHERE gated), 0)::bigint AS gated_count,\n MIN(oldest_job) FILTER (WHERE gated IS NOT TRUE) AS oldest_job\n FROM classified\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "count",
"type_info": "Int8"
},
{
"ordinal": 1,
"name": "gated_count",
"type_info": "Int8"
},
{
"ordinal": 2,
"name": "oldest_job",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Text",
"Interval"
]
},
"nullable": [
null,
null,
null
]
},
"hash": "978ff21db190a357799539a6da07e2047df060d61b77f23f730d4a02e7181040"
}
backend/.sqlx/query-e6c6ca9bac609db490191f2a64f5b324c10ddd7aa8ea92efe2a262dc83b80a04.json
Generated
+35
View File
@@ -0,0 +1,35 @@
{
"db_name": "PostgreSQL",
"query": "\n WITH waiting AS (\n SELECT ck.key AS concurrency_key, cs.concurrent_limit,\n cs.concurrency_time_window_s,\n COUNT(*) AS waiting_count, MIN(q.scheduled_for) AS oldest_job\n FROM v2_job_queue q\n LEFT JOIN concurrency_key ck ON ck.job_id = q.id\n LEFT JOIN runnable_settings rs ON rs.hash = q.runnable_settings_handle\n LEFT JOIN concurrency_settings cs ON cs.hash = rs.concurrency_settings\n WHERE q.tag = $1\n AND q.scheduled_for <= NOW() - $2::interval\n AND q.running = false\n GROUP BY ck.key, cs.concurrent_limit, cs.concurrency_time_window_s\n ),\n classified AS MATERIALIZED (\n SELECT w.waiting_count, w.oldest_job,\n w.concurrent_limit > 0 AND (\n (SELECT COUNT(*) FROM jsonb_object_keys(cc.job_uuids))\n + (SELECT COUNT(*) FROM concurrency_key done\n WHERE done.key = w.concurrency_key\n AND done.ended_at >= NOW() - INTERVAL '1 second'\n * COALESCE(w.concurrency_time_window_s, 0))\n ) >= w.concurrent_limit AS gated\n FROM waiting w\n LEFT JOIN concurrency_counter cc ON cc.concurrency_id = w.concurrency_key\n )\n SELECT COALESCE(SUM(waiting_count) FILTER (WHERE gated IS NOT TRUE), 0)::bigint AS count,\n COALESCE(SUM(waiting_count) FILTER (WHERE gated), 0)::bigint AS gated_count,\n MIN(oldest_job) FILTER (WHERE gated IS NOT TRUE) AS oldest_job\n FROM classified\n ",
"describe": {
"columns": [
{
"ordinal": 0,
"name": "count",
"type_info": "Int8"
},
{
"ordinal": 1,
"name": "gated_count",
"type_info": "Int8"
},
{
"ordinal": 2,
"name": "oldest_job",
"type_info": "Timestamptz"
}
],
"parameters": {
"Left": [
"Text",
"Interval"
]
},
"nullable": [
null,
null,
null
]
},
"hash": "e6c6ca9bac609db490191f2a64f5b324c10ddd7aa8ea92efe2a262dc83b80a04"
}
backend/ee-repo-ref.txt
+1 -1
View File
@@ -1 +1 @@
0c71f555ed74a1c801cdc8e5e6c063afcfb14384
6911fbf69af0c7ced2965dc031096184b173d6c6
backend/windmill-common/tests/jobs_waiting_alerts.rs
+43 -3
View File
@@ -57,8 +57,10 @@ mod tests {
/// A concurrency gate as the push path lays it out: the limit hangs off
/// `runnable_settings_handle` -> `concurrency_settings`, and the live
/// admission count is the number of job uuids in `concurrency_counter`.
/// `running` is how many slots are currently taken, so `running == limit`
/// is a saturated gate.
/// `running` is how many slots are currently taken.
///
/// The window is wide so a completion seeded by `seed_recent_completions`
/// stays inside it for the length of the test.
async fn seed_gate(db: &Pool<Postgres>, key: &str, handle: i64, limit: i32, running: usize) {
let job_uuids: serde_json::Value = (0..running)
.map(|_| (Uuid::new_v4().hyphenated().to_string(), json!({})))
@@ -67,7 +69,7 @@ mod tests {
sqlx::query!(
"INSERT INTO concurrency_settings (hash, concurrency_key, concurrent_limit, concurrency_time_window_s)
VALUES ($1, $2, $3, 5)",
VALUES ($1, $2, $3, 600)",
handle,
key,
limit,
@@ -125,6 +127,22 @@ mod tests {
}
}
/// Jobs that already finished inside the gate's time window. The limiter
/// counts these toward the limit, so they hold the gate shut even though
/// they are gone from `concurrency_counter`.
async fn seed_recent_completions(db: &Pool<Postgres>, key: &str, n: usize) {
for _ in 0..n {
sqlx::query!(
"INSERT INTO concurrency_key (job_id, key, ended_at) VALUES ($1, $2, NOW())",
Uuid::new_v4(),
key,
)
.execute(db)
.await
.expect("seed completion");
}
}
async fn alert_messages(db: &Pool<Postgres>) -> Vec<String> {
sqlx::query_scalar!("SELECT message FROM alerts WHERE alert_type = 'critical_error'")
.fetch_all(db)
@@ -153,6 +171,28 @@ mod tests {
);
}
/// A gate can be shut with nothing running: the limiter admits on running
/// jobs *plus* those that ended inside concurrency_time_window_s, and a
/// completion clears the job from `concurrency_counter` while still holding
/// the window. Reading the counter alone reports the gate free, and the
/// backlog behind it pages -- the same false alert, one completion later.
#[ignore = "requires database setup - run with --ignored flag"]
#[sqlx::test(migrations = "../migrations")]
async fn time_window_saturated_gate_does_not_alert(db: Pool<Postgres>) {
free_the_lock(&db).await;
configure_alert(&db, 100).await;
seed_gate(&db, GATE_KEY, 1, 1, 0).await;
seed_recent_completions(&db, GATE_KEY, 1).await;
queue_jobs(&db, 200, Some((GATE_KEY, 1))).await;
jobs_waiting_alerts(&db).await;
assert!(
alert_messages(&db).await.is_empty(),
"a gate saturated by completions inside its time window must not raise a critical alert"
);
}
/// The exclusion keys off the gate being *full*, not off the job merely
/// carrying a concurrent_limit -- every such job gets a `concurrency_key`
/// row at push time, so excluding on that row alone would stop this alert