mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-08-19 00:02:03 +00:00
refactor(alerts): read the limiter's gate mark instead of re-deriving it
The alert reconstructed the limiter's admission test in SQL to tell a gate-parked backlog from worker starvation. That test spans the running counter, completions inside the time window, per-version setting fallbacks and three bypass paths, and nothing linked the two copies, so each divergence surfaced as a fresh false alert or a suppressed one. Adds v2_job_queue.concurrency_gated, which the limiter sets when it re-queues a job it could not admit; the alert counts unmarked jobs. The query drops to a single scan with no joins. Emitter and limiter changes live in windmill-ee-private (see ee-repo-ref.txt bump).
This commit is contained in:
Ruben Fiszel
+15
@@ -0,0 +1,15 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "\n WITH ping AS (\n UPDATE v2_job_runtime SET ping = null WHERE id = $2\n )\n UPDATE v2_job_queue SET\n running = false,\n started_at = null,\n scheduled_for = $1,\n concurrency_gated = true\n WHERE id = $2",
|
||||
"describe": {
|
||||
"columns": [],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Timestamptz",
|
||||
"Uuid"
|
||||
]
|
||||
},
|
||||
"nullable": []
|
||||
},
|
||||
"hash": "1d2bbf22936979a5413ddfb2ee0a995f7988c40f6582e1673f892d42bbfe4405"
|
||||
}
|
||||
+41
@@ -0,0 +1,41 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "\n SELECT COUNT(*) FILTER (WHERE concurrency_gated IS NOT TRUE) AS count,\n COUNT(*) FILTER (WHERE concurrency_gated) AS gated_count,\n MIN(scheduled_for) FILTER (WHERE concurrency_gated IS NOT TRUE)\n AS oldest_job,\n MIN(scheduled_for) AS oldest_any\n FROM v2_job_queue\n WHERE tag = $1\n AND scheduled_for <= NOW() - $2::interval\n AND running = false\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"
|
||||
},
|
||||
{
|
||||
"ordinal": 3,
|
||||
"name": "oldest_any",
|
||||
"type_info": "Timestamptz"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Text",
|
||||
"Interval"
|
||||
]
|
||||
},
|
||||
"nullable": [
|
||||
null,
|
||||
null,
|
||||
null,
|
||||
null
|
||||
]
|
||||
},
|
||||
"hash": "23dee86f414303a6a9332d7c49b1e1eb6d25dc58bb892d7dc1f07d8ab22cf6a5"
|
||||
}
|
||||
-41
@@ -1,41 +0,0 @@
|
||||
{
|
||||
"db_name": "PostgreSQL",
|
||||
"query": "\n SELECT COUNT(*) AS count,\n MIN(scheduled_for) AS oldest_job,\n COUNT(*) FILTER (WHERE canceled_by IS NOT NULL) AS canceled_count,\n MIN(scheduled_for) FILTER (WHERE canceled_by IS NOT NULL) AS canceled_oldest\n FROM v2_job_queue\n WHERE tag = $1\n AND scheduled_for <= NOW() - $2::interval\n AND running = false\n ",
|
||||
"describe": {
|
||||
"columns": [
|
||||
{
|
||||
"ordinal": 0,
|
||||
"name": "count",
|
||||
"type_info": "Int8"
|
||||
},
|
||||
{
|
||||
"ordinal": 1,
|
||||
"name": "oldest_job",
|
||||
"type_info": "Timestamptz"
|
||||
},
|
||||
{
|
||||
"ordinal": 2,
|
||||
"name": "canceled_count",
|
||||
"type_info": "Int8"
|
||||
},
|
||||
{
|
||||
"ordinal": 3,
|
||||
"name": "canceled_oldest",
|
||||
"type_info": "Timestamptz"
|
||||
}
|
||||
],
|
||||
"parameters": {
|
||||
"Left": [
|
||||
"Text",
|
||||
"Interval"
|
||||
]
|
||||
},
|
||||
"nullable": [
|
||||
null,
|
||||
null,
|
||||
null,
|
||||
null
|
||||
]
|
||||
},
|
||||
"hash": "3f9dd35d19e6badc92397fdb476d39ffadf1ff97931a9506772f76c3f36bd2c7"
|
||||
}
|
||||
-35
@@ -1,35 +0,0 @@
|
||||
{
|
||||
"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 AND q.canceled_by IS NULL\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 w.concurrency_key <> '' 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": "dbacbbbe27d1589dd26d6bcbd131a4ce32944cb2b36323117bbe92460e88b767"
|
||||
}
|
||||
@@ -1 +1 @@
|
||||
d54657fe4641a910cb2396de9cf83e850f118d2c
|
||||
3984a0f87d24770f860a315061a26c9e53368b2e
|
||||
@@ -0,0 +1 @@
|
||||
ALTER TABLE v2_job_queue DROP COLUMN IF EXISTS concurrency_gated;
|
||||
@@ -0,0 +1,7 @@
|
||||
-- Set by the concurrency limiter when it re-queues a job it could not admit.
|
||||
-- Records that the job is parked behind its own concurrency gate rather than
|
||||
-- waiting for a worker, which the queue row cannot otherwise show: the
|
||||
-- limiter's admission test spans concurrency_counter, the completed-window rows
|
||||
-- in concurrency_key, per-version setting fallbacks and several bypass paths.
|
||||
-- Nullable so adding it does not rewrite the table.
|
||||
ALTER TABLE v2_job_queue ADD COLUMN IF NOT EXISTS concurrency_gated BOOLEAN;
|
||||
@@ -174,7 +174,7 @@ usr_to_group: workspace_id(char), group_(char), usr(char)
|
||||
v2_job: id(uuid), raw_code(text), raw_lock(text), raw_flow(jsonb), tag(char), workspace_id(char), created_at(ts), created_by(char), permissioned_as(char), permissioned_as_email(char), kind(job_kind), runnable_id(bigint), runnable_path(char), parent_job(uuid), root_job(uuid), script_lang(script_lang), script_entrypoint_override(char), flow_step(int), flow_step_id(char), flow_innermost_root_job(uuid), trigger(char), trigger_kind(job_trigger_kind), same_worker(bool), visible_to_owner(bool), concurrent_limit(int), concurrency_time_window_s(int), cache_ttl(int), timeout(int), priority(smallint), preprocessed(bool), args(jsonb), labels(text[]), pre_run_error(text)
|
||||
v2_job_completed: id(uuid), workspace_id(char), duration_ms(bigint), result(jsonb), deleted(bool), canceled_by(char), canceled_reason(text), flow_status(jsonb), started_at(ts), memory_peak(int), status(job_status), completed_at(ts), worker(char), workflow_as_code_status(jsonb), result_columns(text[]), retries(uuid[]), extras(jsonb)
|
||||
v2_job_debounce_batch: id(uuid), debounce_batch(bigint)
|
||||
v2_job_queue: id(uuid), workspace_id(char), created_at(ts), started_at(ts), scheduled_for(ts), running(bool), canceled_by(char), canceled_reason(text), suspend(int), suspend_until(ts), tag(char), priority(smallint), worker(char), extras(jsonb), cache_ignore_s3_path(bool), runnable_settings_handle(bigint)
|
||||
v2_job_queue: id(uuid), workspace_id(char), created_at(ts), started_at(ts), scheduled_for(ts), running(bool), canceled_by(char), canceled_reason(text), suspend(int), suspend_until(ts), tag(char), priority(smallint), worker(char), extras(jsonb), cache_ignore_s3_path(bool), runnable_settings_handle(bigint), concurrency_gated(bool)
|
||||
v2_job_runtime: id(uuid), ping(ts), memory_peak(int)
|
||||
FK: (id) -> v2_job_queue(id)
|
||||
v2_job_status: id(uuid), flow_status(jsonb), flow_leaf_jobs(jsonb), workflow_as_code_status(jsonb)
|
||||
|
||||
@@ -19,7 +19,6 @@ mod tests {
|
||||
use windmill_common::ee::jobs_waiting_alerts;
|
||||
|
||||
const TAG: &str = "python3";
|
||||
const GATE_KEY: &str = "test-workspace/u/user/gated_script";
|
||||
|
||||
/// `jobs_waiting_alerts` bails out unless it can take the shared lock, and a
|
||||
/// freshly inserted lock row is never old enough to be claimed on the first
|
||||
@@ -54,108 +53,20 @@ mod tests {
|
||||
.expect("seed alert config");
|
||||
}
|
||||
|
||||
/// 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.
|
||||
///
|
||||
/// 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!({})))
|
||||
.collect::<serde_json::Map<_, _>>()
|
||||
.into();
|
||||
|
||||
sqlx::query!(
|
||||
"INSERT INTO concurrency_settings (hash, concurrency_key, concurrent_limit, concurrency_time_window_s)
|
||||
VALUES ($1, $2, $3, 600)",
|
||||
handle,
|
||||
key,
|
||||
limit,
|
||||
)
|
||||
.execute(db)
|
||||
.await
|
||||
.expect("seed concurrency settings");
|
||||
|
||||
sqlx::query!(
|
||||
"INSERT INTO runnable_settings (hash, concurrency_settings) VALUES ($1, $1)",
|
||||
handle,
|
||||
)
|
||||
.execute(db)
|
||||
.await
|
||||
.expect("seed runnable settings");
|
||||
|
||||
sqlx::query!(
|
||||
"INSERT INTO concurrency_counter (concurrency_id, job_uuids) VALUES ($1, $2)",
|
||||
key,
|
||||
job_uuids,
|
||||
)
|
||||
.execute(db)
|
||||
.await
|
||||
.expect("seed concurrency counter");
|
||||
}
|
||||
|
||||
/// Queue `n` jobs on TAG that have been due for 5 minutes. When `gate` is
|
||||
/// set, each job also gets the `concurrency_key` row that
|
||||
/// `insert_concurrency_key` writes at push time for any runnable carrying a
|
||||
/// concurrent_limit, plus the settings handle that carries the limit.
|
||||
async fn queue_jobs(db: &Pool<Postgres>, n: usize, gate: Option<(&str, i64)>) {
|
||||
queue_jobs_inner(db, n, gate, None).await
|
||||
}
|
||||
|
||||
/// Soft-cancelled jobs stay queued until a worker picks them up, and the
|
||||
/// pull path skips concurrency limiting for them entirely.
|
||||
async fn queue_canceled_jobs(db: &Pool<Postgres>, n: usize, gate: Option<(&str, i64)>) {
|
||||
queue_jobs_inner(db, n, gate, Some("canceller")).await
|
||||
}
|
||||
|
||||
async fn queue_jobs_inner(
|
||||
db: &Pool<Postgres>,
|
||||
n: usize,
|
||||
gate: Option<(&str, i64)>,
|
||||
canceled_by: Option<&str>,
|
||||
) {
|
||||
/// Queue `n` jobs on TAG that have been due for 5 minutes. `gated` mirrors
|
||||
/// what the limiter writes when it re-queues a job it could not admit.
|
||||
async fn queue_jobs(db: &Pool<Postgres>, n: usize, gated: bool) {
|
||||
for _ in 0..n {
|
||||
let id = Uuid::new_v4();
|
||||
sqlx::query!(
|
||||
"INSERT INTO v2_job_queue (id, workspace_id, tag, running, scheduled_for, runnable_settings_handle, canceled_by)
|
||||
VALUES ($1, 'test-workspace', $2, false, NOW() - INTERVAL '5 minutes', $3, $4)",
|
||||
id,
|
||||
"INSERT INTO v2_job_queue (id, workspace_id, tag, running, scheduled_for, concurrency_gated)
|
||||
VALUES ($1, 'test-workspace', $2, false, NOW() - INTERVAL '5 minutes', $3)",
|
||||
Uuid::new_v4(),
|
||||
TAG,
|
||||
gate.map(|(_, handle)| handle),
|
||||
canceled_by,
|
||||
gated.then_some(true),
|
||||
)
|
||||
.execute(db)
|
||||
.await
|
||||
.expect("queue job");
|
||||
|
||||
if let Some((key, _)) = gate {
|
||||
sqlx::query!(
|
||||
"INSERT INTO concurrency_key (job_id, key) VALUES ($1, $2)",
|
||||
id,
|
||||
key,
|
||||
)
|
||||
.execute(db)
|
||||
.await
|
||||
.expect("gate job");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
/// 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");
|
||||
}
|
||||
}
|
||||
|
||||
@@ -166,7 +77,7 @@ mod tests {
|
||||
.expect("read alerts")
|
||||
}
|
||||
|
||||
/// Jobs parked by their own workspace's concurrency gate must not page
|
||||
/// Jobs the limiter parked behind their own concurrency gate must not page
|
||||
/// on-call: they are waiting by design, not for want of a worker, and no
|
||||
/// operator action drains them. Their re-queue timestamps mature in bursts,
|
||||
/// so counting them produces a stream of alerts whose count swings wildly
|
||||
@@ -176,101 +87,26 @@ mod tests {
|
||||
async fn concurrency_gated_backlog_does_not_alert(db: Pool<Postgres>) {
|
||||
free_the_lock(&db).await;
|
||||
configure_alert(&db, 100).await;
|
||||
seed_gate(&db, GATE_KEY, 1, 1, 1).await;
|
||||
queue_jobs(&db, 200, Some((GATE_KEY, 1))).await;
|
||||
queue_jobs(&db, 200, true).await;
|
||||
|
||||
jobs_waiting_alerts(&db).await;
|
||||
|
||||
assert!(
|
||||
alert_messages(&db).await.is_empty(),
|
||||
"a backlog held entirely behind a saturated concurrency gate must not raise a critical alert"
|
||||
"a backlog the limiter parked behind a concurrency gate must not raise a critical alert"
|
||||
);
|
||||
}
|
||||
|
||||
/// 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"
|
||||
);
|
||||
}
|
||||
|
||||
/// Cancellation is a limiter bypass too: the pull path hands a job with
|
||||
/// `canceled_by` set straight to a worker without consulting the counter.
|
||||
/// Such jobs stay queued until a worker processes them, so a worker outage
|
||||
/// piles them up behind whatever key they carry -- and classifying them off
|
||||
/// that key hides an unbounded cancelled backlog from the alert.
|
||||
#[ignore = "requires database setup - run with --ignored flag"]
|
||||
#[sqlx::test(migrations = "../migrations")]
|
||||
async fn canceled_jobs_behind_saturated_gate_still_alert(db: Pool<Postgres>) {
|
||||
free_the_lock(&db).await;
|
||||
configure_alert(&db, 100).await;
|
||||
seed_gate(&db, GATE_KEY, 1, 1, 1).await;
|
||||
queue_canceled_jobs(&db, 200, Some((GATE_KEY, 1))).await;
|
||||
|
||||
jobs_waiting_alerts(&db).await;
|
||||
|
||||
let messages = alert_messages(&db).await;
|
||||
assert_eq!(
|
||||
messages.len(),
|
||||
1,
|
||||
"soft-cancelled jobs bypass the limiter and must still alert, got {messages:?}"
|
||||
);
|
||||
}
|
||||
|
||||
/// An empty concurrency key is a limiter bypass: `apply_concurrency_limit`
|
||||
/// admits unconditionally instead of consulting the counter, so these jobs
|
||||
/// are runnable and any backlog of them is genuine capacity starvation.
|
||||
/// Classifying them off the counter alone would read the gate as full and
|
||||
/// silence the alert during a real outage.
|
||||
#[ignore = "requires database setup - run with --ignored flag"]
|
||||
#[sqlx::test(migrations = "../migrations")]
|
||||
async fn bypassed_empty_key_still_alerts(db: Pool<Postgres>) {
|
||||
free_the_lock(&db).await;
|
||||
configure_alert(&db, 100).await;
|
||||
seed_gate(&db, "", 1, 1, 1).await;
|
||||
queue_jobs(&db, 200, Some(("", 1))).await;
|
||||
|
||||
jobs_waiting_alerts(&db).await;
|
||||
|
||||
let messages = alert_messages(&db).await;
|
||||
assert_eq!(
|
||||
messages.len(),
|
||||
1,
|
||||
"jobs whose concurrency key is empty bypass the limiter and must still \
|
||||
alert, got {messages:?}"
|
||||
);
|
||||
}
|
||||
|
||||
/// 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
|
||||
/// from ever firing for a rate-limited runnable. Half the backlog here sits
|
||||
/// under a gate with free slots: it is starving for workers like the plain
|
||||
/// jobs beside it and must be counted. Excluding it drops the count under
|
||||
/// the threshold and the alert goes silent.
|
||||
/// The complement: jobs no gate is holding are waiting on capacity and must
|
||||
/// page, counted apart from the gated ones so the number can be reconciled
|
||||
/// against the much larger raw queue depth.
|
||||
#[ignore = "requires database setup - run with --ignored flag"]
|
||||
#[sqlx::test(migrations = "../migrations")]
|
||||
async fn jobs_waiting_on_capacity_still_alert(db: Pool<Postgres>) {
|
||||
free_the_lock(&db).await;
|
||||
configure_alert(&db, 150).await;
|
||||
seed_gate(&db, GATE_KEY, 1, 10, 2).await;
|
||||
queue_jobs(&db, 100, None).await;
|
||||
queue_jobs(&db, 100, Some((GATE_KEY, 1))).await;
|
||||
configure_alert(&db, 100).await;
|
||||
queue_jobs(&db, 150, false).await;
|
||||
queue_jobs(&db, 200, true).await;
|
||||
|
||||
jobs_waiting_alerts(&db).await;
|
||||
|
||||
@@ -280,9 +116,14 @@ mod tests {
|
||||
1,
|
||||
"expected exactly one critical alert, got {messages:?}"
|
||||
);
|
||||
assert!(
|
||||
messages[0].contains("150"),
|
||||
"alert must count only the jobs actually waiting on capacity: {}",
|
||||
messages[0]
|
||||
);
|
||||
assert!(
|
||||
messages[0].contains("200"),
|
||||
"every job waiting on capacity must be counted, gated or not: {}",
|
||||
"alert must report the gated jobs it excluded: {}",
|
||||
messages[0]
|
||||
);
|
||||
}
|
||||
|
||||
Reference in New Issue
Block a user