You are viewing a plain text version of this content. The canonical link for it is here.
Posted to commits@datafusion.apache.org by mg...@apache.org on 2026/05/16 08:38:50 UTC
(datafusion-ballista) branch main updated: Merge Executor's brief and extended properties (#1716)
This is an automated email from the ASF dual-hosted git repository.
martin-g pushed a commit to branch main
in repository https://gitbox.apache.org/repos/asf/datafusion-ballista.git
The following commit(s) were added to refs/heads/main by this push:
new baf0b218b Merge Executor's brief and extended properties (#1716)
baf0b218b is described below
commit baf0b218bc5a026b2ab97fe57582a31921599c6f
Author: Martin Grigorov <ma...@users.noreply.github.com>
AuthorDate: Sat May 16 11:38:46 2026 +0300
Merge Executor's brief and extended properties (#1716)
* Merge Executor's brief and extended properties
/api/executors now returns all the information per executor
* Minor improvements
---
ballista-cli/src/tui/app.rs | 29 ++---
ballista-cli/src/tui/domain/executors.rs | 138 +++++++++++++++++----
ballista-cli/src/tui/event.rs | 4 +-
ballista-cli/src/tui/http_client.rs | 5 +-
.../ui/main/executors/executor_details_popup.rs | 7 +-
.../src/tui/ui/main/executors/executors_table.rs | 16 ++-
ballista/scheduler/src/api/handlers.rs | 40 +++---
7 files changed, 159 insertions(+), 80 deletions(-)
diff --git a/ballista-cli/src/tui/app.rs b/ballista-cli/src/tui/app.rs
index d8ee18879..8701ea998 100644
--- a/ballista-cli/src/tui/app.rs
+++ b/ballista-cli/src/tui/app.rs
@@ -343,32 +343,36 @@ impl App {
if self.is_jobs_view() {
self.sort_jobs_by(JobsSortColumn::Status);
} else if self.is_executors_view() {
- self.sort_executors_by(ExecutorsSortColumn::TaskSlots);
+ self.sort_executors_by(ExecutorsSortColumn::CpuCores);
}
}
KeyCode::Char('4') => {
if self.is_jobs_view() {
self.sort_jobs_by(JobsSortColumn::StagesCompleted);
} else if self.is_executors_view() {
- self.sort_executors_by(ExecutorsSortColumn::ProcPhysicalMemoryUsage);
+ self.sort_executors_by(ExecutorsSortColumn::TaskSlots);
}
}
KeyCode::Char('5') => {
if self.is_jobs_view() {
self.sort_jobs_by(JobsSortColumn::PercentComplete);
} else if self.is_executors_view() {
- self.sort_executors_by(ExecutorsSortColumn::PeakPhysicalMemoryUsage);
+ self.sort_executors_by(ExecutorsSortColumn::ProcPhysicalMemoryUsage);
}
}
KeyCode::Char('6') => {
if self.is_jobs_view() {
self.sort_jobs_by(JobsSortColumn::StartTime);
} else if self.is_executors_view() {
- self.sort_executors_by(ExecutorsSortColumn::LastSeen);
+ self.sort_executors_by(ExecutorsSortColumn::PeakPhysicalMemoryUsage);
}
}
- KeyCode::Char('7') if self.is_jobs_view() => {
- self.sort_jobs_by(JobsSortColumn::Duration);
+ KeyCode::Char('7') => {
+ if self.is_jobs_view() {
+ self.sort_jobs_by(JobsSortColumn::Duration);
+ } else if self.is_executors_view() {
+ self.sort_executors_by(ExecutorsSortColumn::LastSeen);
+ }
}
KeyCode::Char('c')
if self.is_jobs_view() && self.input_mode == InputMode::View =>
@@ -689,9 +693,7 @@ mod tests {
use crate::tui::app::{ExecutorsSortColumn, JobsSortColumn, MetricsSortColumn};
use crate::tui::domain::{
SchedulerState, SortOrder,
- executors::{
- Executor, ExecutorDetails, ExecutorDetailsPopup, OsInfo, Specification,
- },
+ executors::{Executor, ExecutorDetailsPopup, OsInfo, Specification},
jobs::Job,
jobs::stages::{JobStagesPopup, JobStagesResponse},
};
@@ -950,12 +952,6 @@ mod tests {
last_seen: None,
specification: Specification { task_slots: 4 },
metrics: vec![],
- }
- }
-
- fn make_executor_details(id: &str) -> ExecutorDetails {
- ExecutorDetails {
- executor_info: make_executor(id),
os_info: OsInfo {
kernel_ver: "5.15".to_string(),
num_disks: 1,
@@ -1000,8 +996,7 @@ mod tests {
#[test]
fn is_executor_details_popup_open_true_when_some() {
let mut app = make_app();
- app.executor_details_popup =
- Some(ExecutorDetailsPopup::new(make_executor_details("e1")));
+ app.executor_details_popup = Some(ExecutorDetailsPopup::new(make_executor("e1")));
assert!(app.is_executor_details_popup_open());
}
diff --git a/ballista-cli/src/tui/domain/executors.rs b/ballista-cli/src/tui/domain/executors.rs
index 5531342b4..d65fb7535 100644
--- a/ballista-cli/src/tui/domain/executors.rs
+++ b/ballista-cli/src/tui/domain/executors.rs
@@ -27,9 +27,14 @@ pub struct Executor {
pub last_seen: Option<u64>,
pub specification: Specification,
pub metrics: Vec<Metric>,
+ pub os_info: OsInfo,
}
impl Executor {
+ pub fn cpu_cores(&self) -> u32 {
+ self.os_info.physical_cores
+ }
+
pub fn task_slots(&self) -> u32 {
self.specification.task_slots
}
@@ -51,12 +56,6 @@ impl Executor {
}
}
-#[derive(Deserialize, Clone, Debug)]
-pub struct ExecutorDetails {
- pub executor_info: Executor,
- pub os_info: OsInfo,
-}
-
#[derive(Deserialize, Clone, Debug)]
pub struct Metric {
#[serde(rename = "type")]
@@ -83,12 +82,12 @@ pub struct OsInfo {
}
pub struct ExecutorDetailsPopup {
- pub executor: ExecutorDetails,
+ pub executor: Executor,
pub scroll_position: u16,
}
impl ExecutorDetailsPopup {
- pub fn new(executor: ExecutorDetails) -> Self {
+ pub fn new(executor: Executor) -> Self {
Self {
executor,
scroll_position: 0,
@@ -113,6 +112,7 @@ pub enum SortColumn {
None,
Host,
Id,
+ CpuCores,
TaskSlots,
ProcPhysicalMemoryUsage,
PeakPhysicalMemoryUsage,
@@ -149,7 +149,7 @@ impl ExecutorsData {
let a_host = format!("{}:{}", a.host, a.port);
let b_host = format!("{}:{}", b.host, b.port);
let cmp = a_host.cmp(&b_host);
- if self.sort_order == crate::tui::domain::SortOrder::Descending {
+ if self.sort_order == SortOrder::Descending {
cmp.reverse()
} else {
cmp
@@ -157,7 +157,15 @@ impl ExecutorsData {
}),
SortColumn::Id => self.executors.sort_by(|a, b| {
let cmp = a.id.cmp(&b.id);
- if self.sort_order == crate::tui::domain::SortOrder::Descending {
+ if self.sort_order == SortOrder::Descending {
+ cmp.reverse()
+ } else {
+ cmp
+ }
+ }),
+ SortColumn::CpuCores => self.executors.sort_by(|a, b| {
+ let cmp = a.cpu_cores().cmp(&b.cpu_cores());
+ if self.sort_order == SortOrder::Descending {
cmp.reverse()
} else {
cmp
@@ -165,7 +173,7 @@ impl ExecutorsData {
}),
SortColumn::TaskSlots => self.executors.sort_by(|a, b| {
let cmp = a.task_slots().cmp(&b.task_slots());
- if self.sort_order == crate::tui::domain::SortOrder::Descending {
+ if self.sort_order == SortOrder::Descending {
cmp.reverse()
} else {
cmp
@@ -175,7 +183,7 @@ impl ExecutorsData {
let cmp = a
.proc_physical_memory_usage()
.cmp(&b.proc_physical_memory_usage());
- if self.sort_order == crate::tui::domain::SortOrder::Descending {
+ if self.sort_order == SortOrder::Descending {
cmp.reverse()
} else {
cmp
@@ -185,7 +193,7 @@ impl ExecutorsData {
let cmp = a
.peak_physical_memory_usage()
.cmp(&b.peak_physical_memory_usage());
- if self.sort_order == crate::tui::domain::SortOrder::Descending {
+ if self.sort_order == SortOrder::Descending {
cmp.reverse()
} else {
cmp
@@ -193,7 +201,7 @@ impl ExecutorsData {
}),
SortColumn::LastSeen => self.executors.sort_by(|a, b| {
let cmp = a.last_seen.cmp(&b.last_seen);
- if self.sort_order == crate::tui::domain::SortOrder::Descending {
+ if self.sort_order == SortOrder::Descending {
cmp.reverse()
} else {
cmp
@@ -283,12 +291,6 @@ mod tests {
value: peak_physical_memory_usage,
},
],
- }
- }
-
- fn make_executor_details(id: &str) -> ExecutorDetails {
- ExecutorDetails {
- executor_info: make_executor("host", 8080, id, 1, 0, 0, Some(0)),
os_info: OsInfo {
kernel_ver: "5.15".to_string(),
num_disks: 1,
@@ -419,6 +421,52 @@ mod tests {
assert_eq!(data.executors[2].id, "id-a");
}
+ #[test]
+ fn sort_by_cpu_cores_ascending() {
+ let mut data = make_executors_data(
+ vec![
+ make_executor("host", 8080, "id-a", 2, 0, 0, Some(100)),
+ make_executor("host", 8080, "id-b", 3, 0, 0, Some(200)),
+ make_executor("host", 8080, "id-c", 1, 0, 0, Some(300)),
+ ],
+ SortColumn::CpuCores,
+ SortOrder::Ascending,
+ );
+ data.executors[0].os_info.physical_cores = 20;
+ data.executors[1].os_info.physical_cores = 200;
+ data.executors[2].os_info.physical_cores = 2;
+ data.sort();
+ assert_eq!(data.executors[0].id, "id-c");
+ assert_eq!(data.executors[0].cpu_cores(), 2);
+ assert_eq!(data.executors[1].id, "id-a");
+ assert_eq!(data.executors[1].cpu_cores(), 20);
+ assert_eq!(data.executors[2].id, "id-b");
+ assert_eq!(data.executors[2].cpu_cores(), 200);
+ }
+
+ #[test]
+ fn sort_by_cpu_cores_descending() {
+ let mut data = make_executors_data(
+ vec![
+ make_executor("host", 8080, "id-a", 1, 0, 0, Some(100)),
+ make_executor("host", 8080, "id-b", 3, 0, 0, Some(200)),
+ make_executor("host", 8080, "id-c", 2, 0, 0, Some(300)),
+ ],
+ SortColumn::CpuCores,
+ SortOrder::Descending,
+ );
+ data.executors[0].os_info.physical_cores = 2;
+ data.executors[1].os_info.physical_cores = 20;
+ data.executors[2].os_info.physical_cores = 200;
+ data.sort();
+ assert_eq!(data.executors[0].id, "id-c");
+ assert_eq!(data.executors[0].cpu_cores(), 200);
+ assert_eq!(data.executors[1].id, "id-b");
+ assert_eq!(data.executors[1].cpu_cores(), 20);
+ assert_eq!(data.executors[2].id, "id-a");
+ assert_eq!(data.executors[2].cpu_cores(), 2);
+ }
+
#[test]
fn sort_by_task_slots_ascending() {
let mut data = make_executors_data(
@@ -711,20 +759,44 @@ mod tests {
#[test]
fn executor_details_popup_new_scroll_position_is_zero() {
- let popup = ExecutorDetailsPopup::new(make_executor_details("id-1"));
+ let popup = ExecutorDetailsPopup::new(make_executor(
+ "host",
+ 8080,
+ "id-1",
+ 1,
+ 0,
+ 0,
+ Some(0),
+ ));
assert_eq!(popup.scroll_position, 0);
}
#[test]
fn executor_details_popup_scroll_down_increments() {
- let mut popup = ExecutorDetailsPopup::new(make_executor_details("id-1"));
+ let mut popup = ExecutorDetailsPopup::new(make_executor(
+ "host",
+ 8080,
+ "id-1",
+ 1,
+ 0,
+ 0,
+ Some(0),
+ ));
popup.scroll_down();
assert_eq!(popup.scroll_position, 1);
}
#[test]
fn executor_details_popup_scroll_down_multiple_times() {
- let mut popup = ExecutorDetailsPopup::new(make_executor_details("id-1"));
+ let mut popup = ExecutorDetailsPopup::new(make_executor(
+ "host",
+ 8080,
+ "id-1",
+ 1,
+ 0,
+ 0,
+ Some(0),
+ ));
popup.scroll_down();
popup.scroll_down();
popup.scroll_down();
@@ -733,7 +805,15 @@ mod tests {
#[test]
fn executor_details_popup_scroll_up_decrements() {
- let mut popup = ExecutorDetailsPopup::new(make_executor_details("id-1"));
+ let mut popup = ExecutorDetailsPopup::new(make_executor(
+ "host",
+ 8080,
+ "id-1",
+ 1,
+ 0,
+ 0,
+ Some(0),
+ ));
popup.scroll_down();
popup.scroll_down();
popup.scroll_up();
@@ -742,7 +822,15 @@ mod tests {
#[test]
fn executor_details_popup_scroll_up_saturates_at_zero() {
- let mut popup = ExecutorDetailsPopup::new(make_executor_details("id-1"));
+ let mut popup = ExecutorDetailsPopup::new(make_executor(
+ "host",
+ 8080,
+ "id-1",
+ 1,
+ 0,
+ 0,
+ Some(0),
+ ));
popup.scroll_up();
popup.scroll_up();
assert_eq!(popup.scroll_position, 0);
diff --git a/ballista-cli/src/tui/event.rs b/ballista-cli/src/tui/event.rs
index caa216bab..1bd16925e 100644
--- a/ballista-cli/src/tui/event.rs
+++ b/ballista-cli/src/tui/event.rs
@@ -21,7 +21,7 @@ use tokio::sync::mpsc;
use crate::tui::domain::{
SchedulerState,
- executors::{Executor, ExecutorDetails},
+ executors::Executor,
jobs::{
Job, JobDetails,
stages::{JobStagesResponse, StagesGraph},
@@ -37,7 +37,7 @@ pub enum UiData {
JobDetails(JobDetails),
JobStagesGraph(StagesGraph),
JobStagesData(String, JobStagesResponse),
- ExecutorDetails(ExecutorDetails),
+ ExecutorDetails(Executor),
}
#[derive(Clone, Debug)]
diff --git a/ballista-cli/src/tui/http_client.rs b/ballista-cli/src/tui/http_client.rs
index 1aa1862e9..c0820a976 100644
--- a/ballista-cli/src/tui/http_client.rs
+++ b/ballista-cli/src/tui/http_client.rs
@@ -20,7 +20,6 @@ use reqwest::{Client, Response};
use serde::de::DeserializeOwned;
use std::time::Duration;
-use crate::tui::domain::executors::ExecutorDetails;
use crate::tui::{
TuiResult,
domain::{
@@ -62,10 +61,10 @@ impl HttpClient {
self.json::<Vec<Executor>>(&url).await
}
- pub async fn get_executor(&self, executor_id: &str) -> TuiResult<ExecutorDetails> {
+ pub async fn get_executor(&self, executor_id: &str) -> TuiResult<Executor> {
let url = self.url(&format!("executor/{}", self.url_encode(executor_id)));
tracing::trace!("Going to GET details for '{}'", &url);
- self.json::<ExecutorDetails>(&url).await
+ self.json::<Executor>(&url).await
}
pub async fn get_jobs(&self) -> TuiResult<Vec<Job>> {
diff --git a/ballista-cli/src/tui/ui/main/executors/executor_details_popup.rs b/ballista-cli/src/tui/ui/main/executors/executor_details_popup.rs
index deb4ea28e..e7f9006f8 100644
--- a/ballista-cli/src/tui/ui/main/executors/executor_details_popup.rs
+++ b/ballista-cli/src/tui/ui/main/executors/executor_details_popup.rs
@@ -16,7 +16,7 @@
// under the License.
use crate::tui::app::App;
-use crate::tui::domain::executors::ExecutorDetails;
+use crate::tui::domain::executors::Executor;
use ratatui::Frame;
use ratatui::prelude::{Color, Style};
use ratatui::text::{Line, Span};
@@ -49,8 +49,7 @@ pub(crate) fn render_executor_details_popup(f: &mut Frame, app: &App) {
f.render_widget(paragraph, area);
}
-fn build_lines<'a>(app: &'a App, executor_details: &'a ExecutorDetails) -> Vec<Line<'a>> {
- let executor = &executor_details.executor_info;
+fn build_lines<'a>(app: &'a App, executor: &'a Executor) -> Vec<Line<'a>> {
let label_style = Style::default().fg(Color::Yellow);
let mut lines = vec![
@@ -68,7 +67,7 @@ fn build_lines<'a>(app: &'a App, executor_details: &'a ExecutorDetails) -> Vec<L
]),
];
- let os = &executor_details.os_info;
+ let os = &executor.os_info;
lines.push(Line::from(""));
lines.push(Line::from(vec![Span::styled(
" OS Info",
diff --git a/ballista-cli/src/tui/ui/main/executors/executors_table.rs b/ballista-cli/src/tui/ui/main/executors/executors_table.rs
index 281a94024..356571bfd 100644
--- a/ballista-cli/src/tui/ui/main/executors/executors_table.rs
+++ b/ballista-cli/src/tui/ui/main/executors/executors_table.rs
@@ -61,6 +61,7 @@ fn render_executors_table(frame: &mut Frame, area: Rect, app: &App) {
let host_suffix = sort_suffix!(SortColumn::Host, sort_column, sort_order);
let id_suffix = sort_suffix!(SortColumn::Id, sort_column, sort_order);
+ let cpu_cores_suffix = sort_suffix!(SortColumn::CpuCores, sort_column, sort_order);
let task_slots_suffix = sort_suffix!(SortColumn::TaskSlots, sort_column, sort_order);
let proc_physical_memory_suffix =
sort_suffix!(SortColumn::ProcPhysicalMemoryUsage, sort_column, sort_order);
@@ -71,6 +72,7 @@ fn render_executors_table(frame: &mut Frame, area: Rect, app: &App) {
let header_row = Row::new(vec![
Cell::from(Text::from(format!("Host{host_suffix}")).centered()),
Cell::from(Text::from(format!("Id{id_suffix}")).centered()),
+ Cell::from(Text::from(format!("CPU cores{cpu_cores_suffix}")).right_aligned()),
Cell::from(Text::from(format!("Task Slots{task_slots_suffix}")).right_aligned()),
Cell::from(
Text::from(format!("Physical Memory{proc_physical_memory_suffix}"))
@@ -100,6 +102,10 @@ fn render_executors_table(frame: &mut Frame, area: Rect, app: &App) {
Text::from(format!("{}:{}", executor.host, executor.port)).centered(),
);
let id_cell = Cell::from(Text::from(executor.id.clone()).centered());
+ let cpu_cores_cell = Cell::from(
+ Text::from(app.format_count(executor.cpu_cores() as usize))
+ .right_aligned(),
+ );
let task_slots_cell = Cell::from(
Text::from(app.format_count(executor.specification.task_slots as usize))
.right_aligned(),
@@ -121,6 +127,7 @@ fn render_executors_table(frame: &mut Frame, area: Rect, app: &App) {
let cells = vec![
host_cell,
id_cell,
+ cpu_cores_cell,
task_slots_cell,
proc_physical_memory_cell,
peak_physical_memory_cell,
@@ -134,10 +141,11 @@ fn render_executors_table(frame: &mut Frame, area: Rect, app: &App) {
[
Constraint::Percentage(10), // Host
Constraint::Percentage(20), // Id
- Constraint::Percentage(10), // Task slots
- Constraint::Percentage(20), // Proc physical memory
- Constraint::Percentage(20), // Peak physical memory
- Constraint::Percentage(20), // Last seen
+ Constraint::Percentage(8), // CPU cores
+ Constraint::Percentage(8), // Task slots
+ Constraint::Percentage(18), // Proc physical memory
+ Constraint::Percentage(18), // Peak physical memory
+ Constraint::Percentage(18), // Last seen
],
)
.block(Block::default().borders(Borders::all()))
diff --git a/ballista/scheduler/src/api/handlers.rs b/ballista/scheduler/src/api/handlers.rs
index b44da81ac..3e3329c75 100644
--- a/ballista/scheduler/src/api/handlers.rs
+++ b/ballista/scheduler/src/api/handlers.rs
@@ -69,18 +69,13 @@ struct SchedulerVersionResponse {
}
#[derive(Debug, serde::Serialize)]
-pub struct ExecutorBriefResponse {
+pub struct ExecutorResponse {
pub id: String,
pub host: String,
pub port: u16,
pub last_seen: Option<u128>,
pub specification: ExecutorSpecification,
pub metrics: Vec<ExecutorMetricResponse>,
-}
-
-#[derive(Debug, serde::Serialize)]
-pub struct ExecutorDetailedResponse {
- pub executor_info: ExecutorBriefResponse,
pub os_info: ExecutorOperatingSystemSpecification,
}
@@ -253,13 +248,13 @@ pub async fn get_executors<
State(data_server): State<Arc<SchedulerServer<T, U>>>,
) -> impl IntoResponse {
let state = &data_server.state;
- let executors: Vec<ExecutorBriefResponse> = state
+ let executors: Vec<ExecutorResponse> = state
.executor_manager
.get_executors_state()
.await
.unwrap_or_default()
.into_iter()
- .map(|(metadata, duration, metrics)| ExecutorBriefResponse {
+ .map(|(metadata, duration, metrics)| ExecutorResponse {
id: metadata.id,
host: metadata.host,
port: metadata.port,
@@ -269,6 +264,7 @@ pub async fn get_executors<
.into_iter()
.filter_map(ExecutorMetricResponse::from_proto)
.collect(),
+ os_info: metadata.os_info,
})
.collect();
@@ -290,23 +286,17 @@ pub async fn get_executor_info<
.unwrap_or_default()
.into_iter()
.find(|(metadata, _, _)| metadata.id == executor_id)
- .map(|(metadata, duration, metrics)| {
- let executor_info = ExecutorBriefResponse {
- id: metadata.id,
- host: metadata.host,
- port: metadata.port,
- last_seen: duration.map(|d| d.as_millis()),
- specification: metadata.specification,
- metrics: metrics
- .into_iter()
- .filter_map(ExecutorMetricResponse::from_proto)
- .collect(),
- };
-
- ExecutorDetailedResponse {
- executor_info,
- os_info: metadata.os_info,
- }
+ .map(|(metadata, duration, metrics)| ExecutorResponse {
+ id: metadata.id,
+ host: metadata.host,
+ port: metadata.port,
+ last_seen: duration.map(|d| d.as_millis()),
+ specification: metadata.specification,
+ metrics: metrics
+ .into_iter()
+ .filter_map(ExecutorMetricResponse::from_proto)
+ .collect(),
+ os_info: metadata.os_info,
});
executor_info
---------------------------------------------------------------------
To unsubscribe, e-mail: commits-unsubscribe@datafusion.apache.org
For additional commands, e-mail: commits-help@datafusion.apache.org