You are viewing a plain text version of this content. The canonical link for it is here.
Posted to github@datafusion.apache.org by "andygrove (via GitHub)" <gi...@apache.org> on 2026/08/09 13:59:04 UTC
[PR] feat(scheduler): add the history server [datafusion-ballista]
andygrove opened a new pull request, #2265:
URL: https://github.com/apache/datafusion-ballista/pull/2265
# Which issue does this PR close?
Closes #1923. Last of four PRs splitting up #1925, after #2256 (merged), #2260 and #2264.
> **Stacked on #2264.** GitHub shows the whole chain until the earlier ones merge, so review the last commit here (`feat(scheduler): add the history server`).
# Rationale for this change
#2264 makes the scheduler write durable per-job logs, but nothing reads them. This serves them.
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 on restart. With this, the TUI can point at a history server and browse completed jobs with no scheduler running at all.
`user-personas.md` lists "a history server / UI" among the things Persona 2 depends on, so this fills in a guarantee already written down.
# What changes are included in this PR?
**`ballista-history-server --event-log-dir <dir>`** loads every completed log at startup and serves the same `/api/*` paths as the live scheduler: `jobs`, `job/{id}`, `job/{id}/stages`, `job/{id}/config`, `job/{id}/dot`.
Corrupt or unreadable logs are logged and skipped rather than failing startup, so one file truncated by a crash cannot hide every other job. `/api/executors` returns empty and `/api/state` a static payload, since there is no cluster behind a history server and the TUI calls both at startup.
**A user-guide page** covering both flags, the endpoints, and the operational caveats.
## How the payloads are served
The stored `/api/job/{id}` and `/api/job/{id}/stages` payloads are relayed as raw JSON rather than deserialized and re-serialized. Clients get the exact bytes the live scheduler produced, and a later change to the REST types cannot make an existing log unservable, which is the compatibility property #2260 is built around.
The job list is the one endpoint that needs structure, since it omits the plan fields. It is rebuilt from the frozen `JobIndex` rather than by editing a stored payload, so that path never parses a response it would mostly discard.
## The end-to-end guarantee
`history_store_serves_byte_identical_json_to_live_scheduler` was held out of #2264 because it needs `HistoryStore`. It lands here. It emits a real `JobEnd` through the real async writer, loads it back through `HistoryStore::load`, and compares the served bytes against what the live builders produce for the same graph.
Storing raw JSON made it a stronger test than it was in #1925: it is now a literal byte comparison rather than a structural one.
# Are there any user-facing changes?
Yes, all additive:
- New `ballista-history-server` binary.
- The TUI can point at it with `--host` / `--port` to browse completed jobs with no live scheduler.
No API changes, and nothing about a cluster that has not enabled event logging.
Verified locally: `cargo test -p ballista-scheduler --lib` passes (337 tests), clippy is clean with `--all-features -D warnings`, fmt, taplo and prettier are clean, and the `--no-default-features` check still passes.
I also ran the binary against the checked-in `schema-v1.eventlog` fixture and curled every endpoint. It loads the job, serves the list, the job, the stages (relaying the multi-partition `partition_id` array verbatim) and the config, and 404s on an unknown job.
# Known limitations
Documented on the user-guide page rather than left to be discovered:
- **Local filesystem only**, and logs are read once at startup. Restart to pick up newly finished jobs.
- **Disk is not reclaimed automatically.** Logs accumulate until pruned.
- **Plans are rendered once, at write time**, so `?plan_format=` has no effect against a history server.
- **Only the terminal record is served.** The per-task timeline is captured so a future UI can show a job progressing, but nothing reads it yet.
# Not yet verified
I have run this against hand-written and fixture logs, not against a log produced by a live cluster end to end. The emission path is covered by unit tests in #2264, but the two halves have not been exercised together on a real cluster. Worth doing before this lands.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(scheduler): add the history server [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on PR #2265:
URL: https://github.com/apache/datafusion-ballista/pull/2265#issuecomment-5249863775
I'll use this comments for follow up ideas, we can filter them later
- Https support
- authentication
- web TUI integration
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(scheduler): add the history server [datafusion-ballista]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #2265:
URL: https://github.com/apache/datafusion-ballista/pull/2265#issuecomment-5455572672
> This fall of the radar, it would be good to get it in for 55 release
>
> @andygrove do you want to merge it as it is, even with static job list, we can iterate on it. Or you prefer to address comments
I have now addressed feedback. Thanks for the reviews, I'll go ahead and merge and we can keep iterating on this.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(scheduler): add the history server [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on PR #2265:
URL: https://github.com/apache/datafusion-ballista/pull/2265#issuecomment-5245225611
simple test scenario
- i have a running server which writes events to predefined directory,
- i have a history server which reads same predefined directory
- run a job
- new event file gets written to predefined directory
number of jobs returned by history server does not change, new job not detected
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(scheduler): add the history server [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on PR #2265:
URL: https://github.com/apache/datafusion-ballista/pull/2265#issuecomment-5435228633
This fall of the radar, it would be good to get it in for 55 release
@andygrove do you want to merge it as it is, even with static job list, we can iterate on it. Or you prefer to address comments
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(scheduler): add the history server [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on PR #2265:
URL: https://github.com/apache/datafusion-ballista/pull/2265#issuecomment-5245566047
#2269 landed, it generates monotonically increasing ids, perhaps it could be used to avoid loading and caching data from a fs
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(scheduler): add the history server [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on PR #2265:
URL: https://github.com/apache/datafusion-ballista/pull/2265#issuecomment-5249720629
Not a bother at all, this is good functionality to have, happy to help
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(scheduler): add the history server [datafusion-ballista]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #2265:
URL: https://github.com/apache/datafusion-ballista/pull/2265#discussion_r3882596629
##########
docs/source/user-guide/history-server.md:
##########
@@ -0,0 +1,130 @@
+<!---
+ 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.
+-->
+
+# 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, `<job_id>.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 scans the directory at startup and indexes every completed log it finds, then
+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.
+
+The directory is scanned once at startup, so restart the history server to pick
Review Comment:
I have now added auto refresh based on an interval
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(scheduler): add the history server [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on PR #2265:
URL: https://github.com/apache/datafusion-ballista/pull/2265#issuecomment-5455918215
Thanks @andygrove
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(scheduler): add the history server [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on code in PR #2265:
URL: https://github.com/apache/datafusion-ballista/pull/2265#discussion_r3745265614
##########
ballista/scheduler/src/history/mod.rs:
##########
@@ -0,0 +1,414 @@
+// 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: loads 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::{ReplayedJob, read_completed_job};
+use datafusion::DATAFUSION_VERSION;
+use http::StatusCode;
+use http::header::CONTENT_TYPE;
+use serde_json::value::RawValue;
+use std::collections::HashMap;
+use std::path::Path;
+use std::sync::Arc;
+
+/// In-memory store of completed jobs, loaded once at startup from a directory
+/// of `<job_id>.eventlog` files.
+#[derive(Default)]
+pub struct HistoryStore {
+ /// Completed jobs keyed by job id.
+ pub jobs: HashMap<String, ReplayedJob>,
+}
+
+impl HistoryStore {
+ /// Load 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.
+ pub fn load(dir: &Path) -> std::io::Result<HistoryStore> {
Review Comment:
same like previous, why do we need to load it all instead of reading it on request ?
##########
ballista/scheduler/src/history/mod.rs:
##########
@@ -0,0 +1,414 @@
+// 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: loads 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::{ReplayedJob, read_completed_job};
+use datafusion::DATAFUSION_VERSION;
+use http::StatusCode;
+use http::header::CONTENT_TYPE;
+use serde_json::value::RawValue;
+use std::collections::HashMap;
+use std::path::Path;
+use std::sync::Arc;
+
+/// In-memory store of completed jobs, loaded once at startup from a directory
+/// of `<job_id>.eventlog` files.
+#[derive(Default)]
+pub struct HistoryStore {
+ /// Completed jobs keyed by job id.
+ pub jobs: HashMap<String, ReplayedJob>,
Review Comment:
why do we want to keep them all in the memory? cant we just read file on request
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(scheduler): add the history server [datafusion-ballista]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #2265:
URL: https://github.com/apache/datafusion-ballista/pull/2265#discussion_r3745570267
##########
ballista/scheduler/src/history/mod.rs:
##########
@@ -0,0 +1,595 @@
+// 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;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+
+/// 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 `<job_id>.eventlog` the rest of the job is read back from.
+ path: PathBuf,
+}
+
+/// 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.
+#[derive(Default)]
+pub struct HistoryStore {
+ /// Completed jobs keyed by job id.
+ jobs: HashMap<String, JobEntry>,
+}
+
+/// Why reading one job's stored payload back produced nothing.
+#[derive(Debug)]
+pub enum JobReadError {
+ /// No job with this id was found when the directory was indexed.
+ NotFound,
+ /// The log was indexed at startup 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 startup.
+ pub fn load(dir: &Path) -> std::io::Result<HistoryStore> {
+ let mut jobs = HashMap::new();
+ if dir.exists() {
+ for entry in std::fs::read_dir(dir)? {
+ let path = entry?.path();
+ if path.extension().and_then(|e| e.to_str()) != Some("eventlog") {
+ continue;
+ }
+ match read_job_index(&path) {
+ Ok(Some(index)) => {
+ jobs.insert(index.job_id.clone(), JobEntry { index, path });
+ }
+ Ok(None) => {}
+ Err(err) => {
+ tracing::warn!(
+ "skipping unreadable event log {}: {err}",
+ path.display()
+ );
+ }
+ }
+ }
+ }
+ Ok(HistoryStore { jobs })
+ }
+
+ /// How many completed jobs were indexed.
+ pub fn len(&self) -> usize {
+ self.jobs.len()
+ }
+
+ /// Whether the log directory held no completed jobs.
+ pub fn is_empty(&self) -> bool {
+ self.jobs.is_empty()
+ }
+
+ /// 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<ReplayedJob, JobReadError> {
+ let entry = self.jobs.get(job_id).ok_or(JobReadError::NotFound)?;
+ match read_completed_job(&entry.path) {
+ Ok(Some(replayed)) => Ok(replayed),
+ Ok(None) => Err(JobReadError::Vanished),
+ Err(e) => Err(JobReadError::Unreadable(e)),
+ }
+ }
+}
+
+/// [`HistoryStore::read_job`] moved off the async runtime.
+///
+/// A log that was indexed at startup 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<HistoryStore>,
+ job_id: String,
+) -> Result<ReplayedJob, SchedulerErrorResponse> {
+ 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<HistoryStore>) -> 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.
+async fn get_jobs(State(store): State<Arc<HistoryStore>>) -> Json<Vec<JobResponse>> {
Review Comment:
I'd like to do this as a separate PR since it will involve API changes. Filed https://github.com/apache/datafusion-ballista/issues/2270 to track
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(scheduler): add the history server [datafusion-ballista]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #2265:
URL: https://github.com/apache/datafusion-ballista/pull/2265#discussion_r3745548484
##########
ballista/scheduler/src/history/mod.rs:
##########
@@ -0,0 +1,414 @@
+// 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: loads 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::{ReplayedJob, read_completed_job};
+use datafusion::DATAFUSION_VERSION;
+use http::StatusCode;
+use http::header::CONTENT_TYPE;
+use serde_json::value::RawValue;
+use std::collections::HashMap;
+use std::path::Path;
+use std::sync::Arc;
+
+/// In-memory store of completed jobs, loaded once at startup from a directory
+/// of `<job_id>.eventlog` files.
+#[derive(Default)]
+pub struct HistoryStore {
+ /// Completed jobs keyed by job id.
+ pub jobs: HashMap<String, ReplayedJob>,
Review Comment:
Thanks, this wasn't intended ... fixed now
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(scheduler): add the history server [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on code in PR #2265:
URL: https://github.com/apache/datafusion-ballista/pull/2265#discussion_r3749607555
##########
ballista/scheduler/src/history/mod.rs:
##########
@@ -0,0 +1,659 @@
+// 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;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+
+/// 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 `<job_id>.eventlog` the rest of the job is read back from.
+ path: PathBuf,
+}
+
+/// 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.
+#[derive(Default)]
+pub struct HistoryStore {
+ /// Completed jobs keyed by job id.
+ jobs: HashMap<String, JobEntry>,
+}
+
+/// Why reading one job's stored payload back produced nothing.
+#[derive(Debug)]
+pub enum JobReadError {
+ /// No job with this id was found when the directory was indexed.
+ NotFound,
+ /// The log was indexed at startup 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 startup.
+ pub fn load(dir: &Path) -> std::io::Result<HistoryStore> {
Review Comment:
will this be called ONLY when history server starts?
can we handle situation where we have one history server running and we have one or more scheduler servers producing events, will new events be visible to history server (after history server startup)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(scheduler): add the history server [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on code in PR #2265:
URL: https://github.com/apache/datafusion-ballista/pull/2265#discussion_r3791290877
##########
docs/source/user-guide/history-server.md:
##########
@@ -0,0 +1,130 @@
+<!---
+ 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.
+-->
+
+# 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, `<job_id>.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 scans the directory at startup and indexes every completed log it finds, then
+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.
+
+The directory is scanned once at startup, so restart the history server to pick
Review Comment:
Do you want to history sever to serve only static list of jobs discovered at startup @andygrove ?
This was my main comment last time, I think it would make sense to discover new events automatically during execution rather than on startup.
Anyway your call, please let me know what you think
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(scheduler): add the history server [datafusion-ballista]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on code in PR #2265:
URL: https://github.com/apache/datafusion-ballista/pull/2265#discussion_r3745568750
##########
ballista/scheduler/src/history/mod.rs:
##########
@@ -0,0 +1,595 @@
+// 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;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+
+/// 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 `<job_id>.eventlog` the rest of the job is read back from.
+ path: PathBuf,
+}
+
+/// 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.
+#[derive(Default)]
+pub struct HistoryStore {
+ /// Completed jobs keyed by job id.
+ jobs: HashMap<String, JobEntry>,
+}
+
+/// Why reading one job's stored payload back produced nothing.
+#[derive(Debug)]
+pub enum JobReadError {
+ /// No job with this id was found when the directory was indexed.
+ NotFound,
+ /// The log was indexed at startup 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 startup.
+ pub fn load(dir: &Path) -> std::io::Result<HistoryStore> {
+ let mut jobs = HashMap::new();
+ if dir.exists() {
+ for entry in std::fs::read_dir(dir)? {
+ let path = entry?.path();
+ if path.extension().and_then(|e| e.to_str()) != Some("eventlog") {
+ continue;
+ }
+ match read_job_index(&path) {
+ Ok(Some(index)) => {
+ jobs.insert(index.job_id.clone(), JobEntry { index, path });
+ }
+ Ok(None) => {}
+ Err(err) => {
+ tracing::warn!(
+ "skipping unreadable event log {}: {err}",
+ path.display()
+ );
+ }
+ }
+ }
+ }
+ Ok(HistoryStore { jobs })
+ }
+
+ /// How many completed jobs were indexed.
+ pub fn len(&self) -> usize {
+ self.jobs.len()
+ }
+
+ /// Whether the log directory held no completed jobs.
+ pub fn is_empty(&self) -> bool {
+ self.jobs.is_empty()
+ }
+
+ /// 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<ReplayedJob, JobReadError> {
+ let entry = self.jobs.get(job_id).ok_or(JobReadError::NotFound)?;
+ match read_completed_job(&entry.path) {
+ Ok(Some(replayed)) => Ok(replayed),
+ Ok(None) => Err(JobReadError::Vanished),
+ Err(e) => Err(JobReadError::Unreadable(e)),
+ }
+ }
+}
+
+/// [`HistoryStore::read_job`] moved off the async runtime.
+///
+/// A log that was indexed at startup 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<HistoryStore>,
+ job_id: String,
+) -> Result<ReplayedJob, SchedulerErrorResponse> {
+ 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<HistoryStore>) -> 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.
+async fn get_jobs(State(store): State<Arc<HistoryStore>>) -> Json<Vec<JobResponse>> {
+ let mut jobs: Vec<JobResponse> =
+ store.jobs.values().map(|j| list_entry(&j.index)).collect();
+ jobs.sort_by(|a, b| a.job_id.cmp(&b.job_id));
Review Comment:
Good catch. Fixed.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(scheduler): add the history server [datafusion-ballista]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove merged PR #2265:
URL: https://github.com/apache/datafusion-ballista/pull/2265
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(scheduler): add the history server [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on code in PR #2265:
URL: https://github.com/apache/datafusion-ballista/pull/2265#discussion_r3791290877
##########
docs/source/user-guide/history-server.md:
##########
@@ -0,0 +1,130 @@
+<!---
+ 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.
+-->
+
+# 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, `<job_id>.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 scans the directory at startup and indexes every completed log it finds, then
+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.
+
+The directory is scanned once at startup, so restart the history server to pick
Review Comment:
Do you want history sever to serve only static list of jobs discovered at startup @andygrove ?
This was my main comment last time, I think it would make sense to discover new events automatically during execution rather than on startup.
Anyway your call, please let me know what you think
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(scheduler): add the history server [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on code in PR #2265:
URL: https://github.com/apache/datafusion-ballista/pull/2265#discussion_r3749562779
##########
ballista/scheduler/src/history/mod.rs:
##########
@@ -0,0 +1,595 @@
+// 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;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+
+/// 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 `<job_id>.eventlog` the rest of the job is read back from.
+ path: PathBuf,
+}
+
+/// 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.
+#[derive(Default)]
+pub struct HistoryStore {
+ /// Completed jobs keyed by job id.
+ jobs: HashMap<String, JobEntry>,
+}
+
+/// Why reading one job's stored payload back produced nothing.
+#[derive(Debug)]
+pub enum JobReadError {
+ /// No job with this id was found when the directory was indexed.
+ NotFound,
+ /// The log was indexed at startup 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 startup.
+ pub fn load(dir: &Path) -> std::io::Result<HistoryStore> {
+ let mut jobs = HashMap::new();
+ if dir.exists() {
+ for entry in std::fs::read_dir(dir)? {
+ let path = entry?.path();
+ if path.extension().and_then(|e| e.to_str()) != Some("eventlog") {
+ continue;
+ }
+ match read_job_index(&path) {
+ Ok(Some(index)) => {
+ jobs.insert(index.job_id.clone(), JobEntry { index, path });
+ }
+ Ok(None) => {}
+ Err(err) => {
+ tracing::warn!(
+ "skipping unreadable event log {}: {err}",
+ path.display()
+ );
+ }
+ }
+ }
+ }
+ Ok(HistoryStore { jobs })
+ }
+
+ /// How many completed jobs were indexed.
+ pub fn len(&self) -> usize {
+ self.jobs.len()
+ }
+
+ /// Whether the log directory held no completed jobs.
+ pub fn is_empty(&self) -> bool {
+ self.jobs.is_empty()
+ }
+
+ /// 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<ReplayedJob, JobReadError> {
+ let entry = self.jobs.get(job_id).ok_or(JobReadError::NotFound)?;
+ match read_completed_job(&entry.path) {
+ Ok(Some(replayed)) => Ok(replayed),
+ Ok(None) => Err(JobReadError::Vanished),
+ Err(e) => Err(JobReadError::Unreadable(e)),
+ }
+ }
+}
+
+/// [`HistoryStore::read_job`] moved off the async runtime.
+///
+/// A log that was indexed at startup 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<HistoryStore>,
+ job_id: String,
+) -> Result<ReplayedJob, SchedulerErrorResponse> {
+ 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<HistoryStore>) -> 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.
+async fn get_jobs(State(store): State<Arc<HistoryStore>>) -> Json<Vec<JobResponse>> {
Review Comment:
Makes sense
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(scheduler): add the history server [datafusion-ballista]
Posted by "andygrove (via GitHub)" <gi...@apache.org>.
andygrove commented on PR #2265:
URL: https://github.com/apache/datafusion-ballista/pull/2265#issuecomment-5248048188
Sorry for the quality of this PR @milenkovicm and thanks for reviews so far. Moving to draft until this is more complete.
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org
Re: [PR] feat(scheduler): add the history server [datafusion-ballista]
Posted by "milenkovicm (via GitHub)" <gi...@apache.org>.
milenkovicm commented on code in PR #2265:
URL: https://github.com/apache/datafusion-ballista/pull/2265#discussion_r3745438056
##########
ballista/scheduler/src/history/mod.rs:
##########
@@ -0,0 +1,595 @@
+// 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;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+
+/// 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 `<job_id>.eventlog` the rest of the job is read back from.
+ path: PathBuf,
+}
+
+/// 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.
+#[derive(Default)]
+pub struct HistoryStore {
+ /// Completed jobs keyed by job id.
+ jobs: HashMap<String, JobEntry>,
+}
+
+/// Why reading one job's stored payload back produced nothing.
+#[derive(Debug)]
+pub enum JobReadError {
+ /// No job with this id was found when the directory was indexed.
+ NotFound,
+ /// The log was indexed at startup 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 startup.
+ pub fn load(dir: &Path) -> std::io::Result<HistoryStore> {
+ let mut jobs = HashMap::new();
+ if dir.exists() {
+ for entry in std::fs::read_dir(dir)? {
+ let path = entry?.path();
+ if path.extension().and_then(|e| e.to_str()) != Some("eventlog") {
+ continue;
+ }
+ match read_job_index(&path) {
+ Ok(Some(index)) => {
+ jobs.insert(index.job_id.clone(), JobEntry { index, path });
+ }
+ Ok(None) => {}
+ Err(err) => {
+ tracing::warn!(
+ "skipping unreadable event log {}: {err}",
+ path.display()
+ );
+ }
+ }
+ }
+ }
+ Ok(HistoryStore { jobs })
+ }
+
+ /// How many completed jobs were indexed.
+ pub fn len(&self) -> usize {
+ self.jobs.len()
+ }
+
+ /// Whether the log directory held no completed jobs.
+ pub fn is_empty(&self) -> bool {
+ self.jobs.is_empty()
+ }
+
+ /// 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<ReplayedJob, JobReadError> {
+ let entry = self.jobs.get(job_id).ok_or(JobReadError::NotFound)?;
+ match read_completed_job(&entry.path) {
+ Ok(Some(replayed)) => Ok(replayed),
+ Ok(None) => Err(JobReadError::Vanished),
+ Err(e) => Err(JobReadError::Unreadable(e)),
+ }
+ }
+}
+
+/// [`HistoryStore::read_job`] moved off the async runtime.
+///
+/// A log that was indexed at startup 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<HistoryStore>,
+ job_id: String,
+) -> Result<ReplayedJob, SchedulerErrorResponse> {
+ 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<HistoryStore>) -> 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.
+async fn get_jobs(State(store): State<Arc<HistoryStore>>) -> Json<Vec<JobResponse>> {
Review Comment:
We would need to add paging if we plan to keep more than few jobs in history store
Making jobs id sorted would help i believe
##########
ballista/scheduler/src/history/mod.rs:
##########
@@ -0,0 +1,595 @@
+// 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;
+use std::path::{Path, PathBuf};
+use std::sync::Arc;
+
+/// 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 `<job_id>.eventlog` the rest of the job is read back from.
+ path: PathBuf,
+}
+
+/// 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.
+#[derive(Default)]
+pub struct HistoryStore {
+ /// Completed jobs keyed by job id.
+ jobs: HashMap<String, JobEntry>,
+}
+
+/// Why reading one job's stored payload back produced nothing.
+#[derive(Debug)]
+pub enum JobReadError {
+ /// No job with this id was found when the directory was indexed.
+ NotFound,
+ /// The log was indexed at startup 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 startup.
+ pub fn load(dir: &Path) -> std::io::Result<HistoryStore> {
+ let mut jobs = HashMap::new();
+ if dir.exists() {
+ for entry in std::fs::read_dir(dir)? {
+ let path = entry?.path();
+ if path.extension().and_then(|e| e.to_str()) != Some("eventlog") {
+ continue;
+ }
+ match read_job_index(&path) {
+ Ok(Some(index)) => {
+ jobs.insert(index.job_id.clone(), JobEntry { index, path });
+ }
+ Ok(None) => {}
+ Err(err) => {
+ tracing::warn!(
+ "skipping unreadable event log {}: {err}",
+ path.display()
+ );
+ }
+ }
+ }
+ }
+ Ok(HistoryStore { jobs })
+ }
+
+ /// How many completed jobs were indexed.
+ pub fn len(&self) -> usize {
+ self.jobs.len()
+ }
+
+ /// Whether the log directory held no completed jobs.
+ pub fn is_empty(&self) -> bool {
+ self.jobs.is_empty()
+ }
+
+ /// 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<ReplayedJob, JobReadError> {
+ let entry = self.jobs.get(job_id).ok_or(JobReadError::NotFound)?;
+ match read_completed_job(&entry.path) {
+ Ok(Some(replayed)) => Ok(replayed),
+ Ok(None) => Err(JobReadError::Vanished),
+ Err(e) => Err(JobReadError::Unreadable(e)),
+ }
+ }
+}
+
+/// [`HistoryStore::read_job`] moved off the async runtime.
+///
+/// A log that was indexed at startup 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<HistoryStore>,
+ job_id: String,
+) -> Result<ReplayedJob, SchedulerErrorResponse> {
+ 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<HistoryStore>) -> 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.
+async fn get_jobs(State(store): State<Arc<HistoryStore>>) -> Json<Vec<JobResponse>> {
+ let mut jobs: Vec<JobResponse> =
+ store.jobs.values().map(|j| list_entry(&j.index)).collect();
+ jobs.sort_by(|a, b| a.job_id.cmp(&b.job_id));
Review Comment:
Should jobs be sorted on time rather than name (if job ids are not monotonically increasing)
--
This is an automated message from the Apache Git Service.
To respond to the message, please log on to GitHub and use the
URL above to go to the specific comment.
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For queries about this service, please contact Infrastructure at:
users@infra.apache.org
---------------------------------------------------------------------
To unsubscribe, e-mail: github-unsubscribe@datafusion.apache.org
For additional commands, e-mail: github-help@datafusion.apache.org