mirror of
https://github.com/windmill-labs/windmill.git
synced 2026-10-29 00:03:09 +00:00
fix(queue): duration-weighted workspace fairness signal (#9329)
* fix(queue): bump EE ref to include worker_ping fairness signal The current ee-repo-ref.txt pointed to 31cda7c (an unrelated merge commit on the asset-graph-view-ee branch) instead of ddc9e80, which contains the workspace-fairness fix that switches the active-share signal from v2_job_queue.running=true to worker_ping. As a result cloud was still computing overload off the legacy signal, so a workspace with many in-flight/suspended flows (lancom01-prod, with 799 suspended flows × 3 v2_job_queue bookkeeping rows each = 2397 running-true rows) was flagged as 95% of cluster activity despite consuming zero worker slots. Bumping to ddc9e80 picks up the worker_ping-based signal, which naturally excludes (a) suspended jobs (no worker pinging them), (b) zombie running-rows from dead workers, and (c) flow/flownode orchestration rows that never run on a worker in the first place. * test(queue): seed v2_job rows + realistic durations for fairness helpers The new duration-weighted fairness algorithm joins v2_job_queue and v2_job_completed to v2_job for the `kind` filter (excluding flow bookkeeping) and reads `duration_ms` for the completed contribution. Update the test helpers to mirror that schema: * `insert_completed` now inserts a matching v2_job row (kind=script) and writes `duration_ms = 1000` with a 1-second [started_at, completed_at] interval, so each completed row contributes ~1 worker-second when fully inside the refresh window. * `insert_queued` likewise pre-inserts v2_job, sets `started_at` to NOW() - 1s when running=true (so running rows contribute ~1 worker-second by the time the refresh runs), and seeds v2_job_runtime.ping so the running side accrues real-time worker seconds (the algorithm bounds end-of-interval by ping). The zombie/suspended insert helpers are intentionally left without v2_job rows — the new algorithm's INNER JOIN excludes them, so they still correctly contribute zero worker-seconds. * chore(queue): bump EE ref to duration-weighted fairness algorithm Companion to windmill-ee-private#<TBD>: switch the EE workspace fairness aggregation from a count-based UNION (worker_ping snapshot + v2_job_completed count) to a worker-seconds aggregation sourced directly from v2_job_queue and v2_job_completed, with kind/suspend filters mirroring handle_zombie_jobs and per-row defenses against zombie inflation on both halves. * chore(queue): bump EE ref for fairness perf fix (inline window_start) * chore(queue): bump EE ref for fairness perf rewrite (driver-side flip) * update ee ref
This commit is contained in:
@@ -1 +1 @@
|
||||
31cda7cea811108e5004842ec5b5467ca48181cd
|
||||
327d23f7438968a21bac9fd42e7f6f027c61477c
|
||||
@@ -79,15 +79,28 @@ async fn create_workspace(db: &Pool<Postgres>, id: &str) {
|
||||
.unwrap();
|
||||
}
|
||||
|
||||
/// Insert `n` completed jobs for `workspace_id`, each ending `secs_ago`
|
||||
/// seconds in the past with a 1-second wall-clock duration. The fairness
|
||||
/// algorithm weights contributions by `duration_ms` (clamped to the window),
|
||||
/// so each job contributes ~1 worker-second when fully inside the window.
|
||||
async fn insert_completed(db: &Pool<Postgres>, workspace_id: &str, n: usize, secs_ago: i32) {
|
||||
for _ in 0..n {
|
||||
let id: Uuid = sqlx::query_scalar(
|
||||
"INSERT INTO v2_job (id, workspace_id, kind)
|
||||
VALUES (gen_random_uuid(), $1, 'script'::job_kind) RETURNING id",
|
||||
)
|
||||
.bind(workspace_id)
|
||||
.fetch_one(db)
|
||||
.await
|
||||
.unwrap();
|
||||
sqlx::query(
|
||||
"INSERT INTO v2_job_completed (id, workspace_id, duration_ms, status,
|
||||
started_at, completed_at)
|
||||
VALUES (gen_random_uuid(), $1, 1, 'success'::job_status,
|
||||
NOW() - make_interval(secs => $2::int),
|
||||
NOW() - make_interval(secs => $2::int))",
|
||||
VALUES ($1, $2, 1000, 'success'::job_status,
|
||||
NOW() - make_interval(secs => ($3::int + 1)),
|
||||
NOW() - make_interval(secs => $3::int))",
|
||||
)
|
||||
.bind(id)
|
||||
.bind(workspace_id)
|
||||
.bind(secs_ago)
|
||||
.execute(db)
|
||||
@@ -106,20 +119,42 @@ async fn insert_queued(
|
||||
let mut ids = Vec::with_capacity(n);
|
||||
for _ in 0..n {
|
||||
let id: Uuid = sqlx::query_scalar(
|
||||
"INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, running, tag)
|
||||
VALUES (gen_random_uuid(), $1, NOW(), $2, $3) RETURNING id",
|
||||
"INSERT INTO v2_job (id, workspace_id, kind, tag)
|
||||
VALUES (gen_random_uuid(), $1, 'script'::job_kind, $2) RETURNING id",
|
||||
)
|
||||
.bind(workspace_id)
|
||||
.bind(running)
|
||||
.bind(tag)
|
||||
.fetch_one(db)
|
||||
.await
|
||||
.unwrap();
|
||||
// Running jobs need a `started_at` for the fairness algorithm to
|
||||
// compute a positive elapsed-time contribution. Backdate by 1s so
|
||||
// each running row contributes ~1 worker-second by the time the
|
||||
// refresh runs, matching the `insert_completed` scale.
|
||||
sqlx::query(
|
||||
"INSERT INTO v2_job_queue (id, workspace_id, scheduled_for, running, tag, started_at)
|
||||
VALUES ($1, $2, NOW(), $3, $4,
|
||||
CASE WHEN $3 THEN NOW() - interval '1 second' ELSE NULL END)",
|
||||
)
|
||||
.bind(id)
|
||||
.bind(workspace_id)
|
||||
.bind(running)
|
||||
.bind(tag)
|
||||
.execute(db)
|
||||
.await
|
||||
.unwrap();
|
||||
if running {
|
||||
// The fairness algorithm reads slot occupancy from `worker_ping`,
|
||||
// not from `v2_job_queue.running = true`. Mirror each running row
|
||||
// with a paired live `worker_ping` so unit tests see the same
|
||||
// signal a real cluster would.
|
||||
// The fairness algorithm bounds the running contribution by the
|
||||
// per-job `v2_job_runtime.ping`. Insert a fresh ping so each
|
||||
// running row accrues real-time worker-seconds.
|
||||
sqlx::query(
|
||||
"INSERT INTO v2_job_runtime (id, ping) VALUES ($1, NOW())
|
||||
ON CONFLICT (id) DO UPDATE SET ping = NOW()",
|
||||
)
|
||||
.bind(id)
|
||||
.execute(db)
|
||||
.await
|
||||
.unwrap();
|
||||
insert_live_worker_ping(db, workspace_id, id).await;
|
||||
}
|
||||
ids.push(id);
|
||||
|
||||
Reference in New Issue
Block a user