feat: add new "dashboard" metrics subscriber (#4527)

## Changes Made

- adds a new "dashboard" metrics subscriber that'll make a rest call to
`DAFT_DASHBOARD_METRICS_URL` with the provided metrics information
- small internal refactoring to split `runtime_stats` into module and
submodules




## Related Issues

<!-- Link to related GitHub issues, e.g., "Closes #123" -->

## Checklist

- [ ] Documented in API Docs (if applicable)
- [ ] Documented in User Guide (if applicable)
- [ ] If adding a new documentation page, doc is added to
`docs/mkdocs.yml` navigation
- [ ] Documentation builds and is formatted properly (tag @/ccmao1130
for docs review)

Author: universalmind303

Cargo.lock

@@ -48,7 +48,6 @@ pub fn init_opentelemetry_providers() {
 }
 
 pub fn flush_opentelemetry_providers() {
-    flush_oltp_metrics_provider();
     flush_oltp_tracer_provider();
 }
 
@@ -77,7 +76,7 @@ async fn init_otlp_metrics_provider(otlp_endpoint: &str) {
     *mg = Some(metrics_provider);
 }
 
-fn flush_oltp_metrics_provider() {
+pub fn flush_oltp_metrics_provider() {
     let mg = GLOBAL_METER_PROVIDER.lock().unwrap();
     if let Some(meter_provider) = mg.as_ref() {
         if let Err(e) = meter_provider.force_flush() {

src/common/tracing/src/lib.rs

@@ -36,6 +36,7 @@ parking_lot = {workspace = true}
 pin-project = "1"
 pyo3 = {workspace = true, optional = true}
 pyo3-async-runtimes = {workspace = true, optional = true}
+reqwest = {version = "0.12.19", default-features = false}
 snafu = {workspace = true}
 tokio = {workspace = true}
 tokio-util = {workspace = true}

src/daft-local-execution/Cargo.toml

@@ -17,7 +17,9 @@ use crate::{
     pipeline::{NodeInfo, PipelineNode, RuntimeContext},
     progress_bar::ProgressBarColor,
     resource_manager::MemoryManager,
-    runtime_stats::{CountingReceiver, CountingSender, RuntimeStatsContext},
+    runtime_stats::{
+        CountingReceiver, CountingSender, RuntimeStatsContext, RuntimeStatsEventHandler,
+    },
     ExecutionRuntimeContext, ExecutionTaskSpawner, OperatorOutput, PipelineExecutionSnafu,
 };
 
@@ -124,12 +126,18 @@ impl IntermediateNode {
         receiver: Receiver<Arc<MicroPartition>>,
         sender: Sender<Arc<MicroPartition>>,
         rt_context: Arc<RuntimeStatsContext>,
+        rt_stats_handler: Arc<RuntimeStatsEventHandler>,
         memory_manager: Arc<MemoryManager>,
     ) -> DaftResult<()> {
         let span = info_span!("IntermediateOp::execute");
         let compute_runtime = get_compute_runtime();
-        let task_spawner =
-            ExecutionTaskSpawner::new(compute_runtime, memory_manager, rt_context, span);
+        let task_spawner = ExecutionTaskSpawner::new(
+            compute_runtime,
+            memory_manager,
+            rt_context,
+            rt_stats_handler,
+            span,
+        );
         let mut state = op.make_state()?;
         while let Some(morsel) = receiver.recv().await {
             loop {
@@ -172,6 +180,7 @@ impl IntermediateNode {
                     input_receiver,
                     output_sender,
                     self.runtime_stats.clone(),
+                    runtime_handle.runtime_stats_handler(),
                     memory_manager.clone(),
                 ),
                 self.intermediate_op.name(),
@@ -236,21 +245,27 @@ impl PipelineNode for IntermediateNode {
             true,
             self.runtime_stats.clone(),
         );
+
         for child in &self.children {
             let child_result_receiver = child.start(maintain_order, runtime_handle)?;
             child_result_receivers.push(CountingReceiver::new(
                 child_result_receiver,
                 self.runtime_stats.clone(),
                 progress_bar.clone(),
+                runtime_handle.runtime_stats_handler(),
             ));
         }
         let op = self.intermediate_op.clone();
         let num_workers = op.max_concurrency().context(PipelineExecutionSnafu {
             node_name: self.name(),
         })?;
         let (destination_sender, destination_receiver) = create_channel(0);
-        let counting_sender =
-            CountingSender::new(destination_sender, self.runtime_stats.clone(), progress_bar);
+        let counting_sender = CountingSender::new(
+            destination_sender,
+            self.runtime_stats.clone(),
+            progress_bar,
+            runtime_handle.runtime_stats_handler(),
+        );
 
         let dispatch_spawner = self
             .intermediate_op

src/daft-local-execution/src/intermediate_ops/intermediate_op.rs

@@ -130,6 +130,7 @@ pub(crate) struct ExecutionRuntimeContext {
     default_morsel_size: usize,
     memory_manager: Arc<MemoryManager>,
     progress_bar_manager: Option<Arc<dyn ProgressBarManager>>,
+    rt_stats_handler: Arc<RuntimeStatsEventHandler>,
 }
 
 impl ExecutionRuntimeContext {
@@ -138,12 +139,14 @@ impl ExecutionRuntimeContext {
         default_morsel_size: usize,
         memory_manager: Arc<MemoryManager>,
         progress_bar_manager: Option<Arc<dyn ProgressBarManager>>,
+        rt_stats_handler: Arc<RuntimeStatsEventHandler>,
     ) -> Self {
         Self {
             worker_set: TaskSet::new(),
             default_morsel_size,
             memory_manager,
             progress_bar_manager,
+            rt_stats_handler,
         }
     }
     pub fn spawn_local(
@@ -196,6 +199,11 @@ impl ExecutionRuntimeContext {
     pub(crate) fn memory_manager(&self) -> Arc<MemoryManager> {
         self.memory_manager.clone()
     }
+
+    #[must_use]
+    pub(crate) fn runtime_stats_handler(&self) -> Arc<RuntimeStatsEventHandler> {
+        self.rt_stats_handler.clone()
+    }
 }
 
 impl Drop for ExecutionRuntimeContext {
@@ -210,6 +218,7 @@ pub(crate) struct ExecutionTaskSpawner {
     runtime_ref: RuntimeRef,
     memory_manager: Arc<MemoryManager>,
     runtime_context: Arc<RuntimeStatsContext>,
+    rt_stats_handler: Arc<RuntimeStatsEventHandler>,
     outer_span: tracing::Span,
 }
 
@@ -218,12 +227,14 @@ impl ExecutionTaskSpawner {
         runtime_ref: RuntimeRef,
         memory_manager: Arc<MemoryManager>,
         runtime_context: Arc<RuntimeStatsContext>,
+        rt_stats_handler: Arc<RuntimeStatsEventHandler>,
         span: tracing::Span,
     ) -> Self {
         Self {
             runtime_ref,
             memory_manager,
             runtime_context,
+            rt_stats_handler,
             outer_span: span,
         }
     }
@@ -242,6 +253,7 @@ impl ExecutionTaskSpawner {
         let timed_fut = TimedFuture::new(
             instrumented,
             self.runtime_context.clone(),
+            self.rt_stats_handler.clone(),
             self.outer_span.clone(),
         );
         let memory_manager = self.memory_manager.clone();
@@ -260,6 +272,7 @@ impl ExecutionTaskSpawner {
         let timed_fut = TimedFuture::new(
             instrumented,
             self.runtime_context.clone(),
+            self.rt_stats_handler.clone(),
             self.outer_span.clone(),
         );
         self.runtime_ref.spawn(timed_fut)
@@ -269,6 +282,8 @@ impl ExecutionTaskSpawner {
 #[cfg(feature = "python")]
 use pyo3::prelude::*;
 
+use crate::runtime_stats::RuntimeStatsEventHandler;
+
 #[derive(Debug, Snafu)]
 pub enum Error {
     #[snafu(display("Error joining spawned task: {}", source))]

src/daft-local-execution/src/lib.rs

@@ -73,20 +73,24 @@ impl OperatorProgressBar {
             return false;
         }
 
-        let prev = self.last_update.load(Ordering::Acquire);
-        let elapsed = (now - self.start_time).as_nanos() as u64;
-        let diff = elapsed.saturating_sub(prev);
-
-        // Fast path - check if enough time has passed
-        if diff < Self::UPDATE_INTERVAL {
-            return false;
+        {
+            {
+                let prev = self.last_update.load(Ordering::Acquire);
+                let elapsed = (now - self.start_time).as_nanos() as u64;
+                let diff = elapsed.saturating_sub(prev);
+
+                // Fast path - check if enough time has passed
+                if diff < Self::UPDATE_INTERVAL {
+                    return false;
+                }
+
+                // Only calculate remainder if we're actually going to update
+                let remainder = diff % Self::UPDATE_INTERVAL;
+                self.last_update
+                    .store(elapsed - remainder, Ordering::Release);
+                true
+            }
         }
-
-        // Only calculate remainder if we're actually going to update
-        let remainder = diff % Self::UPDATE_INTERVAL;
-        self.last_update
-            .store(elapsed - remainder, Ordering::Release);
-        true
     }
 
     pub fn render(&self) {

src/daft-local-execution/src/progress_bar.rs

@@ -37,6 +37,7 @@ use crate::{
     },
     progress_bar::{make_progress_bar_manager, ProgressBarManager},
     resource_manager::get_or_init_memory_manager,
+    runtime_stats::RuntimeStatsEventHandler,
     ExecutionRuntimeContext,
 };
 
@@ -240,6 +241,7 @@ pub struct NativeExecutor {
     cancel: CancellationToken,
     runtime: Option<Arc<tokio::runtime::Runtime>>,
     pb_manager: Option<Arc<dyn ProgressBarManager>>,
+    rt_stats_handler: Arc<RuntimeStatsEventHandler>,
     enable_explain_analyze: bool,
 }
 
@@ -248,8 +250,10 @@ impl Default for NativeExecutor {
         Self {
             cancel: CancellationToken::new(),
             runtime: None,
+            // todo: make progressbar another subscriber instances
             pb_manager: should_enable_progress_bar().then(make_progress_bar_manager),
             enable_explain_analyze: should_enable_explain_analyze(),
+            rt_stats_handler: Arc::new(RuntimeStatsEventHandler::new()),
         }
     }
 }
@@ -290,6 +294,7 @@ impl NativeExecutor {
 
         let rt = self.runtime.clone();
         let pb_manager = self.pb_manager.clone();
+        let stats_handler = self.rt_stats_handler.clone();
         let enable_explain_analyze = self.enable_explain_analyze;
         // todo: split this into a run and run_async method
         // the run_async should spawn a task instead of a thread like this
@@ -308,6 +313,7 @@ impl NativeExecutor {
                     cfg.default_morsel_size,
                     memory_manager.clone(),
                     pb_manager,
+                    stats_handler.clone(),
                 );
                 let receiver = pipeline.start(true, &mut runtime_handle)?;
 
@@ -331,7 +337,7 @@ impl NativeExecutor {
 
             let local_set = tokio::task::LocalSet::new();
             local_set.block_on(&runtime, async {
-                tokio::select! {
+                let result = tokio::select! {
                     biased;
                     () = cancel.cancelled() => {
                         log::info!("Execution engine cancelled");
@@ -342,7 +348,14 @@ impl NativeExecutor {
                         Ok(())
                     }
                     result = execution_task => result,
+                };
+
+                // Flush remaining stats events
+                if let Err(e) = stats_handler.flush().await {
+                    log::warn!("Failed to flush runtime stats: {}", e);
                 }
+
+                result
             })?;
             if enable_explain_analyze {
                 let curr_ms = SystemTime::now()