andygrove commented on code in PR #1925:
URL:
https://github.com/apache/datafusion-ballista/pull/1925#discussion_r3721032355
##########
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:
Good catch, and it was worse than just a missing record. Cancellation is
terminal for a job (the handler below removes the graph), so a cancelled job
never got a `JobEnd`, which meant it was invisible to the history server and
its file handle stayed open for the life of the process. Added a `JobCancel`
arm that appends a terminal event and calls `finish_job`.
One wrinkle: the tee runs before the cancel is applied to the graph, so the
graph still reports the job as running at that point. The new
`job_cancel_event` stamps the cancelled status onto the DTO so the history view
does not show a cancelled job as perpetually `Running`. Added
`JobEndStatus::Cancelled` plus a test for it.
`JobPlanningFailed` I left alone on purpose, and added a comment saying why:
it is only posted when `submit_job` fails, so it happens instead of
`JobSubmitted`. The job has neither an execution graph nor an open log file at
that point, so there is nothing to write or close.
##########
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:
Applied. It does not actually panic today since this is float division and
`NaN as u8` saturates to 0 in Rust, but relying on that is not great, and
`build_job_response_from_overview` right below already guards the same way. Now
they match.
##########
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:
Agreed on the problem, though I went a slightly different way: the load now
happens in `main()` before the tokio runtime is built at all, rather than in
`spawn_blocking`. Nothing else is running at startup, so there is no reason for
a runtime to exist yet, and it avoids having to map a `JoinError` into
`BallistaError`.
##########
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:
The inconsistency was real. `TaskEnd` came straight off the protobuf
`TaskStatus` (u32) while the stage events came off the graph (usize), and I
never reconciled them.
Resolved it toward fixed width: `StageStart` and `StageEnd` are now `u32`
too, with a doc comment on the enum stating the rule. This is a durable,
cross-machine format, so a pointer-width type does not really belong in it.
Callers holding the graph`'`s `usize` stage ids cast on the way in.
--
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]
