1925: feat: event log and history server (Spark History Server equivalent) - #82
1925: feat: event log and history server (Spark History Server equivalent)#82martin-augment wants to merge 14 commits into
Conversation
Move the scheduler REST API's JobResponse/TaskSummary/TaskStatus/Percentiles/ QueryStageSummary/QueryStagesResponse types onto the ballista-history crate's shared DTOs, and extract the graph-to-DTO construction logic that previously lived inline in the handlers into pub(crate) builder functions in a new api::dto_build module. Handlers become thin wrappers over these builders, which a later event-log writer will also call. Behavior is unchanged; the existing handler unit tests move to dto_build.rs alongside the moved helper functions they exercise, plus one new plan-string assertion test.
Emit HistoryEvents from the QueryStageScheduler event loop:
JobSubmitted -> JobStart, TaskUpdating -> TaskEnd (one per finished
task), and JobFinished/JobRunningFailed -> JobEnd. Event builders live
in a new scheduler_server::event_log module and reuse the same
dto_build DTOs the REST API serves, so a job's JobEnd event matches
its live GET /api/job/{id} response. The EventLogWriter is constructed
from SchedulerConfig::event_log_dir and threaded into
QueryStageScheduler; emission is best-effort and a no-op when
event_log_dir is unset.
… missing jobs
- HistoryStore::load now skips unreadable/corrupt .eventlog files with a
warning instead of failing the whole load, so one bad log can't hide
every other completed job.
- Job endpoints (/api/job/{id}, /stages, /config, /dot) now return 404
for unknown job ids instead of 200 with a null body, matching the
live scheduler's behavior.
- Add /api/state with a static payload matching the shape the TUI
deserializes at startup.
- Strengthen router tests: assert stage-response body contents, cover
the corrupt-eventlog skip path, and cover the 404 behavior.
…e scheduler Add an end-to-end DTO parity test that builds live JobResponse/ QueryStagesResponse DTOs, writes the same JobEnd event through the real EventLogWriter, reloads it via HistoryStore::load, and asserts the serialized JSON matches exactly -- proving the history server would serve the same data a running scheduler produced.
EventLogWriter::append enqueues via non-blocking try_send, which drops the event when the channel is saturated. Using it for the terminal JobEnd event meant a completed job's log could end up with JobStart and TaskEnds but no JobEnd, making read_completed_job return None and the job silently disappear from the history server. Add append_final, which awaits channel capacity instead of dropping, and finish_job, which flushes and then closes the per-job file handle (also fixing fd accumulation). Wire both into the JobFinished and JobRunningFailed on_receive arms in place of append + flush_job.
…-api All ballista_history usage in the scheduler is behind the rest-api feature (DTO builders, event-log wiring, history module). Gating the dependency on rest-api keeps it out of the graph for consumers that build the scheduler with default-features = false (e.g. pyballista), and out of non-rest-api builds.
serde_json is only used in production by the rest-api-gated history module (all other uses are in tests, covered by the dev-dependency). Gating it keeps it out of the dependency graph for default-features = false consumers such as pyballista, so python/Cargo.lock stays in sync.
|
Warning Review limit reached
Next review available in: 13 minutes Enable usage-based reviews in Billing to review now. Otherwise, wait until the next included review is available. How can I continue?After more reviews become available, a review can be triggered using the To avoid repeated limits, reduce automatic review volume by pausing incremental auto-reviews earlier, using label-based review opt-in, excluding WIP or generated PR titles, or requesting reviews manually when the PR is ready. If your team needs uninterrupted high-volume reviews, an organization admin can enable usage-based reviews. How do review limits work?CodeRabbit enforces per-developer PR review limits for each organization. Most developers receive the normal plan review availability. For paid Pro and Pro+ PR reviews, CodeRabbit uses adaptive limits for sustained high-volume activity. When a developer's recent PR review activity reaches the 95th percentile or higher among CodeRabbit users, additional reviews become available more gradually as earlier reviews age out of the rolling window. Please refer docs for additional details. Review details⚙️ Run configurationConfiguration used: Organization UI Review profile: CHILL Plan: Pro Run ID: ⛔ Files ignored due to path filters (1)
📒 Files selected for processing (23)
✨ Finishing Touches🧪 Generate unit tests (beta)
Thanks for using CodeRabbit! It's free for OSS, and your support helps us grow. If you like it, consider giving us a shout-out. Comment |
![gemini-code-assist[bot]](https://avatars.githubusercontent.com/in/956858?s=60&v=4){kind=small}
There was a problem hiding this comment.
Code Review
This pull request introduces a new ballista-history crate and a standalone ballista-history-server binary to support recording and replaying job execution histories in Ballista. It implements an asynchronous, buffered event-log writer (EventLogWriter) and a reader (read_completed_job) to persist scheduler events to disk as JSONL and serve them via /api/* endpoints matching the live scheduler. The review feedback highlights several critical improvements: handling the JobCancel event in QueryStageScheduler to prevent file descriptor leaks and ensure cancelled jobs are properly recorded; guarding against division by zero in dto_build.rs when num_stages is zero; optimizing the file handle lookup in EventLogWriter using the entry API to avoid redundant hash lookups; and wrapping the synchronous HistoryStore::load call in tokio::task::spawn_blocking to prevent blocking the async executor thread.
Important
The consumer version of Gemini Code Assist on GitHub is being sunset. Starting June 18, 2026, new organization installations will be blocked, and all code review activity will officially cease on July 17, 2026.
For more details on the timeline and next steps, please review the Help Documentation.
| } | ||
| log.finish_job(job_id.as_str()).await; | ||
| } | ||
| _ => {} |
There was a problem hiding this comment.
The QueryStageSchedulerEvent::JobCancel event is not handled in the event_log match statement. This results in a file descriptor leak because finish_job is never called for cancelled jobs, and also prevents cancelled jobs from being recorded as completed/ended in the event log (so the history server will never load them). We should handle JobCancel by appending a terminal JobEnd event with a Failed status and then finishing the job.
QueryStageSchedulerEvent::JobCancel(job_id) => {
if let Ok(Some(graph)) = self
.state
.task_manager
.get_job_execution_graph(job_id)
.await
{
log.append_final(
job_id.as_str(),
event_log::job_end_event(
&graph,
ballista_history::event::JobEndStatus::Failed(
"Job cancelled".to_string(),
),
graph.start_time(),
timestamp_millis(),
),
)
.await;
}
log.finish_job(job_id.as_str()).await;
}
_ => {}| let percent_complete = | ||
| ((completed_stages as f32 / num_stages as f32) * 100_f32) as u8; |
There was a problem hiding this comment.
If num_stages is 0 (e.g., if a job fails early during planning or initialization), dividing by num_stages will result in a division by zero / NaN value. We should explicitly guard against num_stages == 0 to prevent unexpected behavior, similar to how it is handled in build_job_response_from_overview.
let percent_complete = if num_stages == 0 {
0
} else {
((completed_stages as f32 / num_stages as f32) * 100_f32) as u8
};| if !handles.contains_key(job_id) { | ||
| let path = log_dir.join(format!("{job_id}.eventlog")); | ||
| match tokio::fs::OpenOptions::new() | ||
| .create(true) | ||
| .append(true) | ||
| .open(&path) | ||
| .await | ||
| { | ||
| Ok(f) => { | ||
| handles.insert(job_id.to_string(), f); | ||
| } | ||
| Err(e) => { | ||
| eprintln!("event-log writer: cannot open {}: {e}", path.display()); | ||
| return None; | ||
| } | ||
| } | ||
| } | ||
| handles.get_mut(job_id) |
There was a problem hiding this comment.
Using contains_key followed by insert and get_mut performs multiple redundant hash lookups. We can optimize this by using the entry API to perform a single lookup.
| if !handles.contains_key(job_id) { | |
| let path = log_dir.join(format!("{job_id}.eventlog")); | |
| match tokio::fs::OpenOptions::new() | |
| .create(true) | |
| .append(true) | |
| .open(&path) | |
| .await | |
| { | |
| Ok(f) => { | |
| handles.insert(job_id.to_string(), f); | |
| } | |
| Err(e) => { | |
| eprintln!("event-log writer: cannot open {}: {e}", path.display()); | |
| return None; | |
| } | |
| } | |
| } | |
| handles.get_mut(job_id) | |
| match handles.entry(job_id.to_string()) { | |
| std::collections::hash_map::Entry::Occupied(entry) => Some(entry.into_mut()), | |
| std::collections::hash_map::Entry::Vacant(entry) => { | |
| let path = log_dir.join(format!("{job_id}.eventlog")); | |
| match tokio::fs::OpenOptions::new() | |
| .create(true) | |
| .append(true) | |
| .open(&path) | |
| .await | |
| { | |
| Ok(f) => Some(entry.insert(f)), | |
| Err(e) => { | |
| eprintln!("event-log writer: cannot open {}: {e}", path.display()); | |
| None | |
| } | |
| } | |
| } | |
| } |
|
|
||
| let args = Args::parse(); | ||
|
|
||
| let store = Arc::new(HistoryStore::load(&args.event_log_dir)?); |
There was a problem hiding this comment.
Calling the synchronous blocking I/O function HistoryStore::load directly inside the async function inner can block the Tokio runtime thread. We should run this blocking operation inside tokio::task::spawn_blocking to avoid blocking the executor thread.
| let store = Arc::new(HistoryStore::load(&args.event_log_dir)?); | |
| let event_log_dir = args.event_log_dir.clone(); | |
| let store = Arc::new( | |
| tokio::task::spawn_blocking(move || HistoryStore::load(&event_log_dir)) | |
| .await | |
| .map_err(|e| std::io::Error::new(std::io::ErrorKind::Other, e))??, | |
| ); |
🤖 Augment PR SummarySummary: This PR introduces a Spark History Server–style “history server” for Ballista by persisting per-job event logs and replaying them via an HTTP API. Changes:
Technical Notes: The history server serves stored DTO bundles from 🤖 Was this summary useful? React with 👍 or 👎 |
![augmentcode[bot]](https://avatars.githubusercontent.com/in/1027498?s=60&v=4){kind=small}
|
|
||
| let num_stages = job.stage_count(); | ||
| let completed_stages = job.completed_stages(); | ||
| let percent_complete = |
There was a problem hiding this comment.
At ballista/scheduler/src/api/dto_build.rs:57-58, percent_complete divides by num_stages without guarding num_stages == 0, unlike build_job_response_from_overview. If stage_count() can ever be 0, this yields NaN/inf and can silently produce a misleading percentage.
Severity: medium
🤖 Was this useful? React with 👍 or 👎, or 🚀 if it prevented an incident/outage.
| physical_plan: Option<String>, | ||
| }, | ||
| StageStart { | ||
| stage_id: usize, |
There was a problem hiding this comment.
At ballista/history/src/event.rs:54-64, the on-disk schema uses usize for StageStart/StageEnd.stage_id but u32 for TaskEnd.stage_id. Since usize is architecture-dependent and the id widths are inconsistent, this can make the “frozen” event-log schema less portable/stable across platforms and consumers.
Severity: medium
🤖 Was this useful? React with 👍 or 👎, or 🚀 if it prevented an incident/outage.
| .status | ||
| .as_ref() | ||
| .map(to_api_task_status) | ||
| .unwrap_or(ApiTaskStatus::Running); |
There was a problem hiding this comment.
At ballista/scheduler/src/scheduler_server/event_log.rs:72-78, task_end_events filters out Running but still allows status == None and then records it as ApiTaskStatus::Running in a TaskEnd event. That can produce a terminal timeline record that claims the task is still running, which seems inconsistent for consumers replaying the log.
Severity: medium
🤖 Was this useful? React with 👍 or 👎, or 🚀 if it prevented an incident/outage.
| ) | ||
| .await; | ||
| } | ||
| log.finish_job(job_id.as_str()).await; |
There was a problem hiding this comment.
At ballista/scheduler/src/scheduler_server/query_stage_scheduler.rs:145-162, finish_job() is called even if get_job_execution_graph() fails and no JobEnd was appended. That leaves a .eventlog without a terminal JobEnd record (so the history server will ignore it), potentially dropping completed/failed jobs from history due to transient graph retrieval issues.
Severity: medium
🤖 Was this useful? React with 👍 or 👎, or 🚀 if it prevented an incident/outage.
Review: Event log & history server (Spark History Server equivalent)
|
3 participants
1925: To review by AI