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

fix: measure drift on its own endpoint, not on the schedule's config read

Browse Source
This commit is contained in:
hugocasa
2026-08-28 16:35:15 +02:00
parent 2b0eaf1404
commit b444c4a20c
3 changed files with 85 additions and 49 deletions
Download Patch File Download Diff File Expand all files Collapse all files
backend/windmill-api-schedule/src/lib.rs
+41 -17
View File
@@ -127,6 +127,7 @@ pub fn workspaced_service() -> Router {
.route("/list", get(list_schedule))
.route("/list_with_jobs", get(list_schedule_with_jobs))
.route("/get/{*path}", get(get_schedule))
.route("/interval_drift/{*path}", get(get_interval_drift))
.route("/exists/{*path}", get(exists_schedule))
.route("/create", post(create_schedule))
.route("/update/{*path}", post(edit_schedule))
@@ -1012,6 +1013,17 @@ pub struct IntervalDrift {
pub configured_s: i64,
}
/// What the reader needs on top of the numbers: which way out to offer. A flow
/// already queues its next run when the previous one starts, so its runs are
/// starting late rather than overrunning. Answered from the deployed row, so the
/// editor never has to derive it from the form a draft may be sitting in.
#[derive(Serialize, Deserialize, Debug)]
pub struct IntervalDriftReport {
#[serde(flatten)]
pub drift: IntervalDrift,
pub queues_next_run_at_start: bool,
}
/// Consecutive slots a schedule has to miss before we call it drift: one slow
/// run is not a change of cadence.
const DRIFT_MIN_MISSED_SLOTS: usize = 3;
@@ -1188,13 +1200,6 @@ async fn get_schedule(
let schedule_o = windmill_queue::schedule::get_schedule_opt(&mut *tx, &w_id, path).await?;
tx.commit().await?;
let deployed = match schedule_o {
Some(schedule) => Some(ScheduleWithIntervalDrift {
interval_drift: fetch_interval_drift(&db, &w_id, &schedule).await?,
schedule,
}),
None => None,
};
let overlay = overlay_or_draft_only(
&db,
&w_id,
@@ -1202,21 +1207,35 @@ async fn get_schedule(
UserDraftItemKind::TriggerSchedule,
path,
q.get_draft,
deployed,
schedule_o,
|| Error::NotFound(format!("Schedule not found at path {path}")),
)
.await?;
Ok(Json(overlay))
}
/// The schedule editor reads a schedule through this, so its drift surfaces
/// next to the cron expression that is running behind.
#[derive(Serialize)]
struct ScheduleWithIntervalDrift {
#[serde(flatten)]
schedule: Schedule,
#[serde(skip_serializing_if = "Option::is_none")]
interval_drift: Option<IntervalDrift>,
/// Measured, not configured: kept off `get_schedule`, whose response is the
/// schedule's configuration and is diffed for deploys and read by exporters that
/// fan out over every schedule in a workspace.
async fn get_interval_drift(
authed: ApiAuthed,
Extension(user_db): Extension<UserDB>,
Extension(db): Extension<DB>,
Path((w_id, path)): Path<(String, StripPath)>,
) -> JsonResult<Option<IntervalDriftReport>> {
let path = path.to_path();
check_scopes(&authed, || format!("schedules:read:{}", path))?;
let mut tx = user_db.begin(&authed).await?;
let schedule = windmill_queue::schedule::get_schedule_opt(&mut *tx, &w_id, path).await?;
tx.commit().await?;
let schedule = not_found_if_none(schedule, "Schedule", path)?;
let report = fetch_interval_drift(&db, &w_id, &schedule)
.await?
.map(|drift| IntervalDriftReport {
drift,
queues_next_run_at_start: schedule.is_flow || schedule.dynamic_skip.is_some(),
});
Ok(Json(report))
}
async fn fetch_interval_drift(
@@ -1227,7 +1246,12 @@ async fn fetch_interval_drift(
// A schedule that is off is not running behind, it is not running. Nor is one
// told to skip: a skip handler and `no_flow_overlap` both exist to drop runs, so
// for those two the cadence the cron asks for was never the promise.
if !schedule.enabled || schedule.no_flow_overlap || schedule.dynamic_skip.is_some() {
// `no_flow_overlap` is only consulted by the flow runtime, so on a plain script
// schedule it is inert and must not suppress anything.
if !schedule.enabled
|| (schedule.is_flow && schedule.no_flow_overlap)
|| schedule.dynamic_skip.is_some()
{
return Ok(None);
}
// Query plan: `(workspace_id, runnable_path, created_at DESC)` index, hence the
backend/windmill-api/openapi.yaml
+28 -4
View File
@@ -17035,10 +17035,27 @@ paths:
allOf:
- $ref: "#/components/schemas/Schedule"
- $ref: "#/components/schemas/UserDraftOverlay"
- type: object
properties:
interval_drift:
$ref: "#/components/schemas/ScheduleIntervalDrift"
/w/{workspace}/schedules/interval_drift/{path}:
get:
summary: get how far a schedule's runs have drifted from its cron
operationId: getScheduleIntervalDrift
tags:
- schedule
parameters:
- $ref: "#/components/parameters/WorkspaceId"
- $ref: "#/components/parameters/Path"
responses:
"200":
description: >-
The measurement, or null when the schedule is keeping up, is
disabled, is configured to skip runs, or has not run often enough
under its current cron to tell
content:
application/json:
schema:
$ref: "#/components/schemas/ScheduleIntervalDrift"
nullable: true
/w/{workspace}/schedules/exists/{path}:
get:
@@ -29792,9 +29809,16 @@ components:
configured_s:
type: integer
description: Seconds the cron expression puts between those same two slots
queues_next_run_at_start:
type: boolean
description: >-
True when the schedule queues its next run as the previous one
starts (a flow, or a script with a skip handler), which means its
runs are starting late rather than overrunning
required:
- effective_s
- configured_s
- queues_next_run_at_start
ErrorHandler:
type: string
frontend/src/lib/components/triggers/schedules/ScheduleEditorInner.svelte
+16 -28
View File
@@ -131,20 +131,18 @@
let labels: string[] | undefined = $state(undefined)
let description = $state('')
let no_flow_overlap = $state(false)
// Measured on the deployed schedule, so it is read back rather than edited, and
// so is what the warning says about the runs: the form may be showing a draft
// that is not what has been running.
let deployedDrift:
| { interval: ScheduleIntervalDrift; queuesNextRunAtStart: boolean }
| undefined = $state(undefined)
// Measured rather than configured, so it comes from its own read of the deployed
// schedule: everything else on this form may be a draft that has never run.
let deployedDrift: ScheduleIntervalDrift | undefined = $state(undefined)
// A flow queues its next run when the previous one starts, where a script queues it
// once the previous one has finished, and the two have different ways out. Only
// those two reach here: a schedule set to skip runs is not measured at all.
function readDeployedDrift(deployed: Record<string, any>) {
deployedDrift = deployed.interval_drift
? { interval: deployed.interval_drift, queuesNextRunAtStart: !!deployed.is_flow }
: undefined
async function readDeployedDrift(path: string) {
try {
deployedDrift =
(await ScheduleService.getScheduleIntervalDrift({ workspace: wsId ?? '', path })) ??
undefined
} catch {
deployedDrift = undefined
}
}
let tag: string | undefined = $state(undefined)
let validCRON = $state(true)
@@ -536,9 +534,7 @@
getDraft: true
})
const { draft: draftFromBackend, ...deployedSchedule } = s as any
// Read here and nowhere else: the overlay below merges the draft over these
// same fields, so anything downstream of it would diagnose the draft.
readDeployedDrift(deployedSchedule)
readDeployedDrift(initialPath)
await loadScheduleCfg(deployedSchedule)
return {
overlay: draftFromBackend
@@ -751,15 +747,7 @@
}
// The measurement describes the deployed schedule, which just changed:
// disabling drops it, and enabling brings back what the runs still show.
try {
const deployed = await ScheduleService.getSchedule({
workspace: wsId ?? '',
path: initialPath
})
readDeployedDrift(deployed as any)
} catch {
deployedDrift = undefined
}
await readDeployedDrift(initialPath)
sendUserToast(`${nEnabled ? 'enabled' : 'disabled'} schedule ${initialPath}`)
onUpdate?.(initialPath)
}
@@ -963,11 +951,11 @@
{#if deployedDrift}
<Alert type="warning" size="xs" title="Running less often than configured">
This schedule is running about every {msToReadableTime(
deployedDrift.interval.effective_s * 1000
)} instead of every {msToReadableTime(deployedDrift.interval.configured_s * 1000)}:
deployedDrift.effective_s * 1000
)} instead of every {msToReadableTime(deployedDrift.configured_s * 1000)}:
each of the last runs was queued too late for the slot that would have kept the
cadence.
{#if deployedDrift.queuesNextRunAtStart}
{#if deployedDrift.queues_next_run_at_start}
Its next run is already queued when the previous one starts, so the runs are
starting late rather than overrunning: look at worker capacity or a concurrency
limit.