add metric for measuring task concurrency blockage

Author: ArunPiduguDD

src/topology/builder.rs

@@ -8,7 +8,7 @@ use std::{
 
 use futures::{FutureExt, StreamExt, TryStreamExt, stream::FuturesOrdered};
 use futures_util::stream::FuturesUnordered;
-use metrics::gauge;
+use metrics::{gauge, histogram};
 use stream_cancel::{StreamExt as StreamCancelExt, Trigger, Tripwire};
 use tokio::{
     select,
@@ -748,14 +748,15 @@ impl<'a> Builder<'a> {
         let sender = self
             .utilization_registry
             .add_component(node.key.clone(), gauge!("utilization"));
-        let runner = Runner::new(
+        let mut runner = Runner::new(
             t,
             input_rx,
             sender,
             node.input_details.data_type(),
             outputs,
             LatencyRecorder::new(self.config.global.latency_ewma_alpha),
         );
+        runner.component_id = node.key.id().to_owned();
         let transform = if node.enable_concurrency {
             runner.run_concurrently().boxed()
         } else {
@@ -1130,6 +1131,7 @@ struct Runner {
     timer_tx: UtilizationComponentSender,
     latency_recorder: LatencyRecorder,
     events_received: Registered<EventsReceived>,
+    component_id: String,
 }
 
 impl Runner {
@@ -1149,6 +1151,7 @@ impl Runner {
             timer_tx,
             latency_recorder,
             events_received: register!(EventsReceived),
+            component_id: String::new(),
         }
     }
 
@@ -1206,13 +1209,15 @@ impl Runner {
 
         let mut in_flight = FuturesOrdered::new();
         let mut shutting_down = false;
+        let mut blocked_completions: u64 = 0;
 
         self.timer_tx.try_send_start_wait();
         loop {
             tokio::select! {
                 biased;
 
                 result = in_flight.next(), if !in_flight.is_empty() => {
+                    blocked_completions += 1;
                     match result {
                         Some(Ok(mut outputs_buf)) => {
                             self.send_outputs(&mut outputs_buf).await
@@ -1223,6 +1228,12 @@ impl Runner {
                 }
 
                 input_arrays = input_rx.next(), if in_flight.len() < *TRANSFORM_CONCURRENCY_LIMIT && !shutting_down => {
+                    histogram!(
+                        "transform_concurrent_scheduling_pressure",
+                        "component_id" => self.component_id.clone()
+                    )
+                    .record((blocked_completions as f64 / *TRANSFORM_CONCURRENCY_LIMIT as f64).min(1.0));
+                    blocked_completions = 0;
                     match input_arrays {
                         Some(input_arrays) => {
                             let mut len = 0;
@@ -1249,6 +1260,7 @@ impl Runner {
                 }
 
                 else => {
+                    blocked_completions = 0;
                     if shutting_down {
                         break
                     }