feat(scheduler): add the history server

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<std::io::Error> 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<Option<ReplayedJob>, ReadError> {
+    Ok(read_job_end::<JobEnd>(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<Option<JobIndex>, ReadError> {
+    Ok(read_job_end::<JobEndIndex>(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<Option<ReplayedJob>, ReadError> {
+fn read_job_end<T: for<'de> Deserialize<'de>>(
+    path: &Path,
+) -> Result<Option<T>, 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<Option<ReplayedJob>, ReadError>
             });
         }
 
-        return match record.decode::<JobEnd>() {
-            Ok(end) => Ok(Some(ReplayedJob {
-                index: end.index,
-                job: end.job,
-                stages: end.stages,
-                config: end.config,
-                dot: end.dot,
-            })),
+        return match record.decode::<T>() {
+            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.
     ///

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 }

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<HistoryStore>) -> 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(())
+}