andygrove commented on code in PR #1925: URL: https://github.com/apache/datafusion-ballista/pull/1925#discussion_r3721034481
########## 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: Right, that fallback contradicted both the filter above it and the doc comment. Rewrote it as a `filter_map` so a status-less update is skipped outright rather than being recorded as a `TaskEnd` that claims `Running`. Added a test for the `status: None` case. ########## 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 scheduler 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]
