Add actual task scheduling latency tracking

- Add schedulable_time_millis field to TaskDescription to track when a task became schedulable (when its stage transitioned to running state)
- Update all TaskDescription creation sites to pass RunningStage.stage_running_time
- Calculate actual scheduling latency in record_task_scheduled calls by computing the difference between current time and schedulable_time_millis
- This enables accurate scheduler_task_scheduling_latency_ms metrics instead of the previous placeholder value of 0

Author: phillipleblanc

ballista/scheduler/src/cluster/mod.rs

@@ -512,6 +512,7 @@ pub(crate) async fn bind_task_bias(
                     task_attempt: running_stage.task_failure_numbers[partition_id],
                     plan: running_stage.plan.clone(),
                     session_config: running_stage.session_config.clone(),
+                    schedulable_time_millis: running_stage.stage_running_time,
                 };
                 result.bound_tasks.push((executor_id, task_desc));
 
@@ -614,6 +615,7 @@ pub(crate) async fn bind_task_round_robin(
                     task_attempt: running_stage.task_failure_numbers[partition_id],
                     plan: running_stage.plan.clone(),
                     session_config: running_stage.session_config.clone(),
+                    schedulable_time_millis: running_stage.stage_running_time,
                 };
                 result.bound_tasks.push((executor_id, task_desc));
 
@@ -762,6 +764,7 @@ pub(crate) async fn bind_task_consistent_hash(
                                 [partition_id],
                             plan: running_stage.plan.clone(),
                             session_config: running_stage.session_config.clone(),
+                            schedulable_time_millis: running_stage.stage_running_time,
                         };
                         result.bound_tasks.push((executor_id, task_desc));
 

ballista/scheduler/src/scheduler_server/grpc.rs

@@ -161,17 +161,22 @@ impl<T: 'static + AsLogicalPlan, U: 'static + AsExecutionPlan> SchedulerGrpc
             }
 
             let mut tasks = vec![];
+            let now_millis = SystemTime::now()
+                .duration_since(UNIX_EPOCH)
+                .unwrap_or_default()
+                .as_millis();
+
             for (_, task) in binding_result.bound_tasks {
                 let job_id = task.partition.job_id.clone();
                 let stage_id = task.partition.stage_id;
 
-                // Record task scheduling metric
-                // Note: latency_ms is 0 since we don't currently track when tasks became schedulable.
+                // Record task scheduling metric with actual latency
+                let latency_ms = now_millis.saturating_sub(task.schedulable_time_millis);
                 self.state.metrics_collector.record_task_scheduled(
                     &job_id,
                     stage_id,
                     &executor_id,
-                    0,
+                    latency_ms as u64,
                 );
 
                 match self.state.task_manager.prepare_task_definition(task) {

ballista/scheduler/src/state/execution_graph.rs

@@ -1135,7 +1135,8 @@ impl ExecutionGraph {
                     task_id,
                     task_attempt,
                     plan: stage.plan.clone(),
-                    session_config: self.session_config.clone()
+                    session_config: self.session_config.clone(),
+                    schedulable_time_millis: stage.stage_running_time,
                 })
             } else {
                 Err(BallistaError::General(format!("Stage {stage_id} is not a running stage")))
@@ -1728,6 +1729,9 @@ pub struct TaskDescription {
     pub plan: Arc<dyn ExecutionPlan>,
     /// Session configuration for this task's execution context.
     pub session_config: Arc<SessionConfig>,
+    /// Timestamp (millis since epoch) when this task became schedulable.
+    /// This is when the stage transitioned to running state.
+    pub schedulable_time_millis: u128,
 }
 
 impl Debug for TaskDescription {

ballista/scheduler/src/state/mod.rs

@@ -23,7 +23,7 @@ use datafusion::physical_plan::{ExecutionPlan, ExecutionPlanProperties};
 use std::any::type_name;
 use std::collections::HashMap;
 use std::sync::Arc;
-use std::time::Instant;
+use std::time::{Instant, SystemTime, UNIX_EPOCH};
 
 use crate::scheduler_server::event::QueryStageSchedulerEvent;
 
@@ -302,15 +302,21 @@ impl<T: 'static + AsLogicalPlan, U: 'static + AsExecutionPlan> SchedulerState<T,
         &self,
         bound_tasks: Vec<BoundTask>,
     ) -> Result<Vec<ExecutorSlot>> {
+        // Get current time once for all latency calculations
+        let now_millis = SystemTime::now()
+            .duration_since(UNIX_EPOCH)
+            .unwrap_or_default()
+            .as_millis();
+
         // Record task scheduling metrics for each task
         for (executor_id, task) in &bound_tasks {
-            // Note: latency_ms is 0 since we don't currently track when tasks became schedulable.
-            // This could be enhanced by adding a schedulable_time field to TaskDescription.
+            // Calculate scheduling latency: time from when task became schedulable to now
+            let latency_ms = now_millis.saturating_sub(task.schedulable_time_millis);
             self.metrics_collector.record_task_scheduled(
                 &task.partition.job_id,
                 task.partition.stage_id,
                 executor_id,
-                0, // latency_ms placeholder
+                latency_ms as u64,
             );
         }