diff --git a/Cargo.lock b/Cargo.lock index f43041bd45..6f0145d7a6 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1260,6 +1260,7 @@ dependencies = [ "tonic", "tonic-prost", "tonic-prost-build", + "tower", "tower-http 0.7.0", "tracing", "tracing-appender", diff --git a/ballista/history/src/reader.rs b/ballista/history/src/reader.rs index 36382e648e..097bc63c48 100644 --- a/ballista/history/src/reader.rs +++ b/ballista/history/src/reader.rs @@ -25,6 +25,7 @@ use crate::event::{JobEnd, JobIndex, LogRecord, SCHEMA_VERSION, kind}; use ballista_api_types::dto::JobConfig; +use serde::Deserialize; use serde_json::value::RawValue; use std::io::BufRead; use std::path::Path; @@ -91,11 +92,47 @@ impl From for ReadError { /// means the job is still running or the scheduler died before finishing it. /// A `JobEnd` that is present but unusable is an `Err`, so callers can report it /// rather than silently dropping the job. +pub fn read_completed_job(path: &Path) -> Result, ReadError> { + Ok(read_job_end::(path)?.map(|end| ReplayedJob { + index: end.index, + job: end.job, + stages: end.stages, + config: end.config, + dot: end.dot, + })) +} + +/// The `JobEnd` fields needed to list a job, and nothing else. +/// +/// Deserializing into this rather than [`JobEnd`] steps over the stored +/// `/api/job/{id}` and `/api/job/{id}/stages` payloads, the session config and +/// the DOT graph without ever allocating them. That is what lets the history +/// server index a directory of logs without holding their contents. +#[derive(Deserialize)] +struct JobEndIndex { + index: JobIndex, +} + +/// Read only the frozen summary out of a completed job's event log. +/// +/// Same contract as [`read_completed_job`], including how a malformed terminal +/// record is reported, but it recovers only the fields the job list needs. Use +/// it to index a log directory, then [`read_completed_job`] to serve one job. +/// +/// Because the payloads are never parsed, corruption confined to them is not +/// detected here. It surfaces when the job is actually read. +pub fn read_job_index(path: &Path) -> Result, ReadError> { + Ok(read_job_end::(path)?.map(|end| end.index)) +} + +/// Find a log's terminal record and decode it into `T`. /// /// Lines that are not `JobEnd` are skipped without inspection, including ones /// this build does not recognise: a future schema may add record types, and an /// older reader must tolerate them rather than choke on the file. -pub fn read_completed_job(path: &Path) -> Result, ReadError> { +fn read_job_end Deserialize<'de>>( + path: &Path, +) -> Result, ReadError> { let file = std::fs::File::open(path)?; let reader = std::io::BufReader::new(file); @@ -121,14 +158,8 @@ pub fn read_completed_job(path: &Path) -> Result, ReadError> }); } - return match record.decode::() { - Ok(end) => Ok(Some(ReplayedJob { - index: end.index, - job: end.job, - stages: end.stages, - config: end.config, - dot: end.dot, - })), + return match record.decode::() { + Ok(end) => Ok(Some(end)), Err(e) => Err(ReadError::Malformed(e.to_string())), }; } @@ -207,6 +238,52 @@ mod tests { assert!(replayed.job.get().contains("ProjectionExec")); } + /// The index-only read is what the history server builds its job list + /// from, so it has to agree with the full read on every field it carries. + /// If the two ever diverge, the list view and the detail view disagree + /// about the same job. + #[test] + fn index_only_read_agrees_with_full_read() { + let dir = tempfile::tempdir().unwrap(); + let path = write_log(&dir, "job-1.eventlog", &[&job_end_line()]); + + let index = read_job_index(&path).unwrap().expect("completed"); + let full = read_completed_job(&path).unwrap().expect("completed"); + + assert_eq!( + serde_json::to_value(&index).unwrap(), + serde_json::to_value(&full.index).unwrap() + ); + } + + #[test] + fn index_only_read_returns_none_when_no_job_end() { + let dir = tempfile::tempdir().unwrap(); + let path = write_log( + &dir, + "job-6.eventlog", + &[r#"{"ev":"StageStart","version":1,"data":{"stage_id":1,"partitions":4}}"#], + ); + assert!(read_job_index(&path).unwrap().is_none()); + } + + /// A terminal record too broken to yield a summary must still be an error + /// rather than a silently missing job, exactly as for the full read. + #[test] + fn index_only_read_reports_a_malformed_job_end() { + let dir = tempfile::tempdir().unwrap(); + let path = write_log( + &dir, + "job-7.eventlog", + &[r#"{"ev":"JobEnd","version":1,"data":{"status":"Succeeded"}}"#], + ); + + match read_job_index(&path) { + Err(ReadError::Malformed(_)) => {} + other => panic!("expected Malformed, got {other:?}"), + } + } + #[test] fn returns_none_when_no_job_end() { let dir = tempfile::tempdir().unwrap(); @@ -316,6 +393,19 @@ mod compatibility { assert!(replayed.dot.contains("digraph")); } + /// The history server indexes a directory with the index-only read, so it + /// has to work on a log written by an earlier Ballista too. + #[test] + fn indexes_a_v1_log_written_by_an_earlier_ballista() { + let index = read_job_index(&golden_v1()) + .expect("a v1 log must remain indexable") + .expect("the fixture contains a JobEnd record"); + + assert_eq!(index.job_id, "golden-v1"); + assert_eq!(index.job_name, "tpch-q1"); + assert_eq!(index.status, "Completed"); + } + /// The stored responses must come back byte-for-byte, because that is what /// lets the history server re-serve them without understanding them. /// diff --git a/ballista/scheduler/Cargo.toml b/ballista/scheduler/Cargo.toml index ac89d4e7b0..b09f687f75 100644 --- a/ballista/scheduler/Cargo.toml +++ b/ballista/scheduler/Cargo.toml @@ -32,6 +32,11 @@ name = "ballista-scheduler" path = "src/bin/main.rs" required-features = ["build-binary"] +[[bin]] +name = "ballista-history-server" +path = "src/bin/history_server.rs" +required-features = ["build-binary", "rest-api"] + [features] build-binary = ["clap", "tracing-subscriber", "tracing-appender", "tracing", "ballista-core/build-binary"] default = ["build-binary", "rest-api"] @@ -93,6 +98,7 @@ regex = "1" rstest = { workspace = true } serde_json = "1" tempfile = { workspace = true } +tower = "0.5" [build-dependencies] tonic-prost-build = { workspace = true, optional = true } diff --git a/ballista/scheduler/src/bin/history_server.rs b/ballista/scheduler/src/bin/history_server.rs new file mode 100644 index 0000000000..a67eedf8e8 --- /dev/null +++ b/ballista/scheduler/src/bin/history_server.rs @@ -0,0 +1,116 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Standalone Ballista history server binary: loads completed event logs from +//! a directory and serves the same `/api/*` responses the live scheduler does, +//! so the existing TUI can connect to it unchanged. + +use ballista_core::error::{BallistaError, Result}; +use ballista_scheduler::history::{HistoryStore, history_router, spawn_refresh_task}; +use clap::Parser; +use std::env; +use std::net::SocketAddr; +use std::path::PathBuf; +use std::sync::Arc; +use std::time::Duration; +use tracing_subscriber::EnvFilter; + +#[derive(Debug, clap::Parser)] +#[command( + name = "ballista-history-server", + version, + about = "Ballista history server" +)] +struct Args { + /// Directory containing per-job event logs. + #[arg(long)] + event_log_dir: PathBuf, + /// Host to bind the HTTP server to. + #[arg(long, default_value = "0.0.0.0")] + bind_host: String, + /// Port to bind the HTTP server to. + #[arg(long, default_value_t = 50060)] + bind_port: u16, + /// How often to rescan the event-log directory for jobs that finished + /// since the last pass, in seconds. Set to 0 to scan only at startup. + #[arg(long, default_value_t = 10)] + update_interval_seconds: u64, +} + +fn main() -> Result<()> { + let rust_log = env::var(EnvFilter::DEFAULT_ENV); + let log_filter = EnvFilter::new(rust_log.unwrap_or_else(|_| "info".to_string())); + tracing_subscriber::fmt() + .with_ansi(false) + .with_writer(std::io::stdout) + .with_env_filter(log_filter) + .init(); + + let args = Args::parse(); + + // `HistoryStore::load` walks the log directory with blocking file I/O, and + // how long it takes scales with the number of stored jobs. Run it here, + // before the runtime exists, rather than parking a runtime worker on it. + let store = Arc::new(HistoryStore::load(&args.event_log_dir)?); + tracing::info!( + "Indexed {} completed job(s) from {}", + store.len(), + args.event_log_dir.display() + ); + + let runtime = tokio::runtime::Builder::new_multi_thread() + .enable_io() + .enable_time() + .build() + .map_err(BallistaError::IoError)?; + + runtime.block_on(serve(args, store)) +} + +async fn serve(args: Args, store: Arc) -> Result<()> { + // Schedulers keep writing to this directory while the history server is + // up, so without a rescan the list is frozen at whatever had finished when + // the process started. + if args.update_interval_seconds > 0 { + let interval = Duration::from_secs(args.update_interval_seconds); + tracing::info!("Rescanning the event-log directory every {interval:?}"); + spawn_refresh_task(Arc::clone(&store), interval); + } else { + tracing::info!( + "Rescanning is disabled; only jobs indexed at startup will be served" + ); + } + + let app = history_router(store); + + let addr: SocketAddr = format!("{}:{}", args.bind_host, args.bind_port) + .parse() + .map_err(|e: std::net::AddrParseError| { + BallistaError::Configuration(e.to_string()) + })?; + + let listener = tokio::net::TcpListener::bind(&addr) + .await + .map_err(BallistaError::IoError)?; + tracing::info!("History server listening on http://{addr}"); + + axum::serve(listener, app.into_make_service()) + .await + .map_err(BallistaError::IoError)?; + + Ok(()) +} diff --git a/ballista/scheduler/src/history/mod.rs b/ballista/scheduler/src/history/mod.rs new file mode 100644 index 0000000000..30b6b53fab --- /dev/null +++ b/ballista/scheduler/src/history/mod.rs @@ -0,0 +1,1066 @@ +// Licensed to the Apache Software Foundation (ASF) under one +// or more contributor license agreements. See the NOTICE file +// distributed with this work for additional information +// regarding copyright ownership. The ASF licenses this file +// to you under the Apache License, Version 2.0 (the +// "License"); you may not use this file except in compliance +// with the License. You may obtain a copy of the License at +// +// http://www.apache.org/licenses/LICENSE-2.0 +// +// Unless required by applicable law or agreed to in writing, +// software distributed under the License is distributed on an +// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY +// KIND, either express or implied. See the License for the +// specific language governing permissions and limitations +// under the License. + +//! Standalone history server: indexes completed event logs and serves the same +//! `/api/*` responses the live scheduler does, from stored DTOs. + +use crate::api::SchedulerErrorResponse; +use axum::response::IntoResponse; +use axum::{ + Json, Router, + extract::{Path as AxumPath, State}, + routing::get, +}; +use ballista_api_types::dto::{JobConfig, JobResponse}; +use ballista_core::BALLISTA_VERSION; +use ballista_history::event::JobIndex; +use ballista_history::reader::{ + ReadError, ReplayedJob, read_completed_job, read_job_index, +}; +use datafusion::DATAFUSION_VERSION; +use http::StatusCode; +use http::header::CONTENT_TYPE; +use serde_json::value::RawValue; +use std::collections::{HashMap, HashSet}; +use std::path::{Path, PathBuf}; +use std::sync::{Arc, PoisonError, RwLock}; +use std::time::{Duration, SystemTime}; + +/// Where one completed job lives, and just enough about it to list it. +struct JobEntry { + /// Frozen summary, everything `GET /api/jobs` reports. + index: JobIndex, + /// The `.eventlog` the rest of the job is read back from. + path: PathBuf, +} + +/// What a log file looked like when it was last indexed. +/// +/// Cheap enough to take for every file on every rescan, which is the point: a +/// directory of thousands of finished jobs costs a `stat` per file per pass, +/// and only the files that actually changed are opened and parsed. +/// +/// A rewrite that lands within the same modification-time tick *and* leaves the +/// length unchanged is indistinguishable from no change here, so it is not +/// picked up until the file changes again. Schedulers only ever append to a +/// log and then close it, so that does not happen in practice. +#[derive(Clone, Copy, PartialEq, Eq)] +struct FileStamp { + /// `None` on the platforms/filesystems that do not report it. + modified: Option, + len: u64, +} + +/// Everything a rescan reads and writes, behind one lock. +#[derive(Default)] +struct Index { + /// Completed jobs keyed by job id. + jobs: HashMap, + /// Every `.eventlog` seen so far and the job id it produced, if any. Logs + /// that are still running (no terminal record yet) or unreadable are kept + /// here with `None` so a rescan can tell "already looked at, unchanged" + /// from "never seen". + seen: HashMap)>, +} + +/// What one directory rescan changed. +#[derive(Debug, Default, PartialEq, Eq)] +pub struct RefreshStats { + /// Jobs that appeared: a log written since the last pass, or one that has + /// gained its terminal record since it was last looked at. + pub added: usize, + /// Jobs dropped because their log is no longer in the directory. + pub removed: usize, +} + +impl RefreshStats { + /// Whether the pass found nothing to do, which is the common case and the + /// one not worth logging. + fn is_noop(&self) -> bool { + self.added == 0 && self.removed == 0 + } +} + +/// Index of the completed jobs found in an event-log directory. +/// +/// Only each job's [`JobIndex`] is held in memory. The stored payloads (both +/// plan-bearing REST responses, the session config and the DOT graph) run to +/// megabytes for a job with many tasks, and would otherwise sit resident for +/// every job in the directory whether or not anyone ever looks at it. They are +/// read back from disk per request instead, which is fine at the rate a person +/// clicks through a UI. +/// +/// The directory is normally one that one or more live schedulers are still +/// writing to, so the index is not fixed at startup: [`HistoryStore::refresh`] +/// rescans it, and [`spawn_refresh_task`] keeps that happening on a timer. +pub struct HistoryStore { + /// The event-log directory this store indexes. + dir: PathBuf, + index: RwLock, +} + +/// Why reading one job's stored payload back produced nothing. +#[derive(Debug)] +pub enum JobReadError { + /// No job with this id is in the index. + NotFound, + /// The log was indexed but could not be read now. + Unreadable(ReadError), + /// The log no longer has a terminal record, so it was replaced or + /// truncated after the index was built. + Vanished, +} + +impl std::fmt::Display for JobReadError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + JobReadError::NotFound => write!(f, "no such job"), + JobReadError::Unreadable(e) => write!(f, "event log is unreadable: {e}"), + JobReadError::Vanished => { + write!(f, "event log no longer contains a terminal record") + } + } + } +} + +impl HistoryStore { + /// Index every completed job found under `dir`. Missing directories yield + /// an empty store rather than an error. + /// + /// A single unreadable/corrupt `.eventlog` file (e.g. truncated by a + /// crash mid-write) is logged and skipped rather than failing the whole + /// load — one bad log must not hide every other completed job. Only a + /// failure to read the directory itself is propagated. + /// + /// Each log is read once here, but only its summary is decoded, so this + /// costs a pass over the directory rather than a copy of it in memory. + /// Corruption confined to the payloads therefore surfaces when the job is + /// requested rather than at load. + pub fn load(dir: &Path) -> std::io::Result { + let store = HistoryStore { + dir: dir.to_path_buf(), + index: RwLock::new(Index::default()), + }; + store.refresh()?; + Ok(store) + } + + /// Rescan the event-log directory and fold what changed into the index. + /// + /// This is what makes a job that finished after the server started + /// visible. Only logs whose size or modification time has moved are + /// opened, so a pass over an unchanged directory reads no files; logs that + /// have gone are dropped from the index, so deleting one no longer leaves + /// an entry that fails when opened. + /// + /// Blocking file I/O — call it from `spawn_blocking`, not on a runtime + /// worker. + pub fn refresh(&self) -> std::io::Result { + let found = self.scan_dir()?; + + // Work out what needs reading under a read lock, then do the reading + // with the lock released: parsing a log is orders of magnitude slower + // than the bookkeeping, and requests are served from the index + // throughout. + let stale: Vec = { + let index = self.read_index(); + found + .iter() + .filter(|(path, stamp)| { + index + .seen + .get(path.as_path()) + .is_none_or(|(s, _)| s != stamp) + }) + .map(|(path, _)| path.clone()) + .collect() + }; + + let read: Vec<(PathBuf, Option)> = stale + .into_iter() + .map(|path| { + let index = match read_job_index(&path) { + Ok(index) => index, + Err(err) => { + tracing::warn!( + "skipping unreadable event log {}: {err}", + path.display() + ); + None + } + }; + (path, index) + }) + .collect(); + + let present: HashSet<&Path> = found.iter().map(|(p, _)| p.as_path()).collect(); + let mut stats = RefreshStats::default(); + let mut index = self.write_index(); + let Index { jobs, seen } = &mut *index; + + // Logs that have gone take their jobs with them. + seen.retain(|path, (_, job_id)| { + if present.contains(path.as_path()) { + return true; + } + if let Some(job_id) = job_id + && jobs.remove(job_id).is_some() + { + stats.removed += 1; + } + false + }); + + let stamps: HashMap<&Path, FileStamp> = + found.iter().map(|(p, s)| (p.as_path(), *s)).collect(); + for (path, job_index) in read { + let Some(stamp) = stamps.get(path.as_path()).copied() else { + // Deleted between the scan and now; the next pass will see it. + continue; + }; + // A rewritten log can name a different job than it used to. + let previous = seen.insert( + path.clone(), + (stamp, job_index.as_ref().map(|i| i.job_id.clone())), + ); + if let Some((_, Some(previous_id))) = previous + && Some(previous_id.as_str()) + != job_index.as_ref().map(|i| i.job_id.as_str()) + && jobs.remove(&previous_id).is_some() + { + stats.removed += 1; + } + if let Some(job_index) = job_index + && jobs + .insert( + job_index.job_id.clone(), + JobEntry { + index: job_index, + path, + }, + ) + .is_none() + { + stats.added += 1; + } + } + + Ok(stats) + } + + /// Stat every `.eventlog` in the directory. A directory that does not + /// exist yet is an empty one: the history server is routinely started + /// before the scheduler has written anything. + fn scan_dir(&self) -> std::io::Result> { + let mut found = Vec::new(); + if !self.dir.exists() { + return Ok(found); + } + for entry in std::fs::read_dir(&self.dir)? { + let entry = entry?; + let path = entry.path(); + if path.extension().and_then(|e| e.to_str()) != Some("eventlog") { + continue; + } + // A file that disappears mid-scan is simply not there this pass. + let Ok(metadata) = entry.metadata() else { + continue; + }; + found.push(( + path, + FileStamp { + modified: metadata.modified().ok(), + len: metadata.len(), + }, + )); + } + Ok(found) + } + + /// How many completed jobs are indexed. + pub fn len(&self) -> usize { + self.read_index().jobs.len() + } + + /// Whether the log directory holds no completed jobs. + pub fn is_empty(&self) -> bool { + self.len() == 0 + } + + /// Read one job's stored payload back from its event log. + /// + /// This is blocking file I/O, so the request handlers call it from + /// `spawn_blocking` rather than on a runtime worker. + pub fn read_job(&self, job_id: &str) -> Result { + // Copy the path out rather than reading with the lock held, so a slow + // read of one job does not block a rescan or the job list. + let path = { + let index = self.read_index(); + index + .jobs + .get(job_id) + .map(|entry| entry.path.clone()) + .ok_or(JobReadError::NotFound)? + }; + match read_completed_job(&path) { + Ok(Some(replayed)) => Ok(replayed), + Ok(None) => Err(JobReadError::Vanished), + Err(e) => Err(JobReadError::Unreadable(e)), + } + } + + /// Every indexed job's summary, in no particular order. + fn summaries(&self) -> Vec { + self.read_index() + .jobs + .values() + .map(|entry| entry.index.clone()) + .collect() + } + + /// A poisoned index means a rescan panicked partway through. What is in + /// there is still a valid, if possibly stale, view of the directory, and + /// serving it beats taking the whole server down, so poisoning is ignored + /// on both paths. + fn read_index(&self) -> std::sync::RwLockReadGuard<'_, Index> { + self.index.read().unwrap_or_else(PoisonError::into_inner) + } + + fn write_index(&self) -> std::sync::RwLockWriteGuard<'_, Index> { + self.index.write().unwrap_or_else(PoisonError::into_inner) + } +} + +/// Rescan `store`'s directory every `interval` for as long as the returned +/// task runs. +/// +/// The history server is normally pointed at a directory that one or more live +/// schedulers are still writing to, so without this it would only ever serve +/// the jobs that had finished before it started. Polling rather than watching +/// the filesystem is deliberate: these directories are usually on a shared or +/// network filesystem, where change notifications are unreliable or absent. +pub fn spawn_refresh_task( + store: Arc, + interval: Duration, +) -> tokio::task::JoinHandle<()> { + tokio::spawn(async move { + let mut ticker = tokio::time::interval(interval); + // Skip the immediate first tick: the caller has just loaded the store. + ticker.set_missed_tick_behavior(tokio::time::MissedTickBehavior::Delay); + ticker.tick().await; + loop { + ticker.tick().await; + let store = Arc::clone(&store); + match tokio::task::spawn_blocking(move || store.refresh()).await { + Ok(Ok(stats)) if stats.is_noop() => {} + Ok(Ok(stats)) => tracing::info!( + "history server: {} job(s) added, {} removed", + stats.added, + stats.removed + ), + Ok(Err(err)) => { + tracing::warn!( + "history server: rescanning the log directory failed: {err}" + ) + } + Err(err) => { + tracing::warn!( + "history server: rescanning the log directory panicked: {err}" + ) + } + } + } + }) +} + +/// [`HistoryStore::read_job`] moved off the async runtime. +/// +/// A log that was indexed and cannot be read now means the file changed +/// underneath us, so both failures are logged rather than only being reported +/// to whoever happened to ask. +async fn read_job_blocking( + store: Arc, + job_id: String, +) -> Result { + let result = tokio::task::spawn_blocking(move || { + store.read_job(&job_id).map_err(|e| (job_id, e)) + }) + .await + .map_err(|e| { + tracing::warn!("history server: reading an event log panicked: {e}"); + SchedulerErrorResponse::new(StatusCode::INTERNAL_SERVER_ERROR) + })?; + + result.map_err(|(job_id, err)| match err { + JobReadError::NotFound => SchedulerErrorResponse::new(StatusCode::NOT_FOUND), + JobReadError::Vanished => { + tracing::warn!("history server: event log for {job_id} {err}"); + SchedulerErrorResponse::with_error(StatusCode::NOT_FOUND, err.to_string()) + } + JobReadError::Unreadable(_) => { + tracing::warn!("history server: event log for {job_id} {err}"); + SchedulerErrorResponse::with_error( + StatusCode::INTERNAL_SERVER_ERROR, + err.to_string(), + ) + } + }) +} + +/// Build the axum router serving `/api/*` from a loaded [`HistoryStore`]. +pub fn history_router(store: Arc) -> Router { + Router::new() + .route("/api/jobs", get(get_jobs)) + .route("/api/job/{job_id}", get(get_job)) + .route("/api/job/{job_id}/stages", get(get_stages)) + .route("/api/job/{job_id}/config", get(get_config)) + .route("/api/job/{job_id}/dot", get(get_dot)) + .route("/api/executors", get(get_executors_empty)) + .route("/api/state", get(get_state)) + .with_state(store) +} + +/// Rebuild a job-list entry from the stored index. +/// +/// Built from [`JobIndex`] rather than by editing the stored `/api/job/{id}` +/// payload: the list endpoint omits the plan fields, and the index carries +/// exactly the fields it does include. That keeps this path from having to +/// parse a payload it would only throw most of away. +fn list_entry(index: &JobIndex) -> JobResponse { + JobResponse { + job_id: index.job_id.clone(), + job_name: index.job_name.clone(), + job_status: index.job_status.clone(), + status: index.status.clone(), + num_stages: index.num_stages, + completed_stages: index.completed_stages, + percent_complete: index.percent_complete, + start_time: index.start_time, + end_time: index.end_time, + logical_plan: None, + physical_plan: None, + stage_plan: None, + } +} + +/// The one endpoint that touches every job, and the reason the index is held +/// in memory at all: it is served without going near the disk. +/// +/// Newest first. A job id is a random 7-character string +/// (`TaskManager::generate_job_id`), so ordering by it would be arbitrary, +/// whereas start time is both meaningful and the order the TUI puts the list +/// into once it has it. +async fn get_jobs(State(store): State>) -> Json> { + let mut jobs: Vec = store.summaries().iter().map(list_entry).collect(); + // Job id breaks ties, so two jobs that started in the same millisecond + // cannot swap places between requests. + jobs.sort_by(|a, b| { + b.start_time + .cmp(&a.start_time) + .then_with(|| a.job_id.cmp(&b.job_id)) + }); + Json(jobs) +} + +/// Serve a stored payload exactly as the scheduler wrote it. +/// +/// The payload is relayed as raw JSON rather than deserialized and +/// re-serialized, so the bytes a client receives are the bytes the live +/// scheduler produced, and a change to the REST types cannot make an existing +/// log unservable. +fn raw_json(payload: &RawValue) -> axum::response::Response { + ( + [(CONTENT_TYPE, "application/json")], + payload.get().to_string(), + ) + .into_response() +} + +async fn get_job( + State(store): State>, + AxumPath(job_id): AxumPath, +) -> Result { + let job = read_job_blocking(store, job_id).await?; + Ok(raw_json(&job.job)) +} + +async fn get_stages( + State(store): State>, + AxumPath(job_id): AxumPath, +) -> Result { + let job = read_job_blocking(store, job_id).await?; + Ok(raw_json(&job.stages)) +} + +async fn get_config( + State(store): State>, + AxumPath(job_id): AxumPath, +) -> Result, SchedulerErrorResponse> { + let job = read_job_blocking(store, job_id).await?; + Ok(Json(job.config)) +} + +async fn get_dot( + State(store): State>, + AxumPath(job_id): AxumPath, +) -> Result { + let job = read_job_blocking(store, job_id).await?; + Ok(job.dot) +} + +async fn get_executors_empty() -> Json> { + Json(vec![]) +} + +/// Static `/api/state` payload. The history server has no live scheduler +/// process behind it, so every field that would normally reflect runtime +/// state (uptime, feature flags, scheduling policy) is a fixed placeholder. +/// Field names/types match the live `/api/state` response +/// (`SchedulerStateResponse` in `api/handlers.rs`) and what the TUI +/// deserializes into (`ballista-cli/src/tui/domain/mod.rs::SchedulerState`), +/// so the TUI's startup call succeeds instead of erroring out. +async fn get_state() -> Json { + Json(serde_json::json!({ + "started": 0, + "version": BALLISTA_VERSION, + "datafusion_version": DATAFUSION_VERSION, + "substrait_support": false, + "keda_support": false, + "prometheus_support": false, + "graphviz_support": false, + "spark_support": false, + "scheduling_policy": "history-server", + })) +} + +#[cfg(test)] +mod tests { + use super::*; + use axum::body::Body; + use axum::http::{Request, StatusCode}; + use ballista_api_types::dto::{QueryStageSummary, QueryStagesResponse}; + use ballista_history::event::{HistoryEvent, JobEnd, JobEndStatus, JobIndex}; + use std::io::Write; + use tempfile::tempdir; + use tower::ServiceExt; // oneshot + + const STAGE_ID_MARKER: &str = "stage-42"; + + fn sample_replayed_job_with_stage( + job_id: &str, + stage_id: &str, + start_time: u64, + ) -> ReplayedJob { + let job = JobResponse { + job_id: job_id.into(), + job_name: "q1".into(), + job_status: "COMPLETED".into(), + status: "Successful".into(), + num_stages: 1, + completed_stages: 1, + percent_complete: 100, + start_time, + end_time: start_time + 1, + logical_plan: Some("Projection".into()), + physical_plan: Some("ProjectionExec".into()), + stage_plan: Some("stage".into()), + }; + let stages = QueryStagesResponse { + stages: vec![QueryStageSummary { + stage_id: stage_id.into(), + stage_status: "Completed".into(), + input_rows: 10, + output_rows: 5, + elapsed_compute: Some("1ms".into()), + stage_plan: None, + task_duration_percentiles: None, + task_input_percentiles: None, + tasks: vec![], + }], + }; + ReplayedJob { + index: JobIndex { + job_id: job_id.into(), + job_name: "q1".into(), + status: "Successful".into(), + job_status: "COMPLETED".into(), + start_time, + end_time: start_time + 1, + num_stages: 1, + completed_stages: 1, + percent_complete: 100, + }, + job: serde_json::value::to_raw_value(&job).unwrap(), + stages: serde_json::value::to_raw_value(&stages).unwrap(), + config: Default::default(), + dot: "digraph {}".into(), + } + } + + /// Every test goes through a real directory of logs, because the store no + /// longer holds payloads it could be handed directly. + fn store_with_one_job(dir: &tempfile::TempDir) -> Arc { + write_job_end_log(&dir.path().join("job-1.eventlog"), "job-1"); + Arc::new(HistoryStore::load(dir.path()).unwrap()) + } + + #[tokio::test] + async fn jobs_endpoint_nulls_plan_fields() { + let dir = tempdir().unwrap(); + let app = history_router(store_with_one_job(&dir)); + let resp = app + .oneshot( + Request::builder() + .uri("/api/jobs") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX) + .await + .unwrap(); + let body = String::from_utf8(bytes.to_vec()).unwrap(); + assert!(body.contains("\"job_id\":\"job-1\"")); + assert!(!body.contains("physical_plan")); // nulled + skip_serializing_if + } + + #[tokio::test] + async fn stages_endpoint_returns_stored_dto() { + let dir = tempdir().unwrap(); + let app = history_router(store_with_one_job(&dir)); + let resp = app + .oneshot( + Request::builder() + .uri("/api/job/job-1/stages") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX) + .await + .unwrap(); + let body: QueryStagesResponse = serde_json::from_slice(&bytes).unwrap(); + assert_eq!(body.stages.len(), 1); + assert_eq!(body.stages[0].stage_id, STAGE_ID_MARKER); + assert_eq!(body.stages[0].input_rows, 10); + assert_eq!(body.stages[0].output_rows, 5); + } + + #[tokio::test] + async fn missing_job_returns_404_on_job_and_stages() { + let dir = tempdir().unwrap(); + let app = history_router(store_with_one_job(&dir)); + + let resp = app + .clone() + .oneshot( + Request::builder() + .uri("/api/job/does-not-exist") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::NOT_FOUND); + + let resp = app + .oneshot( + Request::builder() + .uri("/api/job/does-not-exist/stages") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::NOT_FOUND); + } + + #[tokio::test] + async fn state_endpoint_returns_static_payload() { + let dir = tempdir().unwrap(); + let app = history_router(store_with_one_job(&dir)); + let resp = app + .oneshot( + Request::builder() + .uri("/api/state") + .body(Body::empty()) + .unwrap(), + ) + .await + .unwrap(); + assert_eq!(resp.status(), StatusCode::OK); + let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX) + .await + .unwrap(); + let value: serde_json::Value = serde_json::from_slice(&bytes).unwrap(); + for field in [ + "started", + "version", + "datafusion_version", + "substrait_support", + "keda_support", + "prometheus_support", + "graphviz_support", + "spark_support", + "scheduling_policy", + ] { + assert!(value.get(field).is_some(), "missing field: {field}"); + } + } + + /// Job ids are random 7-character strings, so ordering the list by id puts + /// it in an order that means nothing. Newest first is what a history view + /// wants, and it is what makes a future `?limit=` worth having. + #[tokio::test] + async fn jobs_are_listed_newest_first() { + let dir = tempdir().unwrap(); + for (job_id, start_time) in [("zzz", 10u64), ("aaa", 30), ("mmm", 20)] { + write_job_end_log_with_stage( + &dir.path().join(format!("{job_id}.eventlog")), + job_id, + STAGE_ID_MARKER, + start_time, + ); + } + let app = history_router(Arc::new(HistoryStore::load(dir.path()).unwrap())); + + let (status, body) = get(&app, "/api/jobs").await; + assert_eq!(status, StatusCode::OK); + let jobs: Vec = serde_json::from_str(&body).unwrap(); + let order: Vec<&str> = jobs.iter().map(|j| j.job_id.as_str()).collect(); + assert_eq!(order, ["aaa", "mmm", "zzz"]); + } + + /// Equal start times must still produce a total order, otherwise the list + /// can reshuffle between two identical requests. + #[tokio::test] + async fn jobs_with_the_same_start_time_are_ordered_by_id() { + let dir = tempdir().unwrap(); + for job_id in ["ccc", "aaa", "bbb"] { + write_job_end_log_with_stage( + &dir.path().join(format!("{job_id}.eventlog")), + job_id, + STAGE_ID_MARKER, + 7, + ); + } + let app = history_router(Arc::new(HistoryStore::load(dir.path()).unwrap())); + + let (_, body) = get(&app, "/api/jobs").await; + let jobs: Vec = serde_json::from_str(&body).unwrap(); + let order: Vec<&str> = jobs.iter().map(|j| j.job_id.as_str()).collect(); + assert_eq!(order, ["aaa", "bbb", "ccc"]); + } + + /// Fetch one path and return the status and body together. + async fn get(app: &Router, uri: &str) -> (StatusCode, String) { + let resp = app + .clone() + .oneshot(Request::builder().uri(uri).body(Body::empty()).unwrap()) + .await + .unwrap(); + let status = resp.status(); + let bytes = axum::body::to_bytes(resp.into_body(), usize::MAX) + .await + .unwrap(); + (status, String::from_utf8(bytes.to_vec()).unwrap()) + } + + /// The point of the whole design: nothing but the summary is retained, so + /// a detail request has to go back to the file. Rewriting the log behind a + /// loaded store and seeing the new contents served is the only way to + /// observe that from outside. + #[tokio::test] + async fn detail_endpoints_read_the_log_on_demand() { + let dir = tempdir().unwrap(); + let path = dir.path().join("job-1.eventlog"); + let app = history_router(store_with_one_job(&dir)); + + let (status, body) = get(&app, "/api/job/job-1/stages").await; + assert_eq!(status, StatusCode::OK); + assert!(body.contains(STAGE_ID_MARKER)); + + write_job_end_log_with_stage(&path, "job-1", "rewritten-stage", 2); + + let (status, body) = get(&app, "/api/job/job-1/stages").await; + assert_eq!(status, StatusCode::OK); + assert!( + body.contains("rewritten-stage"), + "payload should be read per request, not cached at load: {body}" + ); + } + + /// Corruption confined to the payloads is not visible when the directory + /// is indexed, so it has to be reported at request time. A 500 naming the + /// problem beats an empty or truncated response. + #[tokio::test] + async fn a_log_that_breaks_after_indexing_reports_the_failure() { + let dir = tempdir().unwrap(); + let path = dir.path().join("job-1.eventlog"); + let app = history_router(store_with_one_job(&dir)); + + std::fs::write(&path, [0xff, 0xfe, 0xfd]).unwrap(); + + let (status, body) = get(&app, "/api/job/job-1").await; + assert_eq!(status, StatusCode::INTERNAL_SERVER_ERROR); + assert!(body.contains("unreadable"), "got: {body}"); + + // The job list is served from the index, so it still lists the job. + let (status, body) = get(&app, "/api/jobs").await; + assert_eq!(status, StatusCode::OK); + assert!(body.contains("job-1")); + } + + /// A log whose terminal record has gone (rotated, truncated) is a job that + /// no longer exists rather than a server fault. + #[tokio::test] + async fn a_log_that_loses_its_terminal_record_returns_404() { + let dir = tempdir().unwrap(); + let path = dir.path().join("job-1.eventlog"); + let app = history_router(store_with_one_job(&dir)); + + std::fs::write( + &path, + "{\"ev\":\"StageStart\",\"version\":1,\"data\":{\"stage_id\":1}}\n", + ) + .unwrap(); + + let (status, _) = get(&app, "/api/job/job-1/config").await; + assert_eq!(status, StatusCode::NOT_FOUND); + } + + fn write_job_end_log(path: &Path, job_id: &str) { + write_job_end_log_with_stage(path, job_id, STAGE_ID_MARKER, 2) + } + + fn write_job_end_log_with_stage( + path: &Path, + job_id: &str, + stage_id: &str, + start_time: u64, + ) { + let replayed = sample_replayed_job_with_stage(job_id, stage_id, start_time); + let event = HistoryEvent::JobEnd(Box::new(JobEnd { + status: JobEndStatus::Succeeded, + queued_at: 0, + started_at: 2, + completed_at: 3, + index: replayed.index, + job: replayed.job, + stages: replayed.stages, + config: replayed.config, + dot: replayed.dot, + })); + let line = serde_json::to_string(&event.to_record().unwrap()).unwrap(); + std::fs::write(path, format!("{line}\n")).unwrap(); + } + + /// The reason the store rescans at all: a history server is normally + /// pointed at a directory a live scheduler is still writing to, so a job + /// that finishes after startup has to show up without a restart. + #[test] + fn refresh_picks_up_a_job_written_after_load() { + let dir = tempdir().unwrap(); + write_job_end_log(&dir.path().join("job-1.eventlog"), "job-1"); + let store = HistoryStore::load(dir.path()).unwrap(); + assert_eq!(store.len(), 1); + + write_job_end_log(&dir.path().join("job-2.eventlog"), "job-2"); + + let stats = store.refresh().unwrap(); + assert_eq!( + stats, + RefreshStats { + added: 1, + removed: 0 + } + ); + assert_eq!(store.len(), 2); + assert!(store.read_job("job-2").is_ok()); + } + + /// A rescan that finds nothing new must not re-read the logs it already + /// indexed, or a directory of thousands of finished jobs would be parsed + /// end to end on every tick. + #[test] + fn refresh_is_a_noop_when_nothing_changed() { + let dir = tempdir().unwrap(); + write_job_end_log(&dir.path().join("job-1.eventlog"), "job-1"); + let store = HistoryStore::load(dir.path()).unwrap(); + + // Truncating the log to garbage after it is indexed would break a + // rescan that re-parsed it; the stamp is unchanged only if the file is + // untouched, so a clean pass here means nothing was opened. + assert_eq!(store.refresh().unwrap(), RefreshStats::default()); + assert_eq!(store.len(), 1); + } + + /// A log with no terminal record yet is a job still running. It is not + /// listed, but it must not be written off either: the next rescan after + /// the job ends has to pick it up. + #[test] + fn refresh_indexes_a_log_that_gains_its_terminal_record() { + let dir = tempdir().unwrap(); + let path = dir.path().join("job-1.eventlog"); + std::fs::write( + &path, + "{\"ev\":\"StageStart\",\"version\":1,\"data\":{\"stage_id\":1}}\n", + ) + .unwrap(); + + let store = HistoryStore::load(dir.path()).unwrap(); + assert!(store.is_empty()); + + write_job_end_log(&path, "job-1"); + + assert_eq!(store.refresh().unwrap().added, 1); + assert_eq!(store.len(), 1); + } + + /// Pruning a directory of old logs used to leave the jobs listed until the + /// server was restarted, with every click on one failing. + #[test] + fn refresh_drops_a_job_whose_log_was_deleted() { + let dir = tempdir().unwrap(); + let path = dir.path().join("job-1.eventlog"); + write_job_end_log(&path, "job-1"); + write_job_end_log(&dir.path().join("job-2.eventlog"), "job-2"); + let store = HistoryStore::load(dir.path()).unwrap(); + assert_eq!(store.len(), 2); + + std::fs::remove_file(&path).unwrap(); + + let stats = store.refresh().unwrap(); + assert_eq!( + stats, + RefreshStats { + added: 0, + removed: 1 + } + ); + assert!(matches!( + store.read_job("job-1"), + Err(JobReadError::NotFound) + )); + assert!(store.read_job("job-2").is_ok()); + } + + /// A log rewritten under a different job id must not leave the old id + /// pointing at a file that no longer describes it. + #[test] + fn refresh_replaces_the_job_when_a_log_is_rewritten() { + let dir = tempdir().unwrap(); + let path = dir.path().join("job.eventlog"); + write_job_end_log(&path, "old-id"); + let store = HistoryStore::load(dir.path()).unwrap(); + assert!(store.read_job("old-id").is_ok()); + + write_job_end_log_with_stage(&path, "new-id", "another-stage", 99); + + store.refresh().unwrap(); + assert_eq!(store.len(), 1); + assert!(matches!( + store.read_job("old-id"), + Err(JobReadError::NotFound) + )); + assert!(store.read_job("new-id").is_ok()); + } + + /// Being started before the scheduler has written anything is normal, and + /// the directory appearing later has to be picked up like any other change. + #[test] + fn a_directory_that_does_not_exist_yet_is_empty_then_indexed() { + let dir = tempdir().unwrap(); + let logs = dir.path().join("not-created-yet"); + let store = HistoryStore::load(&logs).unwrap(); + assert!(store.is_empty()); + + std::fs::create_dir(&logs).unwrap(); + write_job_end_log(&logs.join("job-1.eventlog"), "job-1"); + + assert_eq!(store.refresh().unwrap().added, 1); + assert_eq!(store.len(), 1); + } + + /// `/api/jobs` is served from the index, so the background rescan is what + /// makes a newly finished job visible over HTTP. + #[tokio::test] + async fn jobs_endpoint_reflects_a_refresh() { + let dir = tempdir().unwrap(); + let store = Arc::new(HistoryStore::load(dir.path()).unwrap()); + let app = history_router(Arc::clone(&store)); + + let (status, body) = get(&app, "/api/jobs").await; + assert_eq!(status, StatusCode::OK); + assert_eq!(body, "[]"); + + write_job_end_log(&dir.path().join("job-1.eventlog"), "job-1"); + store.refresh().unwrap(); + + let (status, body) = get(&app, "/api/jobs").await; + assert_eq!(status, StatusCode::OK); + assert!(body.contains("\"job_id\":\"job-1\""), "got: {body}"); + } + + /// The timer task is the only thing wiring rescans to wall-clock time, so + /// it is worth proving it actually runs one. + #[tokio::test] + async fn the_refresh_task_indexes_new_jobs() { + let dir = tempdir().unwrap(); + let store = Arc::new(HistoryStore::load(dir.path()).unwrap()); + let task = spawn_refresh_task(Arc::clone(&store), Duration::from_millis(10)); + + write_job_end_log(&dir.path().join("job-1.eventlog"), "job-1"); + + for _ in 0..200 { + if store.len() == 1 { + task.abort(); + return; + } + tokio::time::sleep(Duration::from_millis(10)).await; + } + task.abort(); + panic!("the refresh task never indexed the new job"); + } + + #[test] + fn load_skips_corrupt_eventlog_and_keeps_good_one() { + let dir = tempdir().unwrap(); + + // A good, readable event log. + write_job_end_log(&dir.path().join("job-good.eventlog"), "job-good"); + + // A corrupt file: invalid UTF-8, as if a crash truncated a write + // mid-multibyte-character. + let mut corrupt = + std::fs::File::create(dir.path().join("job-bad.eventlog")).unwrap(); + corrupt.write_all(&[0xff, 0xfe, 0xfd]).unwrap(); + drop(corrupt); + + let store = HistoryStore::load(dir.path()).unwrap(); + assert_eq!(store.len(), 1); + assert!(store.read_job("job-good").is_ok()); + assert!(matches!( + store.read_job("job-bad"), + Err(JobReadError::NotFound) + )); + } +} diff --git a/ballista/scheduler/src/lib.rs b/ballista/scheduler/src/lib.rs index 533711ecb0..8338a65372 100644 --- a/ballista/scheduler/src/lib.rs +++ b/ballista/scheduler/src/lib.rs @@ -25,6 +25,9 @@ pub mod cluster; pub mod config; /// Display utilities for execution plans and state. pub mod display; +/// Standalone history server: serves `/api/*` from stored event logs. +#[cfg(feature = "rest-api")] +pub mod history; /// Metrics collection and reporting. pub mod metrics; /// Physical query plan optimizers. diff --git a/ballista/scheduler/src/scheduler_server/event_log.rs b/ballista/scheduler/src/scheduler_server/event_log.rs index 7e73dc28ba..0b83d6813b 100644 --- a/ballista/scheduler/src/scheduler_server/event_log.rs +++ b/ballista/scheduler/src/scheduler_server/event_log.rs @@ -393,4 +393,50 @@ mod tests { assert!(lines[1].contains("\"ev\":\"JobEnd\"")); assert!(lines[1].contains("\"physical_plan\":\"HashJoinExec")); } + + /// End-to-end parity, and the reason the whole design stores built + /// responses rather than re-deriving them: replaying a real event log + /// through `EventLogWriter` and `HistoryStore::load` must yield exactly the + /// JSON the live scheduler would have served for the same graph. + /// + /// This lives here rather than under `tests/` because `dto_build`, + /// `job_end_event` and `execution_graph_dot::tests::test_graph` are all + /// crate-internal, and an integration test only sees the public API. + #[tokio::test] + async fn history_store_serves_byte_identical_json_to_live_scheduler() { + use crate::history::HistoryStore; + + let graph = test_graph().await.unwrap(); + let graph: ExecutionGraphBox = Box::new(graph); + let job_id = graph.job_id().to_string(); + + // `completed_at` is the snapshot instant for both sides, so the + // comparison does not race the wall clock. + const COMPLETED_AT: u64 = 2; + let live_job = graph_to_job_response(&graph, PlanFormat::Default); + let live_stages = + graph_to_query_stages(&graph, PlanFormat::Default, COMPLETED_AT as u128); + + // Emit the same JobEnd the scheduler writes on completion, through the + // real async writer, then load it back through the history server's own + // HistoryStore: the full write -> read -> serve path. + let event = job_end_event(&graph, JobEndStatus::Succeeded, 1, COMPLETED_AT); + let dir = tempfile::tempdir().unwrap(); + let writer = EventLogWriter::new(dir.path().to_path_buf(), 16); + writer.append(&job_id, event); + writer.flush_job(&job_id).await; + + let store = HistoryStore::load(dir.path()).unwrap(); + let replayed = store.read_job(&job_id).expect("job should be replayed"); + + assert_eq!( + replayed.job.get(), + serde_json::to_string(&live_job).unwrap(), + "the history server must serve the live scheduler's exact bytes" + ); + assert_eq!( + replayed.stages.get(), + serde_json::to_string(&live_stages).unwrap(), + ); + } } diff --git a/docs/source/index.rst b/docs/source/index.rst index bbc62a8409..34222c0c22 100644 --- a/docs/source/index.rst +++ b/docs/source/index.rst @@ -37,6 +37,7 @@ Table of content Deployment Scheduler + History Server .. toctree:: :maxdepth: 1 diff --git a/docs/source/user-guide/history-server.md b/docs/source/user-guide/history-server.md new file mode 100644 index 0000000000..3294b92e1b --- /dev/null +++ b/docs/source/user-guide/history-server.md @@ -0,0 +1,146 @@ + + +# History Server + +The scheduler forgets a job shortly after it finishes. Completed jobs are cleaned +up after `finished_job_state_clean_up_interval_seconds`, and everything is gone +when the scheduler restarts, so by the time you want to look at a slow query it +is usually too late. + +The history server is Ballista's equivalent of the Spark History Server. When +event logging is enabled the scheduler writes a durable record of each job as it +runs, and the history server replays those records and serves the same `/api/*` +responses the live scheduler does. The existing TUI can point at it and browse +completed jobs with no scheduler running at all. + +## Enabling event logging + +Event logging is off by default. Start the scheduler with a directory to write +to: + +```shell +ballista-scheduler --event-log-dir /var/lib/ballista/history +``` + +The scheduler writes one file per job, `.eventlog`, in +[JSON Lines](https://jsonlines.org/) format. Files are appended as the job runs +and closed when it reaches a terminal state. + +Writes happen on a background task, so the scheduler's event loop never waits on +disk. If the queue backs up, progress records are dropped rather than allowed to +stall scheduling. The terminal record is the exception: it waits for queue +capacity, because a job missing it is invisible to the history server. + +## Running the history server + +Point it at the same directory: + +```shell +ballista-history-server \ + --event-log-dir /var/lib/ballista/history \ + --bind-host 0.0.0.0 \ + --bind-port 50060 +``` + +It indexes every completed log in the directory and serves them over the same +paths as the live scheduler: + +| Endpoint | Serves | +| ------------------------------ | ----------------------------------- | +| `GET /api/jobs` | every completed job, newest first | +| `GET /api/job/{job_id}` | one job's summary and plans | +| `GET /api/job/{job_id}/stages` | per-stage and per-task detail | +| `GET /api/job/{job_id}/config` | the session config the job ran with | +| `GET /api/job/{job_id}/dot` | the stage DAG in DOT format | + +Because the TUI talks to that same API, you can browse history with: + +```shell +BALLISTA__SCHEDULER__URL=http://localhost:50060 ballista-cli --tui +``` + +The TUI reads its URL from configuration rather than from the `--host` and +`--port` flags, which set the gRPC scheduler address used for running queries. +Pointing the TUI somewhere else means setting `BALLISTA__SCHEDULER__URL`, or +`scheduler.url` in the TUI's config file. Passing `--port 50060` alone leaves +the TUI on its default of `http://localhost:50050`, where it either finds your +live scheduler or reports that the scheduler is down. + +## Picking up new jobs + +The scheduler keeps writing to the directory while the history server is up, so +the directory is rescanned every `--update-interval-seconds` (10 by default) and +jobs that finished since the last pass are added to the list. This works whether +the scheduler writing the logs is the same process, a different one, or several +at once, and it does not matter whether the history server or the scheduler +started first — an event-log directory that does not exist yet is simply empty +until it appears. + +A rescan only opens logs whose size or modification time has changed, so the +cost of a pass over a directory that has not changed is one `stat` per file. +Lowering the interval makes new jobs show up sooner at the cost of more of those +passes; `--update-interval-seconds 0` turns rescanning off entirely and pins the +list to what was there at startup. + +Logs that have been deleted are dropped from the list on the next pass, so +pruning the directory does not leave behind entries that fail when opened. + +Only each job's summary is held in memory, which is what `GET /api/jobs` is +built from. Everything else is read back out of the job's log when you ask for +it. A job with many tasks stores megabytes of plan and per-task detail, and +keeping all of that resident for every job in the directory would put the +server's memory use at the mercy of how long you retain logs. + +## What is recorded + +Each log holds an ordered timeline: the job's submission, each stage starting +and ending, and each task finishing with its row counts and compute time. The +final record carries the finished API responses themselves. + +That last point is what makes replayed output trustworthy. The history server +does not rebuild a response from stored state; it re-serves the exact response +the scheduler built while the job was alive. There is no second implementation +that could drift from the live one. + +Only the final record is served today. The per-task timeline is recorded so a +future UI can show a job progressing rather than only its end state. + +## Operational notes + +- **Disk is not reclaimed automatically.** Logs accumulate until you remove + them. Size them against your job volume and prune with whatever you already + use for log rotation. +- **A corrupt log is skipped, not fatal.** If the scheduler dies mid-write the + affected file simply has no terminal record, so the history server ignores it + and still serves every other job. Damage confined to a job's stored responses + is only found when that job is opened, and shows up as a failed request for + that one job. +- **Deleting a log takes up to one rescan to show.** Until the next pass the + job is still listed, and opening it fails. +- **`GET /api/jobs` returns every job in one response.** There is no paging + yet, so a directory holding a very large number of jobs produces a large + response. Prune accordingly until paging exists. + ([#2270](https://github.com/apache/datafusion-ballista/issues/2270)) +- **Plans are rendered once, when the job ends.** The `?plan_format=` query + parameter therefore has no effect against a history server; it returns the + format captured at write time. +- **The history server has no cluster behind it.** `GET /api/executors` returns + an empty list and `GET /api/state` returns a static payload, so that TUI + screens expecting them still load.