upmerge

ballista/rust/scheduler/src/api/handlers.rs

@@ -20,9 +20,6 @@ use warp::Rejection;
 #[derive(Debug, serde::Serialize)]
 struct StateResponse {
     executors: Vec<ExecutorMetaResponse>,
-    active_jobs: Vec<String>,
-    completed_jobs: Vec<String>,
-    failed_jobs: Vec<String>,
     started: u128,
     version: &'static str,
 }
@@ -35,6 +32,15 @@ pub struct ExecutorMetaResponse {
     pub last_seen: u128,
 }
 
+#[derive(Debug, serde::Serialize)]
+pub struct JobResponse {
+    pub job_id: String,
+    pub job_status: String,
+    pub num_stages: usize,
+    pub completed_stages: usize,
+    pub percent_complete: u8,
+}
+
 /// Return current scheduler state, including list of executors and active, completed, and failed
 /// job ids.
 pub(crate) async fn get_state<T: AsLogicalPlan, U: AsExecutionPlan>(
@@ -58,27 +64,40 @@ pub(crate) async fn get_state<T: AsLogicalPlan, U: AsExecutionPlan>(
 
     let response = StateResponse {
         executors,
-        active_jobs: state
-            .task_manager
-            .get_active_job_ids()
-            .await
-            .map_err(|_| warp::reject())?,
-        completed_jobs: state
-            .task_manager
-            .get_completed_job_ids()
-            .await
-            .map_err(|_| warp::reject())?,
-        failed_jobs: state
-            .task_manager
-            .get_failed_job_ids()
-            .await
-            .map_err(|_| warp::reject())?,
         started: data_server.start_time,
         version: BALLISTA_VERSION,
     };
     Ok(warp::reply::json(&response))
 }
 
+/// Return list of jobs
+pub(crate) async fn get_jobs<T: AsLogicalPlan, U: AsExecutionPlan>(
+    data_server: SchedulerServer<T, U>,
+) -> Result<impl warp::Reply, Rejection> {
+    // TODO: Display last seen information in UI
+    let state = data_server.state;
+
+    let jobs = state
+        .task_manager
+        .get_jobs()
+        .await
+        .map_err(|_| warp::reject())?;
+
+    let jobs: Vec<JobResponse> = jobs
+        .iter()
+        .map(|job| JobResponse {
+            job_id: job.job_id.to_string(),
+            job_status: format!("{:?}", job.status),
+            num_stages: job.num_stages,
+            completed_stages: job.completed_stages,
+            percent_complete: ((job.completed_stages as f32 / job.num_stages as f32)
+                * 100_f32) as u8,
+        })
+        .collect();
+
+    Ok(warp::reply::json(&jobs))
+}
+
 #[derive(Debug, serde::Serialize)]
 pub struct JobSummaryResponse {
     /// Just show debug output for now but what we really want here is a list of stages with
@@ -107,7 +126,6 @@ pub(crate) async fn get_job_summary<T: AsLogicalPlan, U: AsExecutionPlan>(
         })),
     }
 }
-
 /// Generate a dot graph for the specified job id and return as plain text
 pub(crate) async fn get_job_dot_graph<T: AsLogicalPlan, U: AsExecutionPlan>(
     data_server: SchedulerServer<T, U>,

ballista/rust/scheduler/src/api/mod.rs

@@ -85,18 +85,25 @@ fn with_data_server<T: AsLogicalPlan + Clone, U: 'static + AsExecutionPlan>(
 pub fn get_routes<T: AsLogicalPlan + Clone, U: 'static + AsExecutionPlan>(
     scheduler_server: SchedulerServer<T, U>,
 ) -> BoxedFilter<(impl Reply,)> {
-    let route_state = warp::path("state")
+    let route_state = warp::path!("api" / "state")
         .and(with_data_server(scheduler_server.clone()))
         .and_then(handlers::get_state);
 
-    let route_job_summary = warp::path!("jobs" / String)
+    let route_jobs = warp::path!("api" / "jobs")
+        .and(with_data_server(scheduler_server.clone()))
+        .and_then(|data_server| handlers::get_jobs(data_server));
+
+    let route_job_summary = warp::path!("api" / "job" / String)
         .and(with_data_server(scheduler_server.clone()))
         .and_then(|job_id, data_server| handlers::get_job_summary(data_server, job_id));
 
-    let route_job_dot = warp::path!("jobs" / String / "dot")
+    let route_job_dot = warp::path!("api" / "job" / String / "dot")
         .and(with_data_server(scheduler_server))
         .and_then(|job_id, data_server| handlers::get_job_dot_graph(data_server, job_id));
 
-    let routes = route_state.or(route_job_dot).or(route_job_summary);
+    let routes = route_state
+        .or(route_jobs)
+        .or(route_job_summary)
+        .or(route_job_dot);
     routes.boxed()
 }

ballista/rust/scheduler/src/state/task_manager.rs

@@ -107,21 +107,35 @@ impl<T: 'static + AsLogicalPlan, U: 'static + AsExecutionPlan> TaskManager<T, U>
     }
 
     /// Get a list of active job ids
-    pub async fn get_active_job_ids(&self) -> Result<Vec<String>> {
-        let keys = self.state.scan_keys(Keyspace::ActiveJobs).await?;
-        Ok(keys.iter().cloned().collect::<Vec<_>>())
-    }
-
-    /// Get a list of completed job ids
-    pub async fn get_completed_job_ids(&self) -> Result<Vec<String>> {
-        let keys = self.state.scan_keys(Keyspace::CompletedJobs).await?;
-        Ok(keys.iter().cloned().collect::<Vec<_>>())
-    }
+    pub async fn get_jobs(&self) -> Result<Vec<JobOverview>> {
+        let mut job_ids = vec![];
+        for job_id in self.state.scan_keys(Keyspace::ActiveJobs).await? {
+            job_ids.push(job_id);
+        }
+        for job_id in self.state.scan_keys(Keyspace::CompletedJobs).await? {
+            job_ids.push(job_id);
+        }
+        for job_id in self.state.scan_keys(Keyspace::FailedJobs).await? {
+            job_ids.push(job_id);
+        }
 
-    /// Get a list of failed job ids
-    pub async fn get_failed_job_ids(&self) -> Result<Vec<String>> {
-        let keys = self.state.scan_keys(Keyspace::FailedJobs).await?;
-        Ok(keys.iter().cloned().collect::<Vec<_>>())
+        let mut jobs = vec![];
+        for job_id in &job_ids {
+            let graph = self.get_execution_graph(job_id).await?;
+            let mut completed_stages = 0;
+            for stage in graph.stages().values() {
+                if let ExecutionStage::Completed(_) = stage {
+                    completed_stages += 1;
+                }
+            }
+            jobs.push(JobOverview {
+                job_id: job_id.clone(),
+                status: graph.status(),
+                num_stages: graph.stage_count(),
+                completed_stages,
+            });
+        }
+        Ok(jobs)
     }
 
     /// Get the status of of a job. First look in the active cache.