martin-g commented on code in PR #1925:
URL:
https://github.com/apache/datafusion-ballista/pull/1925#discussion_r3558959953
##########
ballista/scheduler/src/scheduler_server/query_stage_scheduler.rs:
##########
@@ -82,6 +110,88 @@ impl<T: 'static + AsLogicalPlan, U: 'static +
AsExecutionPlan>
tx_event: &mpsc::Sender<QueryStageSchedulerEvent>,
_rx_event: &mpsc::Receiver<QueryStageSchedulerEvent>,
) -> Result<()> {
+ #[cfg(feature = "rest-api")]
+ if let Some(log) = &self.event_log {
+ match &event {
+ QueryStageSchedulerEvent::JobSubmitted {
+ job_id,
+ queued_at,
+ submitted_at,
+ } => {
+ if let Ok(Some(graph)) = self
+ .state
+ .task_manager
+ .get_job_execution_graph(job_id)
+ .await
+ {
+ log.append(
+ job_id.as_str(),
+ event_log::job_start_event(&graph, *queued_at,
*submitted_at),
+ );
+ }
+ }
+ QueryStageSchedulerEvent::TaskUpdating(executor_id, statuses)
=> {
+ for (job_id, group) in group_by_job(statuses) {
+ for ev in event_log::task_end_events(executor_id,
&group) {
+ log.append(&job_id, ev);
+ }
+ }
+ }
+ QueryStageSchedulerEvent::JobFinished {
+ job_id,
+ queued_at,
+ completed_at,
+ } => {
+ if let Ok(Some(graph)) = self
+ .state
+ .task_manager
+ .get_job_execution_graph(job_id)
+ .await
+ {
+ log.append_final(
+ job_id.as_str(),
+ event_log::job_end_event(
+ &graph,
+
ballista_history::event::JobEndStatus::Succeeded,
+ *queued_at,
+ *completed_at,
+ ),
+ )
+ .await;
+ }
+ log.finish_job(job_id.as_str()).await;
+ }
+ QueryStageSchedulerEvent::JobRunningFailed {
+ job_id,
+ fail_message,
+ queued_at,
+ failed_at,
+ } => {
+ if let Ok(Some(graph)) = self
+ .state
+ .task_manager
+ .get_job_execution_graph(job_id)
+ .await
+ {
+ log.append_final(
+ job_id.as_str(),
+ event_log::job_end_event(
+ &graph,
+ ballista_history::event::JobEndStatus::Failed(
+ fail_message.clone(),
+ ),
+ *queued_at,
+ *failed_at,
+ ),
+ )
+ .await;
+ }
+ log.finish_job(job_id.as_str()).await;
+ }
+ _ => {}
Review Comment:
Shouldn't `JobCancel` and `JobPlanningFailed` be handled too ? IMO they
should call `log.finish_job(job_id.as_str()).await;` too
##########
ballista/scheduler/src/api/dto_build.rs:
##########
@@ -0,0 +1,654 @@
+// Licensed 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.
+
+//! Builds the shared REST DTOs (`ballista_history::dto`) from the scheduler's
+//! internal execution-graph state. These builders back both the live REST API
+//! handlers (`api::handlers`) and, eventually, the event-log writer, so both
+//! serialize byte-identical JSON for the same graph state.
+
+use crate::api::handlers::{JobQueryParams, PlanFormat};
+use crate::display::format_stage_metrics;
+use crate::state::execution_graph::{ExecutionGraphBox, ExecutionStage};
+use crate::state::execution_graph_dot::ExecutionGraphDot;
+use crate::state::execution_stage::TaskInfo;
+use crate::state::task_manager::JobOverview;
+use ballista_core::serde::protobuf::job_status::Status;
+use ballista_core::serde::protobuf::{OperatorMetricsSet, task_status};
+use ballista_core::utils::get_current_time;
+use ballista_history::dto::{
+ JobResponse, Percentiles, QueryStageSummary, QueryStagesResponse,
TaskStatus,
+ TaskSummary,
+};
+use datafusion::physical_plan::display::DisplayableExecutionPlan;
+use datafusion::physical_plan::displayable;
+use datafusion::physical_plan::metrics::{MetricsSet, Time};
+use std::fmt;
+use std::time::Duration;
+
+/// Builds the `JobResponse` DTO for a job's execution graph.
+///
+/// `with_plans` controls whether `logical_plan`/`physical_plan`/`stage_plan`
are
+/// populated (the single-job detail view, and eventually the history detail
+/// view) or left `None` (list views, which use
+/// [`build_job_response_from_overview`] instead). `plan_format` selects how
the
+/// physical plan is rendered when `with_plans` is true; it is ignored
otherwise.
+pub(crate) fn build_job_response(
+ graph: &ExecutionGraphBox,
+ with_plans: bool,
+ plan_format: PlanFormat,
+) -> JobResponse {
+ let job = graph.as_ref();
+ let (plain_status, job_status) = format_job_status(
+ &job.status().status,
+ job_elapsed_ms(job.start_time(), job.end_time()),
+ );
+
+ let num_stages = job.stage_count();
+ let completed_stages = job.completed_stages();
+ let percent_complete =
+ ((completed_stages as f32 / num_stages as f32) * 100_f32) as u8;
Review Comment:
`num_stages` could be 0
```suggestion
let percent_complete = if num_stages == 0 {
0
} else {
((completed_stages as f32 / num_stages as f32) * 100_f32) as u8
};
```
##########
ballista/history/src/event.rs:
##########
@@ -0,0 +1,124 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! The on-disk event-log schema. One `HistoryEvent` is serialized per JSONL
line.
+//! This is a frozen public projection of the scheduler's internal events; the
+//! embedded DTOs are the stable contract the history server serves.
+
+use crate::dto::{JobConfig, JobResponse, QueryStagesResponse, TaskStatus};
+use serde::{Deserialize, Serialize};
+
+/// Current on-disk schema version, stamped on `JobStart`/`JobEnd`.
+pub const SCHEMA_VERSION: u32 = 1;
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub enum JobEndStatus {
+ Succeeded,
+ Failed(String),
+}
+
+/// Metrics captured per finished task on the incremental timeline.
+#[derive(Debug, Clone, Serialize, Deserialize)]
+pub struct TaskEndMetrics {
+ pub input_rows: u64,
+ pub output_rows: u64,
+ pub elapsed_compute_nanos: u64,
+}
+
+#[derive(Debug, Clone, Serialize, Deserialize)]
+#[serde(tag = "ev")]
+pub enum HistoryEvent {
+ JobStart {
+ version: u32,
+ job_id: String,
+ job_name: String,
+ queued_at: u64,
+ submitted_at: u64,
+ logical_plan: Option<String>,
+ physical_plan: Option<String>,
+ },
+ StageStart {
+ stage_id: usize,
+ partitions: usize,
+ },
+ StageEnd {
+ stage_id: usize,
+ status: String,
+ },
+ TaskEnd {
+ stage_id: u32,
Review Comment:
Why `u32` ?
Above it uses `usize` (lines 55 and 59)
##########
ballista/scheduler/src/scheduler_server/query_stage_scheduler.rs:
##########
@@ -82,6 +110,88 @@ impl<T: 'static + AsLogicalPlan, U: 'static +
AsExecutionPlan>
tx_event: &mpsc::Sender<QueryStageSchedulerEvent>,
_rx_event: &mpsc::Receiver<QueryStageSchedulerEvent>,
) -> Result<()> {
+ #[cfg(feature = "rest-api")]
+ if let Some(log) = &self.event_log {
+ match &event {
+ QueryStageSchedulerEvent::JobSubmitted {
+ job_id,
+ queued_at,
+ submitted_at,
+ } => {
+ if let Ok(Some(graph)) = self
+ .state
+ .task_manager
+ .get_job_execution_graph(job_id)
+ .await
+ {
+ log.append(
+ job_id.as_str(),
+ event_log::job_start_event(&graph, *queued_at,
*submitted_at),
+ );
+ }
+ }
+ QueryStageSchedulerEvent::TaskUpdating(executor_id, statuses)
=> {
+ for (job_id, group) in group_by_job(statuses) {
+ for ev in event_log::task_end_events(executor_id,
&group) {
+ log.append(&job_id, ev);
+ }
+ }
+ }
+ QueryStageSchedulerEvent::JobFinished {
+ job_id,
+ queued_at,
+ completed_at,
+ } => {
+ if let Ok(Some(graph)) = self
+ .state
+ .task_manager
+ .get_job_execution_graph(job_id)
Review Comment:
What if this returns Err ?
`job_end_event()` won't be called but
`log.finish_job(job_id.as_str()).await;` is still executed.
##########
ballista/history/src/writer.rs:
##########
@@ -0,0 +1,303 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Async, buffered event-log writer. Each job's events append to
+//! `<log_dir>/<job_id>.eventlog` as JSONL. Appends are non-blocking; a
background
+//! task performs the file I/O so the scheduler hot path never waits on disk.
+
+use crate::event::HistoryEvent;
+use std::collections::HashMap;
+use std::path::{Path, PathBuf};
+use tokio::io::AsyncWriteExt;
+use tokio::sync::{mpsc, oneshot};
+
+enum WriterMsg {
+ Event {
+ job_id: String,
+ event: Box<HistoryEvent>,
+ },
+ Flush {
+ job_id: String,
+ done: oneshot::Sender<()>,
+ },
+ Finish {
+ job_id: String,
+ done: oneshot::Sender<()>,
+ },
+}
+
+#[derive(Clone)]
+pub struct EventLogWriter {
+ tx: mpsc::Sender<WriterMsg>,
+}
+
+impl EventLogWriter {
+ pub fn new(log_dir: PathBuf, buffer: usize) -> EventLogWriter {
+ let (tx, rx) = mpsc::channel(buffer.max(1));
+ tokio::spawn(run(log_dir, rx));
+ EventLogWriter { tx }
+ }
+
+ /// Enqueue an event for `job_id`. Never blocks; drops (with a warning) if
the
+ /// channel is full, so logging cannot stall scheduling.
+ pub fn append(&self, job_id: &str, event: HistoryEvent) {
+ let msg = WriterMsg::Event {
+ job_id: job_id.to_string(),
+ event: Box::new(event),
+ };
+ if self.tx.try_send(msg).is_err() {
+ eprintln!(
+ "event-log writer: dropping event for {job_id} (channel full
or closed)"
+ );
+ }
+ }
+
+ /// Await all currently-enqueued writes for `job_id` (best effort).
+ pub async fn flush_job(&self, job_id: &str) {
+ let (done, wait) = oneshot::channel();
+ if self
+ .tx
+ .send(WriterMsg::Flush {
+ job_id: job_id.to_string(),
+ done,
+ })
+ .await
+ .is_ok()
+ {
+ let _ = wait.await;
+ }
+ }
+
+ /// Enqueue a terminal event (e.g. `JobEnd`) for `job_id`. Unlike
`append`, this
+ /// awaits channel capacity instead of dropping the event when the channel
is
+ /// full, so the terminal record is never silently lost. Still best-effort
at
+ /// the process boundary: if the channel is closed (background task gone)
this
+ /// logs and returns rather than panicking.
+ pub async fn append_final(&self, job_id: &str, event: HistoryEvent) {
+ let msg = WriterMsg::Event {
+ job_id: job_id.to_string(),
+ event: Box::new(event),
+ };
+ if self.tx.send(msg).await.is_err() {
+ eprintln!(
+ "event-log writer: failed to enqueue terminal event for
{job_id} (channel closed)"
+ );
+ }
+ }
+
+ /// Flush and close the per-job file handle for `job_id`. Must be called
after
+ /// the terminal event has been enqueued (e.g. via `append_final`) so it is
+ /// ordered after it on the single-consumer FIFO channel. Best effort: if
the
+ /// channel is closed this logs and returns.
+ pub async fn finish_job(&self, job_id: &str) {
+ let (done, wait) = oneshot::channel();
+ if self
+ .tx
+ .send(WriterMsg::Finish {
+ job_id: job_id.to_string(),
+ done,
+ })
+ .await
+ .is_ok()
+ {
+ let _ = wait.await;
+ } else {
+ eprintln!(
+ "event-log writer: failed to enqueue finish for {job_id}
(channel closed)"
+ );
+ }
+ }
+}
+
+async fn run(log_dir: PathBuf, mut rx: mpsc::Receiver<WriterMsg>) {
+ if let Err(e) = tokio::fs::create_dir_all(&log_dir).await {
+ eprintln!("event-log writer: cannot create {}: {e}",
log_dir.display());
+ return;
+ }
+ // One open append handle per job for the life of the process.
+ let mut handles: HashMap<String, tokio::fs::File> = HashMap::new();
+
+ while let Some(msg) = rx.recv().await {
+ match msg {
+ WriterMsg::Event { job_id, event } => {
+ let file = match open_for(&log_dir, &mut handles,
&job_id).await {
+ Some(f) => f,
+ None => continue,
+ };
+ match serde_json::to_string(&*event) {
+ Ok(mut line) => {
+ line.push('\n');
+ if let Err(e) = file.write_all(line.as_bytes()).await {
+ eprintln!("event-log writer: write failed for
{job_id}: {e}");
+ }
+ }
+ Err(e) => eprintln!("event-log writer: serialize failed:
{e}"),
+ }
+ }
+ WriterMsg::Flush { job_id, done } => {
+ if let Some(file) = handles.get_mut(&job_id) {
+ let _ = file.flush().await;
+ }
+ let _ = done.send(());
+ }
+ WriterMsg::Finish { job_id, done } => {
+ if let Some(mut file) = handles.remove(&job_id) {
+ let _ = file.flush().await;
+ // Dropping `file` here closes the fd.
+ }
+ let _ = done.send(());
+ }
+ }
+ }
+}
+
+async fn open_for<'a>(
+ log_dir: &Path,
+ handles: &'a mut HashMap<String, tokio::fs::File>,
+ job_id: &str,
+) -> Option<&'a mut tokio::fs::File> {
+ if !handles.contains_key(job_id) {
+ let path = log_dir.join(format!("{job_id}.eventlog"));
+ match tokio::fs::OpenOptions::new()
+ .create(true)
+ .append(true)
+ .open(&path)
+ .await
+ {
+ Ok(f) => {
+ handles.insert(job_id.to_string(), f);
+ }
+ Err(e) => {
+ eprintln!("event-log writer: cannot open {}: {e}",
path.display());
+ return None;
+ }
+ }
+ }
+ handles.get_mut(job_id)
Review Comment:
Using
https://doc.rust-lang.org/stable/std/collections/struct.HashMap.html#method.entry
will be more idiomatic:
```suggestion
match handles.entry(job_id.to_string()) {
std::collections::hash_map::Entry::Occupied(entry) =>
Some(entry.into_mut()),
std::collections::hash_map::Entry::Vacant(entry) => {
let path = log_dir.join(format!("{job_id}.eventlog"));
match tokio::fs::OpenOptions::new()
.create(true)
.append(true)
.open(&path)
.await
{
Ok(f) => Some(entry.insert(f)),
Err(e) => {
eprintln!("event-log writer: cannot open {}: {e}",
path.display());
None
}
}
}
}
```
##########
ballista/scheduler/src/scheduler_server/event_log.rs:
##########
@@ -0,0 +1,335 @@
+// 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.
+
+//! Builds `HistoryEvent`s from scheduler event-loop state and appends them to
+//! the [`ballista_history::writer::EventLogWriter`]. These are the pure,
+//! synchronous event-builder functions; the actual emission -- deciding
+//! *when* to call them -- lives at the top of
+//! `query_stage_scheduler::QueryStageScheduler::on_receive`.
+//!
+//! The builders reuse `crate::api::dto_build`, the same DTO builders backing
+//! the live REST API, so a job's `JobEnd` event and its `GET /api/job/{id}`
+//! response serialize identically for the same graph state.
+
+use crate::api::dto_build::{
+ build_job_dot, build_job_response, build_query_stages_response,
task_row_counts,
+ to_api_task_status,
+};
+use crate::api::handlers::{JobQueryParams, PlanFormat};
+use crate::state::execution_graph::ExecutionGraphBox;
+use ballista_core::serde::protobuf::{TaskStatus, task_status};
+use ballista_history::dto::TaskStatus as ApiTaskStatus;
+use ballista_history::event::{
+ HistoryEvent, JobEndStatus, SCHEMA_VERSION, TaskEndMetrics,
+};
+use datafusion::physical_plan::display::DisplayableExecutionPlan;
+
+/// Builds the `JobStart` event for a job that has just been submitted.
+pub(crate) fn job_start_event(
+ graph: &ExecutionGraphBox,
+ queued_at: u64,
+ submitted_at: u64,
+) -> HistoryEvent {
+ HistoryEvent::JobStart {
+ version: SCHEMA_VERSION,
+ job_id: graph.job_id().to_string(),
+ job_name: graph.job_name().to_string(),
+ queued_at,
+ submitted_at,
+ logical_plan: graph.logical_plan().map(|p| p.to_string()),
+ // Same rendering the `get_job` handler uses for `PlanFormat::Default`.
+ physical_plan: Some(
+ DisplayableExecutionPlan::new(graph.physical_plan().as_ref())
+ .indent(false)
+ .to_string(),
+ ),
+ }
+}
+
+/// Builds one `TaskEnd` event per finished task in `statuses` (skips tasks
+/// still `Running`, which are transient/in-flight status updates rather than
+/// terminal states).
+pub(crate) fn task_end_events(
+ executor_id: &str,
+ statuses: &[TaskStatus],
+) -> Vec<HistoryEvent> {
+ statuses
+ .iter()
+ .filter(|s| !matches!(s.status, Some(task_status::Status::Running(_))))
+ .map(|s| {
+ let status = s
+ .status
+ .as_ref()
+ .map(to_api_task_status)
+ .unwrap_or(ApiTaskStatus::Running);
Review Comment:
At line 72 you filter out the Running tasks but here if the status is None
you map it to `Running`. The method doc also says that `Running` should be
skipped
##########
ballista/scheduler/src/bin/history_server.rs:
##########
@@ -0,0 +1,94 @@
+// Licensed to the Apache Software Foundation (ASF) under one
+// or more contributor license agreements. See the NOTICE file
+// distributed with this work for additional information
+// regarding copyright ownership. The ASF licenses this file
+// to you under the Apache License, Version 2.0 (the
+// "License"); you may not use this file except in compliance
+// with the License. You may obtain a copy of the License at
+//
+// http://www.apache.org/licenses/LICENSE-2.0
+//
+// Unless required by applicable law or agreed to in writing,
+// software distributed under the License is distributed on an
+// "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY
+// KIND, either express or implied. See the License for the
+// specific language governing permissions and limitations
+// under the License.
+
+//! Standalone Ballista history server binary: loads completed event logs from
+//! a directory and serves the same `/api/*` responses the live scheduler does,
+//! so the existing TUI can connect to it unchanged.
+
+use ballista_core::error::{BallistaError, Result};
+use ballista_scheduler::history::{HistoryStore, history_router};
+use clap::Parser;
+use std::env;
+use std::net::SocketAddr;
+use std::path::PathBuf;
+use std::sync::Arc;
+use tracing_subscriber::EnvFilter;
+
+#[derive(Debug, clap::Parser)]
+#[command(
+ name = "ballista-history-server",
+ version,
+ about = "Ballista history server"
+)]
+struct Args {
+ /// Directory containing per-job event logs.
+ #[arg(long)]
+ event_log_dir: PathBuf,
+ /// Host to bind the HTTP server to.
+ #[arg(long, default_value = "0.0.0.0")]
+ bind_host: String,
+ /// Port to bind the HTTP server to.
+ #[arg(long, default_value_t = 50060)]
+ bind_port: u16,
+}
+
+fn main() -> Result<()> {
+ let runtime = tokio::runtime::Builder::new_multi_thread()
+ .enable_io()
+ .enable_time()
+ .build()
+ .map_err(BallistaError::IoError)?;
+
+ runtime.block_on(inner())
+}
+
+async fn inner() -> Result<()> {
+ let rust_log = env::var(EnvFilter::DEFAULT_ENV);
+ let log_filter = EnvFilter::new(rust_log.unwrap_or_else(|_|
"info".to_string()));
+ tracing_subscriber::fmt()
+ .with_ansi(false)
+ .with_writer(std::io::stdout)
+ .with_env_filter(log_filter)
+ .init();
+
+ let args = Args::parse();
+
+ let store = Arc::new(HistoryStore::load(&args.event_log_dir)?);
Review Comment:
execute the synchronous `HistoryStore::load()` in `spawn_blocking()`:
```suggestion
let event_log_dir = args.event_log_dir.clone();
let store = Arc::new(
tokio::task::spawn_blocking(move ||
HistoryStore::load(&event_log_dir))
.await
.map_err(|e| std::io::Error::new(std::io::ErrorKind::Other,
e))??,
);
```
or make `load()` async and use `tokio::fs::**` APIs inside it.
--
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]
