diff --git a/Cargo.lock b/Cargo.lock index 697df0a10f..cf2a092943 100644 --- a/Cargo.lock +++ b/Cargo.lock @@ -1207,6 +1207,18 @@ dependencies = [ "uuid", ] +[[package]] +name = "ballista-history" +version = "54.0.0" +dependencies = [ + "ballista-api-types", + "log", + "serde", + "serde_json", + "tempfile", + "tokio", +] + [[package]] name = "ballista-scheduler" version = "54.0.0" diff --git a/Cargo.toml b/Cargo.toml index 15817827b1..9869807174 100644 --- a/Cargo.toml +++ b/Cargo.toml @@ -19,10 +19,11 @@ exclude = ["dev/msrvcheck", "python"] members = [ "ballista-cli", + "ballista/api-types", "ballista/client", "ballista/core", "ballista/executor", - "ballista/api-types", + "ballista/history", "ballista/scheduler", "benchmarks", "chaos-testing", diff --git a/ballista/api-types/src/dto.rs b/ballista/api-types/src/dto.rs index 272730af84..094d8b6ddc 100644 --- a/ballista/api-types/src/dto.rs +++ b/ballista/api-types/src/dto.rs @@ -19,6 +19,7 @@ //! server, so both serialize byte-identical JSON. use serde::{Deserialize, Serialize}; +use std::collections::BTreeMap; /// Summary of one job, served by `GET /api/jobs` and `GET /api/job/{job_id}`. #[derive(Debug, Clone, Serialize, Deserialize)] @@ -164,3 +165,10 @@ pub enum PlanFormat { /// `?plan_format=metrics` => indent with aggregated metrics Metrics, } + +/// Session config for one job, as flat key/value pairs. Served by +/// `GET /api/job/{job_id}/config`. +/// +/// A `BTreeMap` so key order is deterministic: the same job must serialize +/// identically whether it is served live or replayed from a stored log. +pub type JobConfig = BTreeMap; diff --git a/ballista/history/Cargo.toml b/ballista/history/Cargo.toml new file mode 100644 index 0000000000..009bf0172c --- /dev/null +++ b/ballista/history/Cargo.toml @@ -0,0 +1,38 @@ +# 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. + +[package] +name = "ballista-history" +description = "Event-log schema, writer, and reader for the Ballista history server" +license = "Apache-2.0" +version = "54.0.0" +homepage = "https://datafusion.apache.org/ballista/" +repository = "https://github.com/apache/datafusion-ballista" +authors = ["Apache DataFusion "] +edition = { workspace = true } +rust-version = { workspace = true } + +[dependencies] +ballista-api-types = { path = "../api-types", version = "54.0.0" } +log = { workspace = true } +serde = { workspace = true, features = ["derive"] } +serde_json = { version = "1", features = ["raw_value"] } +tokio = { workspace = true, features = ["fs", "io-util", "macros", "rt", "sync"] } + +[dev-dependencies] +tempfile = { workspace = true } +tokio = { workspace = true, features = ["macros", "rt-multi-thread"] } diff --git a/ballista/history/src/event.rs b/ballista/history/src/event.rs new file mode 100644 index 0000000000..dcd00af976 --- /dev/null +++ b/ballista/history/src/event.rs @@ -0,0 +1,372 @@ +// 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. + +//! The on-disk event-log schema. One [`LogRecord`] is serialized per JSONL line. +//! +//! # Compatibility +//! +//! A log is written once and may be read years later by a much newer Ballista, +//! so the guarantee runs one way: **a reader accepts any record whose version is +//! less than or equal to its own**. There is no way to upgrade the writer of a +//! file that already exists. +//! +//! That is the opposite of `BALLISTA_PROTOCOL_VERSION`, the strict-equality +//! handshake between scheduler and executor. Both ends of that handshake are +//! live and upgraded together, so refusing to proceed is the safe move. Here +//! there is nothing to negotiate with. +//! +//! Three properties make the format survivable: +//! +//! 1. **Every line self-describes.** [`LogRecord`] carries `ev` and `version` +//! next to an opaque `data` payload, so a reader can decide whether it +//! understands a record before committing to its shape. +//! 2. **Unknown record types are skipped, not fatal.** A future scheduler can +//! add event kinds without breaking today's reader. +//! 3. **The served responses are stored verbatim**, as raw JSON rather than +//! typed structs. See [`JobEnd`]. + +use ballista_api_types::dto::{JobConfig, TaskStatus}; +use serde::{Deserialize, Serialize}; +use serde_json::value::RawValue; + +/// Current on-disk schema version, stamped on every record. +/// +/// Bump this only for a **breaking** change: removing a field, changing a +/// field's type, or changing the meaning of an existing one. Additive changes +/// do not bump it and must remain readable from older logs, which in practice +/// means every new field carries `#[serde(default)]`. +pub const SCHEMA_VERSION: u32 = 1; + +/// One line of the log: a self-describing envelope around an opaque payload. +/// +/// Keeping the payload opaque at this level is what lets a reader route on +/// `ev` and check `version` *before* attempting to parse a shape it may not +/// understand. Deserializing the whole record up front would collapse "written +/// by a newer Ballista" and "corrupt" into the same indistinguishable failure. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct LogRecord { + /// Record kind, e.g. `JobStart` or `JobEnd`. + pub ev: String, + /// Schema version this record was written with. See [`SCHEMA_VERSION`]. + pub version: u32, + /// Kind-specific payload, left unparsed. + pub data: Box, +} + +impl LogRecord { + /// Wrap a payload in an envelope stamped with the current schema version. + pub fn new(ev: &str, payload: &T) -> serde_json::Result { + Ok(LogRecord { + ev: ev.to_string(), + version: SCHEMA_VERSION, + data: serde_json::value::to_raw_value(payload)?, + }) + } + + /// Parse the payload as `T`. + pub fn decode Deserialize<'de>>(&self) -> serde_json::Result { + serde_json::from_str(self.data.get()) + } +} + +/// How a job reached its terminal state. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub enum JobEndStatus { + /// Job completed successfully. + Succeeded, + /// Job terminated with the given error. + Failed(String), + /// Job was cancelled before completing. + Cancelled, +} + +/// Metrics captured per finished task on the incremental timeline. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TaskEndMetrics { + /// Rows the task read. + pub input_rows: u64, + /// Rows the task produced. + pub output_rows: u64, + /// Time the task spent computing, in nanoseconds. + pub elapsed_compute_nanos: u64, +} + +/// Written when the scheduler accepts a job. First record in every log. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct JobStart { + /// Job identifier, matching the log's filename. + pub job_id: String, + /// Human-readable job name. + pub job_name: String, + /// When the job entered the queue. + pub queued_at: u64, + /// When the job was submitted for planning. + pub submitted_at: u64, + /// Rendered logical plan, if one was captured. + pub logical_plan: Option, + /// Rendered physical plan, if one was captured. + pub physical_plan: Option, +} + +/// A stage became runnable. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct StageStart { + /// Stage identifier within the job. + pub stage_id: u32, + /// Number of partitions the stage will produce. + pub partitions: u32, +} + +/// A stage reached a terminal state. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct StageEnd { + /// Stage identifier within the job. + pub stage_id: u32, + /// Terminal stage status. + pub status: String, +} + +/// A task finished, successfully or otherwise. +/// +/// Identifiers are fixed-width `u32` rather than `usize`: this is a durable, +/// cross-machine format, so a log written by a 64-bit scheduler must mean the +/// same thing to any reader. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct TaskEnd { + /// Stage the task belonged to. + pub stage_id: u32, + /// Task's slot within the stage. Under the multi-partition-task model a + /// task owns a slice of partitions, so this names the task rather than a + /// single partition. + pub task_id: u32, + /// Executor that ran the task. + pub executor_id: String, + /// Outcome of the task. + pub status: TaskStatus, + /// When the scheduler launched the task. + pub launch_time: u64, + /// When the executor began running it. + pub start_exec_time: u64, + /// When the executor finished it. + pub end_exec_time: u64, + /// Row counts and compute time for the task. + pub metrics: TaskEndMetrics, +} + +/// Frozen summary of a completed job, owned by this crate rather than shared +/// with the REST API. +/// +/// The history server needs *some* structure to list and sort jobs, but it does +/// not need to understand the full responses. Keeping that structure minimal and +/// local decouples the part that must stay readable forever from +/// `ballista-api-types`, which evolves with the live REST contract. +/// +/// These are exactly the fields `GET /api/jobs` renders, which is what lets the +/// history server build the job list without touching the payloads at all. +/// +/// Adding fields here later is fine; each one needs `#[serde(default)]` so older +/// logs still parse. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct JobIndex { + /// Job identifier. + pub job_id: String, + /// Human-readable job name. + pub job_name: String, + /// Plain status word, e.g. `Completed` or `Failed`. + pub status: String, + /// Verbose status, including completion or failure detail. + pub job_status: String, + /// When the job started executing. + pub start_time: u64, + /// When the job reached its terminal state. + pub end_time: u64, + /// Total number of stages in the job. + pub num_stages: usize, + /// Number of stages that finished successfully. + pub completed_stages: usize, + /// Progress as a percentage of completed stages. + pub percent_complete: u8, +} + +/// Terminal record, and the only one the history server serves from. +/// +/// `job` and `stages` hold the finished REST responses **as raw JSON**, not as +/// typed structs. That is deliberate. Those responses are +/// `ballista-api-types` shapes, which change with the live REST contract: +/// `TaskSummary::partition_id` went from `u32` to `Vec`, and +/// `TaskStatus::Failed` gained a field, both within a single release cycle. +/// +/// If this record stored them typed, a reader built after any such change would +/// fail to deserialize a log written before it, and the job would disappear. +/// Storing raw JSON means nothing ever parses the inner shape: the history +/// server relays the exact bytes the scheduler produced. The log is therefore +/// immune to REST type churn, and replayed output is byte-identical rather than +/// merely equivalent. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct JobEnd { + /// How the job ended. + pub status: JobEndStatus, + /// When the job entered the queue. + pub queued_at: u64, + /// When the job started executing. + pub started_at: u64, + /// When the job reached its terminal state. + pub completed_at: u64, + /// Frozen summary, used to list and sort jobs without parsing the payloads. + pub index: JobIndex, + /// Finished `GET /api/job/{job_id}` response, stored verbatim. + pub job: Box, + /// Finished `GET /api/job/{job_id}/stages` response, stored verbatim. + pub stages: Box, + /// Finished `GET /api/job/{job_id}/config` response. + pub config: JobConfig, + /// Rendered DOT graph of the stage DAG. + pub dot: String, +} + +/// Record-kind discriminators, as written to the `ev` field. +pub mod kind { + /// [`super::JobStart`] + pub const JOB_START: &str = "JobStart"; + /// [`super::StageStart`] + pub const STAGE_START: &str = "StageStart"; + /// [`super::StageEnd`] + pub const STAGE_END: &str = "StageEnd"; + /// [`super::TaskEnd`] + pub const TASK_END: &str = "TaskEnd"; + /// [`super::JobEnd`] + pub const JOB_END: &str = "JobEnd"; +} + +/// An event to append to a job's log. +/// +/// This is the in-memory shape callers build. It is encoded to a [`LogRecord`] +/// on the way to disk rather than serialized directly, so the envelope stays the +/// only thing a reader has to understand unconditionally. +#[derive(Debug, Clone)] +pub enum HistoryEvent { + /// See [`JobStart`]. + JobStart(JobStart), + /// See [`StageStart`]. + StageStart(StageStart), + /// See [`StageEnd`]. + StageEnd(StageEnd), + /// See [`TaskEnd`]. + TaskEnd(TaskEnd), + /// See [`JobEnd`]. + JobEnd(Box), +} + +impl HistoryEvent { + /// The `ev` discriminator this event is written with. + pub fn kind(&self) -> &'static str { + match self { + HistoryEvent::JobStart(_) => kind::JOB_START, + HistoryEvent::StageStart(_) => kind::STAGE_START, + HistoryEvent::StageEnd(_) => kind::STAGE_END, + HistoryEvent::TaskEnd(_) => kind::TASK_END, + HistoryEvent::JobEnd(_) => kind::JOB_END, + } + } + + /// Encode to the envelope written to disk. + pub fn to_record(&self) -> serde_json::Result { + match self { + HistoryEvent::JobStart(p) => LogRecord::new(self.kind(), p), + HistoryEvent::StageStart(p) => LogRecord::new(self.kind(), p), + HistoryEvent::StageEnd(p) => LogRecord::new(self.kind(), p), + HistoryEvent::TaskEnd(p) => LogRecord::new(self.kind(), p), + HistoryEvent::JobEnd(p) => LogRecord::new(self.kind(), p), + } + } +} + +#[cfg(test)] +mod tests { + use super::*; + use std::collections::BTreeMap; + + fn sample_job_end() -> JobEnd { + JobEnd { + status: JobEndStatus::Succeeded, + queued_at: 5, + started_at: 10, + completed_at: 20, + index: JobIndex { + job_id: "job-1".into(), + job_name: "q1".into(), + status: "Completed".into(), + job_status: "COMPLETED".into(), + start_time: 10, + end_time: 20, + num_stages: 1, + completed_stages: 1, + percent_complete: 100, + }, + job: RawValue::from_string(r#"{"job_id":"job-1"}"#.to_string()).unwrap(), + stages: RawValue::from_string(r#"{"stages":[]}"#.to_string()).unwrap(), + config: BTreeMap::from([("k".to_string(), "v".to_string())]), + dot: "digraph {}".into(), + } + } + + #[test] + fn job_end_round_trips_through_jsonl() { + let event = HistoryEvent::JobEnd(Box::new(sample_job_end())); + let line = serde_json::to_string(&event.to_record().unwrap()).unwrap(); + + assert!(line.contains(r#""ev":"JobEnd""#)); + assert!(line.contains(r#""version":1"#)); + + let record: LogRecord = serde_json::from_str(&line).unwrap(); + assert_eq!(record.ev, kind::JOB_END); + let back: JobEnd = record.decode().unwrap(); + assert_eq!(back.index.job_id, "job-1"); + } + + /// The stored responses are relayed as raw JSON and never round-tripped + /// through a typed struct, so a log stays readable when the REST types it + /// came from change shape. This is the property that keeps old logs alive + /// across releases. + #[test] + fn stored_payloads_survive_shapes_this_build_cannot_model() { + let mut end = sample_job_end(); + end.job = RawValue::from_string( + r#"{"a_field_from_the_future":[1,2,3],"partition_id":{"nested":true}}"# + .to_string(), + ) + .unwrap(); + let line = serde_json::to_string(&end).unwrap(); + + let back: JobEnd = serde_json::from_str(&line).unwrap(); + assert!(back.job.get().contains("a_field_from_the_future")); + // Byte-for-byte, not merely equivalent. + assert_eq!(back.job.get(), end.job.get()); + } + + /// The envelope is readable without knowing the payload's shape at all, + /// which is what lets a reader check the version before parsing. + #[test] + fn envelope_is_readable_without_understanding_the_payload() { + let line = + r#"{"ev":"SomeFutureEvent","version":99,"data":{"anything":[1,{"x":null}]}}"#; + let record: LogRecord = serde_json::from_str(line).unwrap(); + assert_eq!(record.ev, "SomeFutureEvent"); + assert_eq!(record.version, 99); + assert!(record.data.get().contains("anything")); + } +} diff --git a/ballista/history/src/lib.rs b/ballista/history/src/lib.rs new file mode 100644 index 0000000000..814783a8c8 --- /dev/null +++ b/ballista/history/src/lib.rs @@ -0,0 +1,58 @@ +// 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. + +#![warn(missing_docs)] + +//! Durable event logs for completed Ballista jobs. +//! +//! The scheduler appends one JSONL record per event to +//! `/.eventlog` while a job runs. The history server +//! reads those files back and serves the same `/api/*` responses the live +//! scheduler does, long after the scheduler has forgotten the job. +//! +//! # Write once, replay verbatim +//! +//! The terminal [`event::JobEnd`] record embeds the finished `/api/*` responses +//! the scheduler built from its live execution graph, stored as raw JSON. Replay +//! relays those bytes unchanged, so the history server never re-derives a +//! response and there is no second implementation to drift. +//! +//! The earlier records ([`event::JobStart`], [`event::StageStart`], +//! [`event::StageEnd`], [`event::TaskEnd`]) form an incremental timeline. +//! Nothing reads them yet; they exist so a future UI can show a job progressing +//! rather than only its final state. +//! +//! # Compatibility +//! +//! Logs outlive the binaries that wrote them, so a newer reader must keep +//! reading older logs indefinitely. [`event::SCHEMA_VERSION`] documents the +//! policy, [`event::LogRecord`] is the self-describing envelope that makes it +//! enforceable, and `testdata/schema-v1.eventlog` is a frozen log that CI +//! replays on every build to catch regressions. +//! +//! # Durability +//! +//! [`writer::EventLogWriter`] does all file I/O on a background task, so the +//! scheduler's event loop never waits on a disk write. Timeline events are +//! dropped rather than allowed to block if the queue backs up, on the grounds +//! that losing a progress record is better than stalling scheduling. The +//! terminal `JobEnd` is the exception: it waits for queue capacity, because a +//! job missing its `JobEnd` is invisible to the history server. + +pub mod event; +pub mod reader; +pub mod writer; diff --git a/ballista/history/src/reader.rs b/ballista/history/src/reader.rs new file mode 100644 index 0000000000..36382e648e --- /dev/null +++ b/ballista/history/src/reader.rs @@ -0,0 +1,373 @@ +// 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. + +//! Reads a completed `.eventlog` into the payload the history server +//! serves. A file is "completed" once it contains a `JobEnd` record. +//! +//! Reading is deliberately forgiving about lines it does not recognise and +//! deliberately loud about a `JobEnd` it cannot use. An unreadable terminal +//! record means a job that exists on disk will not appear in the UI, and that +//! must never be indistinguishable from a job that is simply still running. + +use crate::event::{JobEnd, JobIndex, LogRecord, SCHEMA_VERSION, kind}; +use ballista_api_types::dto::JobConfig; +use serde_json::value::RawValue; +use std::io::BufRead; +use std::path::Path; + +/// The served payload recovered from a completed job's event log. +#[derive(Debug, Clone)] +pub struct ReplayedJob { + /// Frozen summary, used to list and sort jobs. + pub index: JobIndex, + /// `GET /api/job/{job_id}` response, verbatim as the scheduler wrote it. + pub job: Box, + /// `GET /api/job/{job_id}/stages` response, verbatim. + pub stages: Box, + /// `GET /api/job/{job_id}/config` response. + pub config: JobConfig, + /// Rendered DOT graph of the stage DAG. + pub dot: String, +} + +/// Why a log that exists on disk yielded no servable job. +#[derive(Debug)] +pub enum ReadError { + /// The file could not be read at all. + Io(std::io::Error), + /// A `JobEnd` record is present but was written by a newer schema than + /// this build understands. + UnsupportedVersion { + /// Version stamped on the record. + found: u32, + /// Highest version this build can read. + supported: u32, + }, + /// A `JobEnd` record is present and claims a supported version, but could + /// not be deserialized. Distinct from a missing `JobEnd`: this is a real + /// problem worth surfacing, not a job that is still running. + Malformed(String), +} + +impl std::fmt::Display for ReadError { + fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result { + match self { + ReadError::Io(e) => write!(f, "{e}"), + ReadError::UnsupportedVersion { found, supported } => write!( + f, + "event log schema version {found} is newer than the highest \ + supported version {supported}; upgrade the history server" + ), + ReadError::Malformed(e) => write!(f, "malformed JobEnd record: {e}"), + } + } +} + +impl std::error::Error for ReadError {} + +impl From for ReadError { + fn from(e: std::io::Error) -> Self { + ReadError::Io(e) + } +} + +/// Read a completed job's payload out of its event log. +/// +/// Returns `Ok(None)` only when the log genuinely has no `JobEnd` record, which +/// 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. +/// +/// 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> { + let file = std::fs::File::open(path)?; + let reader = std::io::BufReader::new(file); + + for line in reader.lines() { + let line = line?; + if line.trim().is_empty() { + continue; + } + + // Route on the envelope alone. A line we cannot even read an envelope + // from is treated as a record we do not understand, not as a failure. + let Ok(record) = serde_json::from_str::(&line) else { + continue; + }; + if record.ev != kind::JOB_END { + continue; + } + + if record.version > SCHEMA_VERSION { + return Err(ReadError::UnsupportedVersion { + found: record.version, + supported: SCHEMA_VERSION, + }); + } + + return match record.decode::() { + Ok(end) => Ok(Some(ReplayedJob { + index: end.index, + job: end.job, + stages: end.stages, + config: end.config, + dot: end.dot, + })), + Err(e) => Err(ReadError::Malformed(e.to_string())), + }; + } + + Ok(None) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::event::{HistoryEvent, JobEnd, JobEndStatus, SCHEMA_VERSION}; + use std::io::Write; + + fn sample_index() -> JobIndex { + JobIndex { + job_id: "job-1".into(), + job_name: "q1".into(), + status: "Completed".into(), + job_status: "COMPLETED".into(), + start_time: 2, + end_time: 3, + num_stages: 1, + completed_stages: 1, + percent_complete: 100, + } + } + + fn job_end_line() -> String { + let event = HistoryEvent::JobEnd(Box::new(JobEnd { + status: JobEndStatus::Succeeded, + queued_at: 1, + started_at: 2, + completed_at: 3, + index: sample_index(), + job: RawValue::from_string( + r#"{"job_id":"job-1","physical_plan":"ProjectionExec"}"#.to_string(), + ) + .unwrap(), + stages: RawValue::from_string(r#"{"stages":[]}"#.to_string()).unwrap(), + config: Default::default(), + dot: "digraph {}".into(), + })); + serde_json::to_string(&event.to_record().unwrap()).unwrap() + } + + fn write_log( + dir: &tempfile::TempDir, + name: &str, + lines: &[&str], + ) -> std::path::PathBuf { + let path = dir.path().join(name); + let mut f = std::fs::File::create(&path).unwrap(); + for line in lines { + writeln!(f, "{line}").unwrap(); + } + path + } + + #[test] + fn reads_job_end_and_ignores_unknown_timeline_lines() { + let dir = tempfile::tempdir().unwrap(); + let path = write_log( + &dir, + "job-1.eventlog", + &[ + r#"{"ev":"StageStart","version":1,"data":{"stage_id":1,"partitions":4}}"#, + &job_end_line(), + ], + ); + + let replayed = read_completed_job(&path).unwrap().expect("completed"); + assert_eq!(replayed.index.job_id, "job-1"); + assert_eq!(replayed.dot, "digraph {}"); + // The payload is re-served verbatim rather than round-tripped through a + // typed struct. + assert!(replayed.job.get().contains("ProjectionExec")); + } + + #[test] + fn returns_none_when_no_job_end() { + let dir = tempfile::tempdir().unwrap(); + let path = write_log( + &dir, + "job-2.eventlog", + &[r#"{"ev":"StageStart","version":1,"data":{"stage_id":1,"partitions":4}}"#], + ); + assert!(read_completed_job(&path).unwrap().is_none()); + } + + /// A record type this build has never heard of must not stop it reading the + /// rest of the file. This is the forward-compatibility guarantee: an older + /// history server keeps working against logs from a newer scheduler, as + /// long as the schema version itself has not been bumped. + #[test] + fn skips_unrecognized_record_types() { + let dir = tempfile::tempdir().unwrap(); + let path = write_log( + &dir, + "job-3.eventlog", + &[ + r#"{"ev":"SomeFutureEvent","version":1,"data":{"whatever":true}}"#, + &job_end_line(), + ], + ); + assert!(read_completed_job(&path).unwrap().is_some()); + } + + /// A newer schema version is reported as such rather than being silently + /// skipped, so an operator finds out their history server is too old. + #[test] + fn newer_schema_version_is_reported_not_skipped() { + let dir = tempfile::tempdir().unwrap(); + let line = format!( + r#"{{"ev":"JobEnd","version":{},"data":{{}}}}"#, + SCHEMA_VERSION + 1 + ); + let path = write_log(&dir, "job-4.eventlog", &[&line]); + + match read_completed_job(&path) { + Err(ReadError::UnsupportedVersion { found, supported }) => { + assert_eq!(found, SCHEMA_VERSION + 1); + assert_eq!(supported, SCHEMA_VERSION); + } + other => panic!("expected UnsupportedVersion, got {other:?}"), + } + } + + /// A corrupt terminal record must be distinguishable from a job that has + /// simply not finished. Both used to surface as `Ok(None)`, which meant a + /// job could vanish from the UI with nothing logged. + #[test] + fn malformed_job_end_is_an_error_not_a_missing_job() { + let dir = tempfile::tempdir().unwrap(); + let path = write_log( + &dir, + "job-5.eventlog", + &[r#"{"ev":"JobEnd","version":1,"data":{"status":"Succeeded"}}"#], + ); + + match read_completed_job(&path) { + Err(ReadError::Malformed(_)) => {} + other => panic!("expected Malformed, got {other:?}"), + } + } +} + +/// Compatibility tests against a checked-in log from an earlier schema version. +/// +/// `testdata/schema-v1.eventlog` is a **frozen artifact**. It is never +/// regenerated: the whole point is that it was written by an older Ballista and +/// must stay readable forever. If a change here makes this module fail, the +/// change breaks every event log already on disk in the field. +/// +/// Fixing such a failure by editing the fixture defeats the test. The options +/// are to make the change backward-compatible (usually `#[serde(default)]` on a +/// new field), or to bump [`SCHEMA_VERSION`] and keep a path that can still read +/// the old version. +#[cfg(test)] +mod compatibility { + use super::*; + + fn golden_v1() -> std::path::PathBuf { + std::path::Path::new(env!("CARGO_MANIFEST_DIR")) + .join("testdata") + .join("schema-v1.eventlog") + } + + #[test] + fn reads_a_v1_log_written_by_an_earlier_ballista() { + let replayed = read_completed_job(&golden_v1()) + .expect("a v1 log must remain readable") + .expect("the fixture contains a JobEnd record"); + + assert_eq!(replayed.index.job_id, "golden-v1"); + assert_eq!(replayed.index.job_name, "tpch-q1"); + assert_eq!(replayed.index.status, "Completed"); + assert_eq!(replayed.index.start_time, 1005); + assert_eq!(replayed.index.end_time, 1100); + assert_eq!( + replayed + .config + .get("datafusion.execution.target_partitions"), + Some(&"4".to_string()) + ); + assert!(replayed.dot.contains("digraph")); + } + + /// The stored responses must come back byte-for-byte, because that is what + /// lets the history server re-serve them without understanding them. + /// + /// Compared against the raw file text rather than a parsed + /// `serde_json::Value`: `Value` sorts object keys, so round-tripping + /// through it would reorder the payload and hide exactly the property + /// under test. + #[test] + fn v1_payloads_are_relayed_verbatim() { + let raw = std::fs::read_to_string(golden_v1()).unwrap(); + let replayed = read_completed_job(&golden_v1()).unwrap().unwrap(); + + assert!( + raw.contains(replayed.job.get()), + "the job payload must appear verbatim in the log, got: {}", + replayed.job.get() + ); + assert!( + raw.contains(replayed.stages.get()), + "the stages payload must appear verbatim in the log" + ); + // Key order survives, which a typed round trip would not preserve. + assert!( + replayed.job.get().starts_with(r#"{"job_id":"golden-v1""#), + "original key order must be preserved, got: {}", + replayed.job.get() + ); + } + + /// The v1 fixture deliberately contains a record kind this build has never + /// heard of, standing in for one a future scheduler might write. An older + /// reader must step over it and still find the terminal record. + #[test] + fn unknown_record_kinds_in_a_v1_log_do_not_break_reading() { + let raw = std::fs::read_to_string(golden_v1()).unwrap(); + assert!( + raw.contains("AnEventKindFromTheFuture"), + "fixture should exercise the unknown-record path" + ); + assert!(read_completed_job(&golden_v1()).unwrap().is_some()); + } + + /// The stages payload in the fixture carries `partition_id` as an array, + /// the shape that broke the TUI when it was declared as a scalar (#2257). + /// Storing payloads opaquely means a change like that can never make an + /// existing log unreadable. + #[test] + fn a_rest_shape_change_cannot_orphan_a_stored_log() { + let replayed = read_completed_job(&golden_v1()).unwrap().unwrap(); + assert!( + replayed.stages.get().contains(r#""partition_id":[0,1]"#), + "fixture should carry the multi-partition shape verbatim" + ); + } +} diff --git a/ballista/history/src/writer.rs b/ballista/history/src/writer.rs new file mode 100644 index 0000000000..ccf9939889 --- /dev/null +++ b/ballista/history/src/writer.rs @@ -0,0 +1,309 @@ +// 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. + +//! Async, buffered event-log writer. Each job's events append to +//! `/.eventlog` as JSONL. Appends are non-blocking; a background +//! task performs the file I/O so the scheduler hot path never waits on disk. + +use crate::event::HistoryEvent; +use std::collections::HashMap; +use std::path::{Path, PathBuf}; +use tokio::io::AsyncWriteExt; +use tokio::sync::{mpsc, oneshot}; + +enum WriterMsg { + Event { + job_id: String, + event: Box, + }, + Flush { + job_id: String, + done: oneshot::Sender<()>, + }, + Finish { + job_id: String, + done: oneshot::Sender<()>, + }, +} + +/// Handle to the background event-log writer task. +/// +/// Cloning is cheap and every clone feeds the same task, so the scheduler can +/// hand one to each component that needs to record events. +#[derive(Clone)] +pub struct EventLogWriter { + tx: mpsc::Sender, +} + +impl EventLogWriter { + /// Spawn the writer task, appending logs under `log_dir`. + /// + /// `buffer` bounds the in-flight event queue; beyond it, timeline events are + /// dropped rather than allowed to stall the caller. + pub fn new(log_dir: PathBuf, buffer: usize) -> EventLogWriter { + let (tx, rx) = mpsc::channel(buffer.max(1)); + tokio::spawn(run(log_dir, rx)); + EventLogWriter { tx } + } + + /// Enqueue an event for `job_id`. Never blocks; drops (with a warning) if the + /// channel is full, so logging cannot stall scheduling. + pub fn append(&self, job_id: &str, event: HistoryEvent) { + let msg = WriterMsg::Event { + job_id: job_id.to_string(), + event: Box::new(event), + }; + if self.tx.try_send(msg).is_err() { + log::warn!( + "event-log writer: dropping event for {job_id} (channel full or closed)" + ); + } + } + + /// Await all currently-enqueued writes for `job_id` (best effort). + pub async fn flush_job(&self, job_id: &str) { + let (done, wait) = oneshot::channel(); + if self + .tx + .send(WriterMsg::Flush { + job_id: job_id.to_string(), + done, + }) + .await + .is_ok() + { + let _ = wait.await; + } + } + + /// Enqueue a terminal event (e.g. `JobEnd`) for `job_id`. Unlike `append`, this + /// awaits channel capacity instead of dropping the event when the channel is + /// full, so the terminal record is never silently lost. Still best-effort at + /// the process boundary: if the channel is closed (background task gone) this + /// logs and returns rather than panicking. + pub async fn append_final(&self, job_id: &str, event: HistoryEvent) { + let msg = WriterMsg::Event { + job_id: job_id.to_string(), + event: Box::new(event), + }; + if self.tx.send(msg).await.is_err() { + log::warn!( + "event-log writer: failed to enqueue terminal event for {job_id} (channel closed)" + ); + } + } + + /// Flush and close the per-job file handle for `job_id`. Must be called after + /// the terminal event has been enqueued (e.g. via `append_final`) so it is + /// ordered after it on the single-consumer FIFO channel. Best effort: if the + /// channel is closed this logs and returns. + pub async fn finish_job(&self, job_id: &str) { + let (done, wait) = oneshot::channel(); + if self + .tx + .send(WriterMsg::Finish { + job_id: job_id.to_string(), + done, + }) + .await + .is_ok() + { + let _ = wait.await; + } else { + log::warn!( + "event-log writer: failed to enqueue finish for {job_id} (channel closed)" + ); + } + } +} + +async fn run(log_dir: PathBuf, mut rx: mpsc::Receiver) { + if let Err(e) = tokio::fs::create_dir_all(&log_dir).await { + log::warn!("event-log writer: cannot create {}: {e}", log_dir.display()); + return; + } + // One open append handle per job for the life of the process. + let mut handles: HashMap = HashMap::new(); + + while let Some(msg) = rx.recv().await { + match msg { + WriterMsg::Event { job_id, event } => { + let file = match open_for(&log_dir, &mut handles, &job_id).await { + Some(f) => f, + None => continue, + }; + match event.to_record().and_then(|r| serde_json::to_string(&r)) { + Ok(mut line) => { + line.push('\n'); + if let Err(e) = file.write_all(line.as_bytes()).await { + log::warn!( + "event-log writer: write failed for {job_id}: {e}" + ); + } + } + Err(e) => log::warn!("event-log writer: serialize failed: {e}"), + } + } + WriterMsg::Flush { job_id, done } => { + if let Some(file) = handles.get_mut(&job_id) { + let _ = file.flush().await; + } + let _ = done.send(()); + } + WriterMsg::Finish { job_id, done } => { + if let Some(mut file) = handles.remove(&job_id) { + let _ = file.flush().await; + // Dropping `file` here closes the fd. + } + let _ = done.send(()); + } + } + } +} + +async fn open_for<'a>( + log_dir: &Path, + handles: &'a mut HashMap, + job_id: &str, +) -> Option<&'a mut tokio::fs::File> { + 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) => { + log::warn!("event-log writer: cannot open {}: {e}", path.display()); + return None; + } + } + } + handles.get_mut(job_id) +} + +#[cfg(test)] +mod tests { + use super::*; + use crate::event::{ + HistoryEvent, JobEnd, JobEndStatus, JobIndex, JobStart, StageStart, + }; + use serde_json::value::RawValue; + use std::collections::BTreeMap; + + #[tokio::test] + async fn terminal_job_end_is_not_dropped_on_a_saturated_channel() { + let dir = tempfile::tempdir().unwrap(); + // Tiny buffer so the non-blocking `append` path would readily drop events + // under load; `append_final` must still guarantee delivery. + let writer = EventLogWriter::new(dir.path().to_path_buf(), 1); + + writer.append( + "job-1", + HistoryEvent::JobStart(JobStart { + job_id: "job-1".into(), + job_name: "q1".into(), + queued_at: 1, + submitted_at: 2, + logical_plan: None, + physical_plan: None, + }), + ); + for stage_id in 0..10 { + writer.append( + "job-1", + HistoryEvent::StageStart(StageStart { + stage_id, + partitions: 4, + }), + ); + } + + let job_end = HistoryEvent::JobEnd(Box::new(JobEnd { + status: JobEndStatus::Succeeded, + queued_at: 1, + started_at: 2, + completed_at: 20, + index: JobIndex { + job_id: "job-1".into(), + job_name: "q1".into(), + status: "Completed".into(), + job_status: "COMPLETED".into(), + start_time: 10, + end_time: 20, + num_stages: 1, + completed_stages: 1, + percent_complete: 100, + }, + job: RawValue::from_string(r#"{"job_id":"job-1"}"#.to_string()).unwrap(), + stages: RawValue::from_string(r#"{"stages":[]}"#.to_string()).unwrap(), + config: BTreeMap::new(), + dot: "digraph {}".into(), + })); + writer.append_final("job-1", job_end).await; + writer.finish_job("job-1").await; + + let path = dir.path().join("job-1.eventlog"); + let contents = tokio::fs::read_to_string(&path).await.unwrap(); + let lines: Vec<&str> = contents.lines().collect(); + assert!( + lines.iter().any(|l| l.contains("\"ev\":\"JobEnd\"")), + "expected a JobEnd line in the event log, got: {contents}" + ); + assert_eq!( + lines.last().map(|l| l.contains("\"ev\":\"JobEnd\"")), + Some(true), + "JobEnd should be the last line written" + ); + } + + #[tokio::test] + async fn append_writes_one_jsonl_line_per_event() { + let dir = tempfile::tempdir().unwrap(); + let writer = EventLogWriter::new(dir.path().to_path_buf(), 16); + writer.append( + "job-1", + HistoryEvent::JobStart(JobStart { + job_id: "job-1".into(), + job_name: "q1".into(), + queued_at: 1, + submitted_at: 2, + logical_plan: None, + physical_plan: None, + }), + ); + writer.append( + "job-1", + HistoryEvent::StageStart(StageStart { + stage_id: 1, + partitions: 4, + }), + ); + writer.flush_job("job-1").await; + + let path = dir.path().join("job-1.eventlog"); + let contents = tokio::fs::read_to_string(&path).await.unwrap(); + let lines: Vec<&str> = contents.lines().collect(); + assert_eq!(lines.len(), 2); + assert!(lines[0].contains("\"ev\":\"JobStart\"")); + assert!(lines[1].contains("\"ev\":\"StageStart\"")); + } +} diff --git a/ballista/history/testdata/schema-v1.eventlog b/ballista/history/testdata/schema-v1.eventlog new file mode 100644 index 0000000000..9d15a69e84 --- /dev/null +++ b/ballista/history/testdata/schema-v1.eventlog @@ -0,0 +1,7 @@ +{"ev":"JobStart","version":1,"data":{"job_id":"golden-v1","job_name":"tpch-q1","queued_at":1000,"submitted_at":1005,"logical_plan":"Projection: l_returnflag","physical_plan":"ProjectionExec: expr=[l_returnflag@0 as l_returnflag]"}} +{"ev":"StageStart","version":1,"data":{"stage_id":1,"partitions":4}} +{"ev":"TaskEnd","version":1,"data":{"stage_id":1,"task_id":0,"executor_id":"executor-a","status":"Successful","launch_time":1010,"start_exec_time":1012,"end_exec_time":1040,"metrics":{"input_rows":1000,"output_rows":250,"elapsed_compute_nanos":28000000}}} +{"ev":"TaskEnd","version":1,"data":{"stage_id":1,"task_id":1,"executor_id":"executor-b","status":{"Failed":{"reason":"ExecutionError","error":"divide by zero"}},"launch_time":1010,"start_exec_time":1013,"end_exec_time":1020,"metrics":{"input_rows":0,"output_rows":0,"elapsed_compute_nanos":0}}} +{"ev":"StageEnd","version":1,"data":{"stage_id":1,"status":"Successful"}} +{"ev":"AnEventKindFromTheFuture","version":1,"data":{"whatever":[1,2,{"nested":true}]}} +{"ev":"JobEnd","version":1,"data":{"status":"Succeeded","queued_at":1000,"started_at":1005,"completed_at":1100,"index":{"job_id":"golden-v1","job_name":"tpch-q1","status":"Completed","job_status":"Completed. Produced 4 partitions containing 1000 rows. Elapsed time: 95 ms.","start_time":1005,"end_time":1100,"num_stages":1,"completed_stages":1,"percent_complete":100},"job":{"job_id":"golden-v1","job_name":"tpch-q1","job_status":"Completed. Produced 4 partitions containing 1000 rows. Elapsed time: 95 ms.","status":"Completed","num_stages":1,"completed_stages":1,"percent_complete":100,"start_time":1005,"end_time":1100,"logical_plan":"Projection: l_returnflag","physical_plan":"ProjectionExec: expr=[l_returnflag@0 as l_returnflag]","stage_plan":"ShuffleWriterExec"},"stages":{"stages":[{"stage_id":"1","stage_status":"Successful","input_rows":1000,"output_rows":250,"elapsed_compute":"28ms","stage_plan":"ProjectionExec","task_duration_percentiles":{"min":7,"p25":7,"median":28,"p75":28,"max":28},"task_input_percentiles":{"min":0,"p25":0,"median":1000,"p75":1000,"max":1000},"tasks":[{"id":0,"status":"Successful","partition_id":[0,1],"scheduled_time":1008,"launch_time":1010,"start_exec_time":1012,"end_exec_time":1040,"exec_duration":28,"finish_time":1042,"input_rows":1000,"output_rows":250}]}]},"config":{"datafusion.execution.batch_size":"8192","datafusion.execution.target_partitions":"4"},"dot":"digraph G {\n stage_1;\n}"}} diff --git a/dev/release/README.md b/dev/release/README.md index c6b0de31c0..502b0a71af 100644 --- a/dev/release/README.md +++ b/dev/release/README.md @@ -288,6 +288,7 @@ of the following crates: - [ballista-core](https://crates.io/crates/ballista-core) - [ballista-executor](https://crates.io/crates/ballista-executor) - [ballista-api-types](https://crates.io/crates/ballista-api-types) +- [ballista-history](https://crates.io/crates/ballista-history) - [ballista-scheduler](https://crates.io/crates/ballista-scheduler) Download and unpack the official release tarball @@ -308,6 +309,7 @@ dot -Tsvg dev/release/crate-deps.dot > dev/release/crate-deps.svg (cd ballista/core && cargo publish) (cd ballista/executor && cargo publish) (cd ballista/api-types && cargo publish) +(cd ballista/history && cargo publish) (cd ballista/scheduler && cargo publish) (cd ballista/client && cargo publish) (cd ballista-cli && cargo publish) diff --git a/dev/release/crate-deps.dot b/dev/release/crate-deps.dot index 27bee709ba..637030aca5 100644 --- a/dev/release/crate-deps.dot +++ b/dev/release/crate-deps.dot @@ -19,6 +19,7 @@ digraph G { ballista_core ballista_api_types + ballista_history ballista_scheduler ballista_executor ballista @@ -27,6 +28,8 @@ digraph G { ballista_scheduler -> ballista_core ballista_scheduler -> ballista_api_types + ballista_history -> ballista_api_types + ballista_executor -> ballista_core ballista -> ballista_core diff --git a/dev/update_ballista_versions.py b/dev/update_ballista_versions.py index ca5bef3bf8..926520f42d 100755 --- a/dev/update_ballista_versions.py +++ b/dev/update_ballista_versions.py @@ -44,6 +44,7 @@ def update_cargo_toml(cargo_toml: str, new_version: str): 'ballista-core', 'ballista-executor', 'ballista-api-types', + 'ballista-history', 'ballista-scheduler', 'ballista-cli', ) @@ -82,6 +83,7 @@ def main(): 'ballista-cli', 'ballista/core', 'ballista/api-types', + 'ballista/history', 'ballista/scheduler', 'ballista/executor', 'ballista/client',