feat(system): Log service back pressure as metric (#1583)

This adds central back pressure monitoring via statsd metrics to all
services that use the new tokio-based service framework. Channels track
the number of pending messages and emit a gauge metric with that number.

The metric is `service.back_pressure` and tagged with a `service` tag to
identify the specific service. The metric is debounced once per second.
Between these intervals, the back pressure number is not checked, so
this metric is not suitable to detect very short-lived spikes.
Additionally, the current implementation requires that the service makes
progress by polling `recv`.

The purpose of this metric is to detect lasting increases in backlogs
that indicate bottlenecks or degraded system behavior.

Author: jan-auer

.github/workflows/deploy.yml

@@ -131,7 +131,11 @@ jobs:
         uses: actions-rs/cargo@v1
         with:
           command: run
-          args: -p document-metrics -- -o relay_metrics.json relay-server/src/statsd.rs relay-metrics/src/statsd.rs relay-kafka/src/statsd.rs
+          args: -p document-metrics -- -o relay_metrics.json
+            relay-kafka/src/statsd.rs
+            relay-metrics/src/statsd.rs
+            relay-server/src/statsd.rs
+            relay-system/src/statsd.rs
 
       - name: Deploy
         if: github.ref == 'refs/heads/master'

CHANGELOG.md

@@ -1,5 +1,11 @@
 # Changelog
 
+## Unreleased
+
+**Internal**:
+
+- Emit a `service.back_pressure` metric that measures internal back pressure by service. ([#1583](https://github.com/getsentry/relay/pull/1583))
+
 ## 22.11.0
 
 **Features**:

Cargo.lock

@@ -27,6 +27,7 @@ tokio = { version = "1.0", features = ["macros", "time"] }
 criterion = "0.3"
 futures01 = { version = "0.1.28", package = "futures" }
 insta = "1.19.0"
+relay-statsd = { path = "../relay-statsd", features = ["test"] }
 relay-test = { path = "../relay-test" }
 
 [[bench]]

relay-metrics/Cargo.toml

@@ -14,3 +14,7 @@ cadence = "0.26.0"
 parking_lot = "0.12.1"
 rand = "0.7.3"
 relay-log = { path = "../relay-log" }
+
+[features]
+default = []
+test = []

relay-statsd/Cargo.toml

@@ -62,8 +62,7 @@ use std::ops::{Deref, DerefMut};
 use std::sync::Arc;
 
 use cadence::{
-    BufferedUdpMetricSink, Metric, MetricBuilder, QueuingMetricSink, SpyMetricSink, StatsdClient,
-    UdpMetricSink,
+    BufferedUdpMetricSink, Metric, MetricBuilder, QueuingMetricSink, StatsdClient, UdpMetricSink,
 };
 use parking_lot::RwLock;
 use rand::distributions::{Distribution, Uniform};
@@ -164,8 +163,9 @@ pub fn set_client(client: MetricsClient) {
 }
 
 /// Set a test client for the period of the called function (only affects the current thread).
+#[cfg(feature = "test")]
 pub fn with_capturing_test_client(f: impl FnOnce()) -> Vec<String> {
-    let (rx, sink) = SpyMetricSink::new();
+    let (rx, sink) = cadence::SpyMetricSink::new();
     let test_client = MetricsClient {
         statsd_client: StatsdClient::from_sink("", sink),
         default_tags: Default::default(),
@@ -573,7 +573,6 @@ macro_rules! metric {
 
 #[cfg(test)]
 mod tests {
-
     use cadence::{NopMetricSink, StatsdClient};
 
     use crate::{set_client, with_capturing_test_client, with_client, GaugeMetric, MetricsClient};
@@ -608,7 +607,7 @@ mod tests {
         });
 
         assert_eq!(
-            captures.as_slice(),
+            captures,
             [
                 "foo:123|g|#server:server1,host:host1",
                 "bar:456|g|#server:server2,host:host2"

relay-statsd/src/lib.rs

@@ -16,5 +16,10 @@ futures01 = { version = "0.1.28", package = "futures" }
 futures = { version = "0.3", package = "futures", features = ["compat"] }
 once_cell = "1.13.1"
 relay-log = { path = "../relay-log" }
-tokio = { version = "1.0", features = ["rt-multi-thread", "sync"] }
+relay-statsd = { path = "../relay-statsd" }
+tokio = { version = "1.0", features = ["rt-multi-thread", "sync", "macros"] }
 tokio01 = { version = "0.1", package = "tokio" }
+
+[dev-dependencies]
+relay-statsd = { path = "../relay-statsd", features = ["test"] }
+tokio = { version = "1.0", features = ["test-util"] }

relay-system/Cargo.toml

@@ -17,6 +17,7 @@
 pub mod compat;
 mod controller;
 mod service;
+mod statsd;
 
 pub use self::controller::*;
 pub use self::service::*;

relay-system/src/lib.rs

@@ -2,9 +2,18 @@ use std::fmt;
 use std::future::Future;
 use std::marker::PhantomData;
 use std::pin::Pin;
+use std::sync::atomic::{AtomicU64, Ordering};
+use std::sync::Arc;
 use std::task::Context;
+use std::time::Duration;
 
 use tokio::sync::{mpsc, oneshot};
+use tokio::time::MissedTickBehavior;
+
+use crate::statsd::SystemGauges;
+
+/// Interval for recording backlog metrics on service channels.
+const BACKLOG_INTERVAL: Duration = Duration::from_secs(1);
 
 /// A message interface for [services](Service).
 ///
@@ -305,12 +314,14 @@ pub trait FromMessage<M>: Interface {
 /// long as the service is running. It can be freely cloned.
 pub struct Addr<I: Interface> {
     tx: mpsc::UnboundedSender<I>,
+    queue_size: Arc<AtomicU64>,
 }
 
 impl<I: Interface> fmt::Debug for Addr<I> {
     fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
         f.debug_struct("Addr")
             .field("open", &!self.tx.is_closed())
+            .field("queue_size", &self.queue_size.load(Ordering::Relaxed))
             .finish()
     }
 }
@@ -321,6 +332,7 @@ impl<I: Interface> Clone for Addr<I> {
     fn clone(&self) -> Self {
         Self {
             tx: self.tx.clone(),
+            queue_size: self.queue_size.clone(),
         }
     }
 }
@@ -341,6 +353,7 @@ impl<I: Interface> Addr<I> {
         I: FromMessage<M>,
     {
         let (tx, rx) = I::Response::channel();
+        self.queue_size.fetch_add(1, Ordering::SeqCst);
         self.tx.send(I::from_message(message, tx)).ok(); // it's ok to drop, the response will fail
         rx
     }
@@ -352,15 +365,78 @@ impl<I: Interface> Addr<I> {
 ///
 /// Instances are created automatically when [spawning](Service::spawn_handler) a service, or can be
 /// created through [`channel`]. The channel closes when all associated [`Addr`]s are dropped.
-pub type Receiver<I> = mpsc::UnboundedReceiver<I>;
+pub struct Receiver<I: Interface> {
+    rx: mpsc::UnboundedReceiver<I>,
+    name: &'static str,
+    interval: tokio::time::Interval,
+    queue_size: Arc<AtomicU64>,
+}
+
+impl<I: Interface> Receiver<I> {
+    /// Receives the next value for this receiver.
+    ///
+    /// This method returns `None` if the channel has been closed and there are
+    /// no remaining messages in the channel's buffer. This indicates that no
+    /// further values can ever be received from this `Receiver`. The channel is
+    /// closed when all senders have been dropped.
+    ///
+    /// If there are no messages in the channel's buffer, but the channel has
+    /// not yet been closed, this method will sleep until a message is sent or
+    /// the channel is closed.
+    pub async fn recv(&mut self) -> Option<I> {
+        loop {
+            tokio::select! {
+                biased;
+
+                _ = self.interval.tick() => {
+                    let backlog = self.queue_size.load(Ordering::Relaxed);
+                    relay_statsd::metric!(
+                        gauge(SystemGauges::ServiceBackPressure) = backlog,
+                        service = self.name
+                    );
+                },
+                message = self.rx.recv() => {
+                    self.queue_size.fetch_sub(1, Ordering::SeqCst);
+                    return message;
+                },
+            }
+        }
+    }
+}
+
+impl<I: Interface> fmt::Debug for Receiver<I> {
+    fn fmt(&self, f: &mut fmt::Formatter<'_>) -> fmt::Result {
+        f.debug_struct("Receiver")
+            .field("name", &self.name)
+            .field("queue_size", &self.queue_size.load(Ordering::Relaxed))
+            .finish()
+    }
+}
 
 /// Creates an unbounded channel for communicating with a [`Service`].
 ///
 /// The `Addr` as the sending part provides public access to the service, while the `Receiver`
 /// should remain internal to the service.
-pub fn channel<I: Interface>() -> (Addr<I>, Receiver<I>) {
+pub fn channel<I: Interface>(name: &'static str) -> (Addr<I>, Receiver<I>) {
+    let queue_size = Arc::new(AtomicU64::new(0));
     let (tx, rx) = mpsc::unbounded_channel();
-    (Addr { tx }, rx)
+
+    let addr = Addr {
+        tx,
+        queue_size: queue_size.clone(),
+    };
+
+    let mut interval = tokio::time::interval(BACKLOG_INTERVAL);
+    interval.set_missed_tick_behavior(MissedTickBehavior::Skip);
+
+    let receiver = Receiver {
+        rx,
+        name,
+        interval,
+        queue_size,
+    };
+
+    (addr, receiver)
 }
 
 /// An asynchronous unit responding to messages.
@@ -426,8 +502,105 @@ pub trait Service: Sized {
 
     /// Starts the service in the current runtime and returns an address for it.
     fn start(self) -> Addr<Self::Interface> {
-        let (addr, rx) = channel();
+        let (addr, rx) = channel(Self::name());
         self.spawn_handler(rx);
         addr
     }
+
+    /// Returns a unique name for this service implementation.
+    ///
+    /// This is used for internal diagnostics and uses the fully qualified type name of the service
+    /// implementor by default.
+    fn name() -> &'static str {
+        std::any::type_name::<Self>()
+    }
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    struct MockMessage;
+
+    impl Interface for MockMessage {}
+
+    impl FromMessage<Self> for MockMessage {
+        type Response = NoResponse;
+
+        fn from_message(message: Self, _: ()) -> Self {
+            message
+        }
+    }
+
+    struct MockService;
+
+    impl Service for MockService {
+        type Interface = MockMessage;
+
+        fn spawn_handler(self, mut rx: Receiver<Self::Interface>) {
+            tokio::spawn(async move {
+                while rx.recv().await.is_some() {
+                    tokio::time::sleep(BACKLOG_INTERVAL * 2).await;
+                }
+            });
+        }
+
+        fn name() -> &'static str {
+            "mock"
+        }
+    }
+
+    #[test]
+    fn test_backpressure_metrics() {
+        let rt = tokio::runtime::Builder::new_current_thread()
+            .enable_time()
+            .build()
+            .unwrap();
+
+        let _guard = rt.enter();
+        tokio::time::pause();
+
+        // Mock service takes 2 * BACKLOG_INTERVAL for every message
+        let addr = MockService.start();
+
+        // Advance the timer by a tiny offset to trigger the first metric emission.
+        let captures = relay_statsd::with_capturing_test_client(|| {
+            rt.block_on(async {
+                tokio::time::sleep(Duration::from_millis(10)).await;
+            })
+        });
+
+        assert_eq!(captures, ["service.back_pressure:0|g|#service:mock"]);
+
+        // Send messages and advance to 0.5 * INTERVAL. No metrics expected at this point.
+        let captures = relay_statsd::with_capturing_test_client(|| {
+            rt.block_on(async {
+                addr.send(MockMessage); // will be pulled immediately
+                addr.send(MockMessage);
+                addr.send(MockMessage);
+
+                tokio::time::sleep(BACKLOG_INTERVAL / 2).await;
+            })
+        });
+
+        assert!(captures.is_empty());
+
+        // Advance to 6.5 * INTERVAL. The service should pull the first message immediately, another
+        // message every 2 INTERVALS. The messages are fully handled after 6 INTERVALS, but we
+        // cannot observe that since the last message exits the queue at 4.
+        let captures = relay_statsd::with_capturing_test_client(|| {
+            rt.block_on(async {
+                tokio::time::sleep(BACKLOG_INTERVAL * 6).await;
+            })
+        });
+
+        assert_eq!(
+            captures,
+            [
+                "service.back_pressure:2|g|#service:mock", // 2 * INTERVAL
+                "service.back_pressure:1|g|#service:mock", // 4 * INTERVAL
+                "service.back_pressure:0|g|#service:mock", // 6 * INTERVAL
+            ]
+        );
+    }
 }

relay-system/src/service.rs

@@ -0,0 +1,22 @@
+use relay_statsd::GaugeMetric;
+
+/// Gauge metrics for Relay system components.
+pub enum SystemGauges {
+    /// A number of messages queued in a services inbound message channel.
+    ///
+    /// This metric is emitted once per second for every running service. Without backlogs, this
+    /// number should be close to `0`. If this number is monotonically increasing, the service is
+    /// not able to process the inbound message volume.
+    ///
+    /// This metric is tagged with:
+    ///  - `service`: The fully qualified type name of the service implementation.
+    ServiceBackPressure,
+}
+
+impl GaugeMetric for SystemGauges {
+    fn name(&self) -> &'static str {
+        match *self {
+            SystemGauges::ServiceBackPressure => "service.back_pressure",
+        }
+    }
+}