[timeseries] add metrics for ingest backpressure diagnosis (opendata-oss#436)

Author: agavra

common/src/coordinator/handle.rs

@@ -1,3 +1,4 @@
+use super::metrics;
 use super::{BroadcastedView, WriteCommand};
 use super::{Delta, Durability, WriteError, WriteResult};
 use crate::StorageRead;
@@ -6,7 +7,7 @@ use crate::storage::StorageSnapshot;
 use futures::FutureExt;
 use futures::future::Shared;
 use std::sync::Arc;
-use std::time::Duration;
+use std::time::{Duration, Instant};
 use tokio::sync::{broadcast, mpsc, oneshot, watch};
 
 /// A point-in-time view of all applied writes, broadcast by the coordinator
@@ -142,18 +143,21 @@ impl<M: Clone + Send + 'static> WriteHandle<M> {
 /// This is the main interface for interacting with the write coordinator.
 /// It can be cloned and shared across tasks.
 pub struct WriteCoordinatorHandle<D: Delta> {
+    name: Arc<str>,
     write_tx: mpsc::Sender<WriteCommand<D>>,
     watchers: EpochWatcher,
     view: Arc<BroadcastedView<D>>,
 }
 
 impl<D: Delta> WriteCoordinatorHandle<D> {
     pub(crate) fn new(
+        name: String,
         write_tx: mpsc::Sender<WriteCommand<D>>,
         watchers: EpochWatcher,
         view: Arc<BroadcastedView<D>>,
     ) -> Self {
         Self {
+            name: Arc::from(name),
             write_tx,
             watchers,
             view,
@@ -167,6 +171,35 @@ impl<D: Delta> WriteCoordinatorHandle<D> {
     pub fn flushed_epoch(&self) -> u64 {
         *self.watchers.written_rx.borrow()
     }
+
+    /// Sample queue depth on each send. Cheap atomic reads on `mpsc::Sender`.
+    fn record_queue_depth(&self) {
+        let max = self.write_tx.max_capacity();
+        let free = self.write_tx.capacity();
+        ::metrics::gauge!(
+            metrics::COORDINATOR_QUEUE_DEPTH,
+            "channel" => self.name.to_string(),
+        )
+        .set(max.saturating_sub(free) as f64);
+    }
+
+    fn record_send(&self, command: &'static str, status: &'static str, started: Instant) {
+        ::metrics::histogram!(
+            metrics::COORDINATOR_SEND_DURATION_SECONDS,
+            "command" => command,
+            "status" => status,
+        )
+        .record(started.elapsed().as_secs_f64());
+    }
+
+    fn record_backpressure(&self, command: &'static str, reason: &'static str) {
+        ::metrics::counter!(
+            metrics::COORDINATOR_QUEUE_BACKPRESSURE_TOTAL,
+            "command" => command,
+            "reason" => reason,
+        )
+        .increment(1);
+    }
 }
 
 impl<D: Delta> WriteCoordinatorHandle<D> {
@@ -186,27 +219,37 @@ impl<D: Delta> WriteCoordinatorHandle<D> {
         write: D::Write,
         timeout: Duration,
     ) -> Result<WriteHandle<D::ApplyResult>, WriteError<D::Write>> {
+        const COMMAND: &str = "write_timeout";
+        self.record_queue_depth();
+        let started = Instant::now();
         let (tx, rx) = oneshot::channel();
-        self.write_tx
+        let send_result = self
+            .write_tx
             .send_timeout(
                 WriteCommand::Write {
                     write,
                     result_tx: tx,
                 },
                 timeout,
             )
-            .await
-            .map_err(|e| match e {
-                mpsc::error::SendTimeoutError::Timeout(WriteCommand::Write { write, .. }) => {
-                    WriteError::TimeoutError(write)
-                }
-                mpsc::error::SendTimeoutError::Closed(WriteCommand::Write { write, .. }) => {
-                    WriteError::Shutdown
-                }
-                _ => unreachable!("sent a Write command"),
-            })?;
-
-        Ok(WriteHandle::new(rx, self.watchers.clone()))
+            .await;
+        match send_result {
+            Ok(()) => {
+                self.record_send(COMMAND, "ok", started);
+                Ok(WriteHandle::new(rx, self.watchers.clone()))
+            }
+            Err(mpsc::error::SendTimeoutError::Timeout(WriteCommand::Write { write, .. })) => {
+                self.record_send(COMMAND, "timeout", started);
+                self.record_backpressure(COMMAND, "timeout");
+                Err(WriteError::TimeoutError(write))
+            }
+            Err(mpsc::error::SendTimeoutError::Closed(WriteCommand::Write { .. })) => {
+                self.record_send(COMMAND, "shutdown", started);
+                self.record_backpressure(COMMAND, "closed");
+                Err(WriteError::Shutdown)
+            }
+            Err(_) => unreachable!("sent a Write command"),
+        }
     }
 
     /// Submit a write to the coordinator, blocking indefinitely until there is
@@ -222,19 +265,29 @@ impl<D: Delta> WriteCoordinatorHandle<D> {
         &self,
         write: D::Write,
     ) -> Result<WriteHandle<D::ApplyResult>, WriteError<D::Write>> {
+        const COMMAND: &str = "write";
+        self.record_queue_depth();
+        let started = Instant::now();
         let (tx, rx) = oneshot::channel();
-        self.write_tx
+        let send_result = self
+            .write_tx
             .send(WriteCommand::Write {
                 write,
                 result_tx: tx,
             })
-            .await
-            .map_err(|e| match e {
-                mpsc::error::SendError(WriteCommand::Write { write, .. }) => WriteError::Shutdown,
-                _ => unreachable!("sent a Write command"),
-            })?;
-
-        Ok(WriteHandle::new(rx, self.watchers.clone()))
+            .await;
+        match send_result {
+            Ok(()) => {
+                self.record_send(COMMAND, "ok", started);
+                Ok(WriteHandle::new(rx, self.watchers.clone()))
+            }
+            Err(mpsc::error::SendError(WriteCommand::Write { .. })) => {
+                self.record_send(COMMAND, "shutdown", started);
+                self.record_backpressure(COMMAND, "closed");
+                Err(WriteError::Shutdown)
+            }
+            Err(_) => unreachable!("sent a Write command"),
+        }
     }
 
     /// Submit a write to the coordinator.
@@ -247,23 +300,31 @@ impl<D: Delta> WriteCoordinatorHandle<D> {
         &self,
         write: D::Write,
     ) -> Result<WriteHandle<D::ApplyResult>, WriteError<D::Write>> {
+        const COMMAND: &str = "try_write";
+        self.record_queue_depth();
+        let started = Instant::now();
         let (tx, rx) = oneshot::channel();
-        self.write_tx
-            .try_send(WriteCommand::Write {
-                write,
-                result_tx: tx,
-            })
-            .map_err(|e| match e {
-                mpsc::error::TrySendError::Full(WriteCommand::Write { write, .. }) => {
-                    WriteError::Backpressure(write)
-                }
-                mpsc::error::TrySendError::Closed(WriteCommand::Write { write, .. }) => {
-                    WriteError::Shutdown
-                }
-                _ => unreachable!("sent a Write command"),
-            })?;
-
-        Ok(WriteHandle::new(rx, self.watchers.clone()))
+        let send_result = self.write_tx.try_send(WriteCommand::Write {
+            write,
+            result_tx: tx,
+        });
+        match send_result {
+            Ok(()) => {
+                self.record_send(COMMAND, "ok", started);
+                Ok(WriteHandle::new(rx, self.watchers.clone()))
+            }
+            Err(mpsc::error::TrySendError::Full(WriteCommand::Write { write, .. })) => {
+                self.record_send(COMMAND, "backpressure", started);
+                self.record_backpressure(COMMAND, "full");
+                Err(WriteError::Backpressure(write))
+            }
+            Err(mpsc::error::TrySendError::Closed(WriteCommand::Write { .. })) => {
+                self.record_send(COMMAND, "shutdown", started);
+                self.record_backpressure(COMMAND, "closed");
+                Err(WriteError::Shutdown)
+            }
+            Err(_) => unreachable!("sent a Write command"),
+        }
     }
 
     /// Request a flush of the current delta.
@@ -273,18 +334,30 @@ impl<D: Delta> WriteCoordinatorHandle<D> {
     /// to guarantee durability, and the durable watermark will be advanced.
     /// Returns a handle that can be used to wait for the flush to complete.
     pub async fn flush(&self, flush_storage: bool) -> WriteResult<WriteHandle> {
+        const COMMAND: &str = "flush";
+        self.record_queue_depth();
+        let started = Instant::now();
         let (tx, rx) = oneshot::channel();
-        self.write_tx
-            .try_send(WriteCommand::Flush {
-                epoch_tx: tx,
-                flush_storage,
-            })
-            .map_err(|e| match e {
-                mpsc::error::TrySendError::Full(_) => WriteError::Backpressure(()),
-                mpsc::error::TrySendError::Closed(_) => WriteError::Shutdown,
-            })?;
-
-        Ok(WriteHandle::new(rx, self.watchers.clone()))
+        let send_result = self.write_tx.try_send(WriteCommand::Flush {
+            epoch_tx: tx,
+            flush_storage,
+        });
+        match send_result {
+            Ok(()) => {
+                self.record_send(COMMAND, "ok", started);
+                Ok(WriteHandle::new(rx, self.watchers.clone()))
+            }
+            Err(mpsc::error::TrySendError::Full(_)) => {
+                self.record_send(COMMAND, "backpressure", started);
+                self.record_backpressure(COMMAND, "full");
+                Err(WriteError::Backpressure(()))
+            }
+            Err(mpsc::error::TrySendError::Closed(_)) => {
+                self.record_send(COMMAND, "shutdown", started);
+                self.record_backpressure(COMMAND, "closed");
+                Err(WriteError::Shutdown)
+            }
+        }
     }
 
     pub fn view(&self) -> Arc<View<D>> {
@@ -299,6 +372,7 @@ impl<D: Delta> WriteCoordinatorHandle<D> {
 impl<D: Delta> Clone for WriteCoordinatorHandle<D> {
     fn clone(&self) -> Self {
         Self {
+            name: self.name.clone(),
             write_tx: self.write_tx.clone(),
             watchers: self.watchers.clone(),
             view: self.view.clone(),

common/src/coordinator/metrics.rs

@@ -0,0 +1,108 @@
+//! Metrics for the write coordinator.
+//!
+//! These metric names are stable strings shared between the coordinator and
+//! the flush task. Calls are no-ops when no global recorder is installed.
+
+// ── Write-queue ingress (the channel that backs WriteCoordinatorHandle) ──
+
+/// Histogram of how long it takes to enqueue a coordinator command.
+/// Labels: `command` ∈ {`write`, `write_timeout`, `try_write`, `flush`},
+/// `status` ∈ {`ok`, `backpressure`, `timeout`, `shutdown`}.
+pub(crate) const COORDINATOR_SEND_DURATION_SECONDS: &str =
+    "opendata_write_coordinator_send_duration_seconds";
+
+/// Counter incremented when the queue rejects a send because it is full.
+/// Labels: `command`, `reason` ∈ {`full`, `timeout`, `closed`}.
+pub(crate) const COORDINATOR_QUEUE_BACKPRESSURE_TOTAL: &str =
+    "opendata_write_coordinator_queue_backpressure_total";
+
+/// Approximate depth of the write queue, sampled on each send.
+/// Labels: `channel`.
+pub(crate) const COORDINATOR_QUEUE_DEPTH: &str = "opendata_write_coordinator_queue_depth";
+
+// ── Delta lifecycle (apply / freeze / flush dispatch) ──
+
+/// Histogram of `Delta::apply` durations on the coordinator hot path.
+pub(crate) const COORDINATOR_DELTA_APPLY_DURATION_SECONDS: &str =
+    "opendata_write_coordinator_delta_apply_duration_seconds";
+
+/// Estimated current delta size in bytes (gauge), sampled after each apply.
+pub(crate) const COORDINATOR_DELTA_ESTIMATED_BYTES: &str =
+    "opendata_write_coordinator_delta_estimated_bytes";
+
+/// Histogram of `Delta::freeze` durations.
+pub(crate) const COORDINATOR_DELTA_FREEZE_DURATION_SECONDS: &str =
+    "opendata_write_coordinator_delta_freeze_duration_seconds";
+
+/// Histogram of how long the coordinator waits when sending a flush event to
+/// the flush task. A high value means the flush task is the bottleneck and
+/// new writes will not be accepted until the event is dispatched.
+pub(crate) const COORDINATOR_FLUSH_EVENT_SEND_DURATION_SECONDS: &str =
+    "opendata_write_coordinator_flush_event_send_duration_seconds";
+
+/// Approximate depth of the flush event channel, sampled on each send.
+pub(crate) const COORDINATOR_FLUSH_EVENT_QUEUE_DEPTH: &str =
+    "opendata_write_coordinator_flush_event_queue_depth";
+
+/// Counter of flush triggers, labeled by reason.
+/// Labels: `reason` ∈ {`size_threshold`, `interval`, `explicit`, `shutdown`}.
+pub(crate) const COORDINATOR_FLUSH_TOTAL: &str = "opendata_write_coordinator_flush_total";
+
+// ── Flush task (background) ──
+
+/// Histogram of `Flusher::flush_delta` durations on the background flush task.
+pub(crate) const COORDINATOR_FLUSH_DELTA_DURATION_SECONDS: &str =
+    "opendata_write_coordinator_flush_delta_duration_seconds";
+
+/// Histogram of `Flusher::flush_storage` durations on the background flush task.
+pub(crate) const COORDINATOR_FLUSH_STORAGE_DURATION_SECONDS: &str =
+    "opendata_write_coordinator_flush_storage_duration_seconds";
+
+/// Describe all coordinator metrics on the global recorder. Safe to call more
+/// than once; the `metrics` crate dedupes descriptions.
+pub fn describe_coordinator_metrics() {
+    metrics::describe_histogram!(
+        COORDINATOR_SEND_DURATION_SECONDS,
+        "Latency of submitting a command to the write coordinator queue (seconds)"
+    );
+    metrics::describe_counter!(
+        COORDINATOR_QUEUE_BACKPRESSURE_TOTAL,
+        "Coordinator queue rejections due to a full or closed queue"
+    );
+    metrics::describe_gauge!(
+        COORDINATOR_QUEUE_DEPTH,
+        "Approximate write coordinator queue depth, sampled on each send"
+    );
+    metrics::describe_histogram!(
+        COORDINATOR_DELTA_APPLY_DURATION_SECONDS,
+        "Time spent in Delta::apply on the coordinator hot path (seconds)"
+    );
+    metrics::describe_gauge!(
+        COORDINATOR_DELTA_ESTIMATED_BYTES,
+        "Estimated bytes in the current (mutable) delta"
+    );
+    metrics::describe_histogram!(
+        COORDINATOR_DELTA_FREEZE_DURATION_SECONDS,
+        "Time spent freezing the current delta on the coordinator hot path (seconds)"
+    );
+    metrics::describe_histogram!(
+        COORDINATOR_FLUSH_EVENT_SEND_DURATION_SECONDS,
+        "Time spent dispatching a flush event from the coordinator to the flush task (seconds)"
+    );
+    metrics::describe_gauge!(
+        COORDINATOR_FLUSH_EVENT_QUEUE_DEPTH,
+        "Approximate depth of the coordinator's flush-event channel"
+    );
+    metrics::describe_counter!(
+        COORDINATOR_FLUSH_TOTAL,
+        "Coordinator flush triggers labeled by reason"
+    );
+    metrics::describe_histogram!(
+        COORDINATOR_FLUSH_DELTA_DURATION_SECONDS,
+        "Background Flusher::flush_delta duration (seconds)"
+    );
+    metrics::describe_histogram!(
+        COORDINATOR_FLUSH_STORAGE_DURATION_SECONDS,
+        "Background Flusher::flush_storage duration (seconds)"
+    );
+}