[adapters] Add metrics and statistics for pipeline backpressure stalls.

blp · blp · commit afbc16039b78 · 2026-02-26T17:49:10.000Z
I noticed these showing up in the log and immediately wanted them to show
up elsewhere, since they're important.

Signed-off-by: Ben Pfaff &lt;blp@feldera.com&gt;
diff --git a/crates/adapters/src/controller.rs b/crates/adapters/src/controller.rs
@@ -1230,6 +1230,16 @@ impl Controller {
             labels,
             COMPACTION_STALL_TIME_NANOSECONDS.load(Ordering::Relaxed) as f64 / 1_000_000_000.0,
         );
+        metrics.counter(
+            "output_stall_seconds_total",
+            "Time in seconds that the pipeline was stalled because one or more output connectors' output buffers were full.\n\nThis value is greater than or equal to `output_stall_seconds`.",
+            labels,
+            status.global_metrics.total_output_stall_duration().as_secs_f64());
+        metrics.gauge(
+            "output_stall_seconds",
+            "If the pipeline is currently stalled because one or more output connectors' output buffers were full, this is the time in seconds for which it has been stalled.\n\nIf the pipeline is not currently stalled, this is zero.\n\nIf this is nonzero, then the output connectors causing the stall can be identified by observing which values of `output_connector_queued_records` are greater than or equal to the configured maximum (which defaults to 1,000,000).",
+            labels,
+            status.global_metrics.current_output_stall_duration().as_secs_f64());
         metrics.counter(
             "files_created_total",
             "Total number of files created.",
@@ -2440,7 +2450,12 @@ impl CircuitThread {
 
             // Backpressure in the output pipeline: wait for room in output buffers to
             // become available.
-            if self.controller.output_buffers_full() {
+            let stalled = self.controller.output_buffers_full();
+            self.controller
+                .status
+                .global_metrics
+                .update_output_stall_start(stalled);
+            if stalled {
                 debug!("circuit thread: park waiting for output buffer space");
                 let warning = output_backpressure_warning
                     .get_or_insert_with(|| LongOperationWarning::new(Duration::from_secs(1)));
diff --git a/crates/adapters/src/controller/stats.rs b/crates/adapters/src/controller/stats.rs
@@ -74,7 +74,7 @@ use std::{
         Mutex,
         atomic::{AtomicBool, AtomicU64, Ordering},
     },
-    time::Duration,
+    time::{Duration, Instant},
 };
 use tokio::sync::{broadcast, watch};
 use tracing::{debug, error, info, warn};
@@ -237,6 +237,14 @@ pub struct GlobalControllerMetrics {
     /// all outputs derived from it have been processed by all output connectors.
     pub total_completed_records: AtomicU64,
 
+    /// If the pipeline is stalled because one or more output connectors' output
+    /// buffers are full, this is the time at which the stall began.
+    pub output_stall_start: Mutex<Option<Instant>>,
+
+    /// The amount of time the pipeline has stalled due to output connectors,
+    /// excluding any current stall in `output_stall_start`.
+    pub accumulated_output_stall: Mutex<Duration>,
+
     /// Number of steps that have been initiated.
     ///
     /// # Interpretation
@@ -297,6 +305,8 @@ impl GlobalControllerMetrics {
             total_processed_records: AtomicU64::new(processed_records),
             total_processed_bytes: AtomicU64::new(0),
             total_completed_records: AtomicU64::new(processed_records),
+            output_stall_start: Mutex::new(None),
+            accumulated_output_stall: Mutex::new(Duration::ZERO),
             step_requested: AtomicBool::new(false),
             total_initiated_steps: Atomic::new(0),
             total_completed_steps: Atomic::new(0),
@@ -395,6 +405,28 @@ impl GlobalControllerMetrics {
     pub fn set_commit_progress(&self, commit_progress: Option<CommitProgressSummary>) {
         *self.commit_progress.lock().unwrap() = commit_progress;
     }
+
+    pub fn update_output_stall_start(&self, stalled: bool) {
+        let mut output_stall_start = self.output_stall_start.lock().unwrap();
+        if stalled != output_stall_start.is_some() {
+            if let Some(start) = &*output_stall_start {
+                *self.accumulated_output_stall.lock().unwrap() += start.elapsed();
+            }
+            *output_stall_start = stalled.then(Instant::now);
+        }
+    }
+
+    pub fn current_output_stall_duration(&self) -> Duration {
+        self.output_stall_start
+            .lock()
+            .unwrap()
+            .map(|instant| instant.elapsed())
+            .unwrap_or_default()
+    }
+
+    pub fn total_output_stall_duration(&self) -> Duration {
+        self.current_output_stall_duration() + *self.accumulated_output_stall.lock().unwrap()
+    }
 }
 
 // `ShardedLock` is a read/write lock optimized for fast reads.
@@ -1201,6 +1233,12 @@ impl ControllerStatus {
                 .global_metrics
                 .total_completed_records
                 .load(Ordering::Acquire),
+            output_stall_msecs: self
+                .global_metrics
+                .current_output_stall_duration()
+                .as_millis()
+                .try_into()
+                .unwrap_or(u64::MAX),
             total_initiated_steps: self.global_metrics.total_initiated_steps(),
             total_completed_steps: self.global_metrics.total_completed_steps(),
             pipeline_complete,
diff --git a/crates/feldera-types/src/adapter_stats.rs b/crates/feldera-types/src/adapter_stats.rs
@@ -283,6 +283,16 @@ pub struct ExternalGlobalControllerMetrics {
     pub total_processed_bytes: u64,
     /// Total number of input records processed to completion.
     pub total_completed_records: u64,
+    /// If the pipeline is stalled because one or more output connectors' output
+    /// buffers are full, this is the number of milliseconds that the current
+    /// stall has lasted.
+    ///
+    /// If this is nonzero, then the output connectors causing the stall can be
+    /// identified by noticing `ExternalOutputEndpointMetrics::queued_records`
+    /// is greater than or equal to `ConnectorConfig::max_queued_records`.
+    ///
+    /// In the ordinary case, the pipeline is not stalled, and this value is 0.
+    pub output_stall_msecs: u64,
     /// Number of steps that have been initiated.
     ///
     /// # Interpretation
diff --git a/docs.feldera.com/docs/operations/metrics.md b/docs.feldera.com/docs/operations/metrics.md
@@ -71,6 +71,8 @@ which Feldera is built.
 | <a name='dbsp_runtime_elapsed_seconds_total'>`dbsp_runtime_elapsed_seconds_total`</a> |counter | Time elapsed while the pipeline is executing a step, multiplied by the number of foreground and background threads, in seconds. |
 | <a name='dbsp_step_latency_seconds'>`dbsp_step_latency_seconds`</a> |histogram | Latency of DBSP steps over the last 60 seconds or 1000 steps, whichever is less, in seconds |
 | <a name='dbsp_steps_total'>`dbsp_steps_total`</a> |counter | Total number of DBSP steps executed. |
+| <a name='output_stall_seconds'>`output_stall_seconds`</a> |gauge | If the pipeline is currently stalled because one or more output connectors' output buffers were full, this is the time in seconds for which it has been stalled.<br/><br/>If the pipeline is not currently stalled, this is zero.<br/><br/>If this is nonzero, then the output connectors causing the stall can be identified by observing which values of `output_connector_queued_records` are greater than or equal to the configured maximum (which defaults to 1,000,000). |
+| <a name='output_stall_seconds_total'>`output_stall_seconds_total`</a> |counter | Time in seconds that the pipeline was stalled because one or more output connectors' output buffers were full.<br/><br/>This value is greater than or equal to `output_stall_seconds`. |
 
 ## Record Processing
 
diff --git a/docs.feldera.com/docs/operations/metrics.md.in b/docs.feldera.com/docs/operations/metrics.md.in
@@ -56,7 +56,7 @@ which Feldera is built.
 
 [DBSP]: https://docs.rs/dbsp/latest/dbsp/
 
-{{dbsp_|compaction_}}
+{{dbsp_|compaction_|output_stall_}}
 
 ## Record Processing
 
diff --git a/openapi.json b/openapi.json
@@ -8266,6 +8266,7 @@
           "total_processed_records",
           "total_processed_bytes",
           "total_completed_records",
+          "output_stall_msecs",
           "total_initiated_steps",
           "total_completed_steps",
           "pipeline_complete"
@@ -8312,6 +8313,12 @@
             "description": "Time at which the pipeline process from which we resumed started, in seconds since the epoch.",
             "minimum": 0
           },
+          "output_stall_msecs": {
+            "type": "integer",
+            "format": "int64",
+            "description": "If the pipeline is stalled because one or more output connectors' output\nbuffers are full, this is the number of milliseconds that the current\nstall has lasted.\n\nIf this is nonzero, then the output connectors causing the stall can be\nidentified by noticing `ExternalOutputEndpointMetrics::queued_records`\nis greater than or equal to `ConnectorConfig::max_queued_records`.\n\nIn the ordinary case, the pipeline is not stalled, and this value is 0.",
+            "minimum": 0
+          },
           "pipeline_complete": {
             "type": "boolean",
             "description": "True if the pipeline has processed all input data to completion."
