feat: event log and history server (Spark History Server equivalent)

Cargo.toml

@@ -17,7 +17,7 @@
 
 [workspace]
 exclude = ["dev/msrvcheck", "python"]
-members = ["ballista-cli", "ballista/client", "ballista/core", "ballista/executor", "ballista/scheduler", "benchmarks", "examples"]
+members = ["ballista-cli", "ballista/client", "ballista/core", "ballista/executor", "ballista/history", "ballista/scheduler", "benchmarks", "examples"]
 resolver = "3"
 
 [workspace.package]

ballista/history/Cargo.toml

@@ -0,0 +1,33 @@
+# 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 Ballista's history server"
+license = "Apache-2.0"
+version = "53.0.0"
+edition = { workspace = true }
+rust-version = { workspace = true }
+
+[dependencies]
+serde = { workspace = true, features = ["derive"] }
+serde_json = "1"
+tokio = { workspace = true, features = ["rt", "sync", "fs", "io-util", "macros"] }
+
+[dev-dependencies]
+tokio = { workspace = true, features = ["rt", "sync", "fs", "io-util", "macros", "rt-multi-thread"] }
+tempfile = { workspace = true }

ballista/history/src/dto.rs

@@ -0,0 +1,97 @@
+// 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.
+
+//! REST response DTOs shared by the live scheduler API handlers and the history
+//! server, so both serialize byte-identical JSON.
+
+use serde::{Deserialize, Serialize};
+use std::collections::BTreeMap;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct JobResponse {
+    pub job_id: String,
+    pub job_name: String,
+    pub job_status: String,
+    pub status: String,
+    pub num_stages: usize,
+    pub completed_stages: usize,
+    pub percent_complete: u8,
+    pub start_time: u64,
+    pub end_time: u64,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub logical_plan: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub physical_plan: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub stage_plan: Option<String>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum TaskStatus {
+    Running,
+    Successful,
+    Failed,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct TaskSummary {
+    pub id: usize,
+    pub status: TaskStatus,
+    pub partition_id: u32,
+    pub scheduled_time: u64,
+    pub launch_time: u64,
+    pub start_exec_time: u64,
+    pub end_exec_time: u64,
+    pub exec_duration: u64,
+    pub finish_time: u64,
+    pub input_rows: usize,
+    pub output_rows: usize,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct Percentiles {
+    pub min: u64,
+    pub p25: u64,
+    pub median: u64,
+    pub p75: u64,
+    pub max: u64,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct QueryStageSummary {
+    pub stage_id: String,
+    pub stage_status: String,
+    pub input_rows: usize,
+    pub output_rows: usize,
+    pub elapsed_compute: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub stage_plan: Option<String>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub task_duration_percentiles: Option<Percentiles>,
+    #[serde(skip_serializing_if = "Option::is_none")]
+    pub task_input_percentiles: Option<Percentiles>,
+    pub tasks: Vec<Option<TaskSummary>>,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct QueryStagesResponse {
+    pub stages: Vec<QueryStageSummary>,
+}
+
+/// Session config as flat key/value pairs (from `SessionConfig::to_props()`),
+/// sorted for stable output.
+pub type JobConfig = BTreeMap<String, String>;

ballista/history/src/event.rs

@@ -0,0 +1,132 @@
+// 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 `HistoryEvent` is serialized per JSONL line.
+//! This is a frozen public projection of the scheduler's internal events; the
+//! embedded DTOs are the stable contract the history server serves.
+
+use crate::dto::{JobConfig, JobResponse, QueryStagesResponse, TaskStatus};
+use serde::{Deserialize, Serialize};
+
+/// Current on-disk schema version, stamped on `JobStart`/`JobEnd`.
+pub const SCHEMA_VERSION: u32 = 1;
+
+/// How a job reached its terminal state.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum JobEndStatus {
+    Succeeded,
+    Failed(String),
+    Cancelled,
+}
+
+/// Metrics captured per finished task on the incremental timeline.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct TaskEndMetrics {
+    pub input_rows: u64,
+    pub output_rows: u64,
+    pub elapsed_compute_nanos: u64,
+}
+
+/// A single record in a job's event log.
+///
+/// Stage and partition identifiers are fixed-width (`u32`) throughout 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. Callers holding the
+/// scheduler's own `usize` stage ids cast on the way in.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(tag = "ev")]
+pub enum HistoryEvent {
+    JobStart {
+        version: u32,
+        job_id: String,
+        job_name: String,
+        queued_at: u64,
+        submitted_at: u64,
+        logical_plan: Option<String>,
+        physical_plan: Option<String>,
+    },
+    StageStart {
+        stage_id: u32,
+        partitions: u32,
+    },
+    StageEnd {
+        stage_id: u32,
+        status: String,
+    },
+    TaskEnd {
+        stage_id: u32,
+        partition: u32,
+        executor_id: String,
+        status: TaskStatus,
+        launch_time: u64,
+        start_exec_time: u64,
+        end_exec_time: u64,
+        metrics: TaskEndMetrics,
+    },
+    JobEnd {
+        version: u32,
+        status: JobEndStatus,
+        queued_at: u64,
+        started_at: u64,
+        completed_at: u64,
+        job: Box<JobResponse>,
+        stages: Box<QueryStagesResponse>,
+        config: JobConfig,
+        dot: String,
+    },
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+    use crate::dto::{JobResponse, QueryStagesResponse};
+    use std::collections::BTreeMap;
+
+    #[test]
+    fn job_end_round_trips_through_jsonl() {
+        let job = JobResponse {
+            job_id: "job-1".into(),
+            job_name: "q1".into(),
+            job_status: "COMPLETED".into(),
+            status: "Successful".into(),
+            num_stages: 2,
+            completed_stages: 2,
+            percent_complete: 100,
+            start_time: 10,
+            end_time: 20,
+            logical_plan: Some("Projection".into()),
+            physical_plan: Some("ProjectionExec".into()),
+            stage_plan: Some("stage plan".into()),
+        };
+        let event = HistoryEvent::JobEnd {
+            version: SCHEMA_VERSION,
+            status: JobEndStatus::Succeeded,
+            queued_at: 5,
+            started_at: 10,
+            completed_at: 20,
+            job: Box::new(job),
+            stages: Box::new(QueryStagesResponse { stages: vec![] }),
+            config: BTreeMap::from([("k".to_string(), "v".to_string())]),
+            dot: "digraph {}".into(),
+        };
+        let line = serde_json::to_string(&event).unwrap();
+        assert!(line.contains("\"ev\":\"JobEnd\""));
+        let back: HistoryEvent = serde_json::from_str(&line).unwrap();
+        // Re-serialize and compare strings (stable, discriminating).
+        assert_eq!(line, serde_json::to_string(&back).unwrap());
+    }
+}

ballista/history/src/lib.rs

@@ -0,0 +1,21 @@
+// 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.
+
+pub mod dto;
+pub mod event;
+pub mod reader;
+pub mod writer;