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: [email protected] For queries about this service, please contact Infrastructure at: [email protected] --------------------------------------------------------------------- To unsubscribe, e-mail: [email protected] For additional commands, e-mail: [email protected]
