feat(scheduler): record a per-job event log behind --event-log-dir - #2264
Merged
Conversation
…or out dto_build Move the scheduler's REST response types into a new leaf crate, `ballista-history`, and pull the graph-to-DTO construction out of the axum handlers into a pure `api::dto_build` module. Behavior preserving: the same DTOs are produced from the same state, so live REST responses are byte-identical. The existing handler tests cover this, and the helper unit tests move alongside the functions they test. This is the first step toward a history server that replays completed jobs and serves the same `/api/*` responses without a live scheduler. Splitting the DTOs into a serde-only crate lets that server build the identical wire types without depending on the scheduler's live execution graph, and moving construction out of the handlers means it can run against state that did not come from a handler request. `JobResponse::job_id` becomes a `String` rather than `ballista_core::JobId` so the new crate stays serde-only. `JobId` is `#[serde(transparent)]` over `String`, so the JSON is unchanged.
Follow-up cleanups on the extraction: - Collapse the three near-identical ExecutionStage arms in graph_to_query_stages into one destructuring match, dropping the mutable placeholder-zero summary. - Take PlanFormat by value instead of &JobQueryParams, and move PlanFormat into ballista-history. It is part of the wire contract, and the pure builder no longer imports an axum query-param type back out of the handler module. - Inject `now` into graph_to_query_stages rather than reading the clock, so replaying a stored log renders stable elapsed times. - Share percent_complete and min_start_time; use displayable() instead of the longhand DisplayableExecutionPlan::new(). - Drop the dead JobConfig alias and the unused serde_json dev-dependency, make task_status_to_dto private, and remove a duplicated test. - Enable #![warn(missing_docs)] on ballista-history and document the types, matching the other Ballista crates. - Register ballista-history with the release tooling: version bump script, publish order, and crate dependency graph.
The previous wording left it ambiguous whether the history server re-derives responses from stored execution state or replays stored DTOs. It replays them: the scheduler builds each response once against the live graph and writes it to the event log, so byte-identical output is a structural property rather than two implementations agreeing. Also records the consequence, that anything not captured at write time cannot be recovered at replay time.
The crate holds the /api/* wire types, and it has three parties, not one: the scheduler serves them, the web TUI deserializes them, and a future history server will serve replayed copies. Naming it after the history server made it awkward for the TUI, which parses live scheduler responses and today keeps its own duplicate declarations. Renaming it after the contract it defines removes that friction. The event-log schema, writer, and reader can then land as a separate ballista-history crate that depends on this one.
Second step toward the history server (apache#1923), after the wire-type extraction in apache#2256. Adds the durable format and the machinery to read and write it. Nothing in the scheduler calls this yet. - A versioned JSONL schema. JobStart / StageStart / StageEnd / TaskEnd form an incremental timeline; the terminal JobEnd embeds the finished REST responses, so replay re-serves what the scheduler already built rather than re-deriving it. - An async buffered EventLogWriter. All file I/O runs on a background task so the scheduler's event loop never waits on disk. Timeline events are dropped rather than allowed to block when the queue backs up, since losing a progress record beats stalling scheduling. JobEnd is the exception and waits for capacity, because a job missing it is invisible to the history server. - A reader that folds a completed log back into the served payload, skipping malformed lines rather than failing. Restores the JobConfig alias to ballista-api-types, which now has a real consumer in the JobEnd record. TaskEnd names a task rather than a partition: under the multi-partition task model a task owns a slice of partitions, and TaskStatus carries task_id, not partition_id.
A log is written once and may be read years later by a much newer binary, so the guarantee has to run one way: a reader accepts any log whose version is not newer than its own. That is the opposite of BALLISTA_PROTOCOL_VERSION, the strict-equality handshake between scheduler and executor, where both ends are live and upgraded together. The format did not support that yet. Four changes: Self-describing envelope. Every line is now a LogRecord carrying `ev`, `version` and an opaque `data` payload, so a reader can route on the kind and check the version before committing to a shape it may not understand. Previously only JobStart and JobEnd carried a version at all. Stored responses are opaque. JobEnd holds the finished /api/* responses as raw JSON rather than typed structs, plus a small frozen JobIndex for listing. Those responses are ballista-api-types shapes, which change with the live REST contract: partition_id went from u32 to Vec<u32> and TaskStatus::Failed gained a field within one release cycle. Stored typed, either change would have made every older log unreadable, and the job would have silently disappeared. Stored raw, nothing ever parses the inner shape and replay relays the exact bytes. Unreadable is no longer indistinguishable from absent. read_completed_job returned Ok(None) both when a log had no JobEnd and when it had one that could not be parsed, and the loader treated that as "still running" and said nothing. It now returns a ReadError distinguishing an unsupported version from a malformed record. The version is actually checked. A record newer than SCHEMA_VERSION is reported as such rather than skipped. Adds testdata/schema-v1.eventlog, a frozen log replayed by CI on every build. It includes a record kind this build does not know and a stages payload carrying the multi-partition shape, so it exercises both forward-compatibility paths. Verified it fails as intended: a field rename in a stored type breaks all four compatibility tests.
# Conflicts: # Cargo.toml # ballista/api-types/src/dto.rs # dev/release/README.md # dev/release/crate-deps.dot # dev/update_ballista_versions.py
Third step toward the history server (apache#1923). Wires the event-log crate from apache#2260 into the scheduler. Nothing reads these logs yet; the history server that serves them is the next slice. - New --event-log-dir flag, off by default. When unset there is no channel, no background task, no file, and no per-event work beyond one Option check in the event loop. - event_log.rs builds HistoryEvents from execution-graph state, reusing the same api::dto_build builders that back the live REST API, so a job's stored record and its GET /api/job/{id} response are the same bytes for the same graph. - A tee at the top of QueryStageScheduler::on_receive maps JobSubmitted, TaskUpdating, JobFinished, JobRunningFailed and JobCancel onto history events. JobCancel is handled in the tee specifically because the handler below it drops the graph, making that the last point at which a cancelled job can be recorded. Its status is overridden on the typed DTO before anything is serialized, rather than by rewriting the stored JSON, so the payload stays a faithful serialization of one value and can be relayed verbatim. JobPlanningFailed is deliberately absent: it is posted instead of JobSubmitted, so the job has neither an execution graph nor an open log. Failure to build an event costs the job its record, never its execution. A missing graph or a serialization error is logged and skipped. The stage snapshot embedded in JobEnd is rendered as of completed_at rather than the wall clock, so replaying a log is deterministic.
The index exists so the history server can render GET /api/jobs without parsing the stored payloads, but it was missing num_stages, completed_stages and percent_complete, which that response includes. It could not actually serve the list it was there for. Adds the three fields and updates the v1 fixture to match. Doing this before release, while the schema is still unpublished, so the frozen fixture stays a faithful record of what v1 actually looks like.
Contributor
|
a bit late comment on |
milenkovicm
approved these changes
Aug 9, 2026
milenkovicm
left a comment
milenkovicm
left a comment
Contributor
There was a problem hiding this comment.
tested it, it works as expected. thanks @andygrove
two comments for consideration
- make event log use object store abstraction so we can ship them to a remote (object) store
- i've wanted to change job ids for some time now to use some kind of uuid or ulid, i think sorted job id would make sense if we keep job history
Member
Author
great idea about using object store, will file issue |
Member
Author
|
Member
Author
|
Thanks for testing this feature @milenkovicm! |
Contributor
|
i have changed job ids format in #2269 |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
2 participants
Add this suggestion to a batch that can be applied as a single commit.This suggestion is invalid because no changes were made to the code.Suggestions cannot be applied while the pull request is closed.Suggestions cannot be applied while viewing a subset of changes.Only one suggestion per line can be applied in a batch.Add this suggestion to a batch that can be applied as a single commit.Applying suggestions on deleted lines is not supported.You must change the existing code in this line in order to create a valid suggestion.Outdated suggestions cannot be applied.This suggestion has been applied or marked resolved.Suggestions cannot be applied from pending reviews.Suggestions cannot be applied on multi-line comments.Suggestions cannot be applied while the pull request is queued to merge.Suggestion cannot be applied right now. Please check back later.
Which issue does this PR close?
Part of #1923. Third of four PRs splitting up #1925, after #2256 (merged) and #2260.
Rationale for this change
#2260 added the event-log format and the machinery to read and write it, but nothing produces one. This wires it into the scheduler so a running cluster actually emits logs. Nothing reads them yet: the history server that serves them is the last slice.
Splitting it out this way keeps the reviewable question narrow. This is the only slice that touches the scheduler's event loop, so it is the one where the cost of being wrong is a scheduling regression rather than a missing feature. That deserves its own read, rather than being buried in a PR that also introduces a binary and a docs page.
What changes are included in this PR?
A new
--event-log-dirflag, off by default. When it is unset there is no channel, no background task, no file, and no per-event work beyond oneOptioncheck inon_receive. A cluster that does not opt in is unaffected.scheduler_server/event_log.rsbuildsHistoryEvents from execution-graph state. It reusesapi::dto_build, the same builders backing the live REST API, so a job's stored record and itsGET /api/job/{id}response are the same bytes for the same graph. That reuse is the whole reason #2256 extracted those builders out of the axum handlers.A tee at the top of
QueryStageScheduler::on_receivemapsJobSubmitted,TaskUpdating,JobFinished,JobRunningFailedandJobCancelonto history events.Three decisions in there worth review attention:
JobCancelis handled in the tee rather than alongside the other terminal events, because the handler below it drops the graph. That makes the tee the last point at which a cancelled job can be recorded at all. Without it, a cancelled job would never get a terminal record, leaving it invisible to the history server and its file handle open for the life of the process.Running. Patching the serialized payload would have worked, but it would also have broken the property that a stored payload is a faithful serialization of exactly one value, which is what lets the history server relay it verbatim.JobPlanningFailedis deliberately absent. It is posted instead ofJobSubmittedwhensubmit_jobfails, so the job has neither an execution graph nor an open log file. There is nothing to record and nothing to close.Event logging never fails a job. A missing execution graph or a serialization error is logged and skipped. The worst case is a job with no record, never a job that does not run.
The stage snapshot embedded in
JobEndis rendered as ofcompleted_at, not the wall clock, so replaying a log is deterministic rather than dependent on when the writer happened to run.Are there any user-facing changes?
One new opt-in flag,
--event-log-dir <dir>, disabled by default. No behaviour change for clusters that do not set it, no API changes, no configuration defaults altered.Verified locally:
cargo test -p ballista-scheduler --libpasses (331 tests, 7 of them new), clippy is clean with--all-features -D warnings, fmt and taplo are clean, and the--no-default-featurescheck CI runs still passes.What is not here
The end-to-end guarantee.
history_store_serves_byte_identical_json_to_live_schedulerasserts that the history server serves exactly what the live scheduler would for the same graph, but it needsHistoryStore, which arrives with the server in the next slice. It moves there rather than being duplicated or weakened here.So this slice lands with the emission path covered by unit tests on the event builders and a round trip through the real async writer, but the parity claim itself is proved one PR later. Naming that explicitly because it is the one place the split is not clean.
Follow-up