feat(redis): Add counter for commands run (#5345)

Author: loewenheim

Cargo.lock

@@ -30,8 +30,9 @@ serde = { workspace = true }
 tokio = { workspace = true, features = ["time"] }
 thiserror = { workspace = true }
 
+relay-statsd = { workspace = true, optional = true }
 relay-system = { workspace = true }
 
 [features]
 default = []
-impl = ["dep:deadpool", "dep:deadpool-redis", "dep:redis"]
+impl = ["dep:deadpool", "dep:deadpool-redis", "dep:relay-statsd", "dep:redis"]

relay-redis/Cargo.toml

@@ -25,6 +25,9 @@ mod scripts;
 #[cfg(feature = "impl")]
 pub use self::scripts::*;
 
+#[cfg(feature = "impl")]
+mod statsd;
+
 #[cfg(not(feature = "impl"))]
 mod noop;
 

relay-redis/src/lib.rs

@@ -10,6 +10,9 @@ use crate::redis::cluster_async::ClusterConnection;
 use crate::redis::{
     Cmd, IntoConnectionInfo, Pipeline, RedisError, RedisFuture, RedisResult, Value,
 };
+use crate::statsd::RedisCounters;
+
+use relay_statsd::metric;
 
 /// A connection pool for Redis cluster deployments.
 pub type CustomClusterPool = Pool<CustomClusterManager, CustomClusterConnection>;
@@ -34,12 +37,13 @@ impl<C> TrackedConnection<C> {
         }
     }
 
-    /// Returns `true` when a [`RedisError`] should lead to the [`TrackedConnection`] being detached
+    /// Returns `true` when the result of a Redis call should lead to the [`TrackedConnection`] being detached
     /// from the pool.
     ///
-    /// An error leads to the connection being detached if it is unrecoverable.
-    fn should_be_detached(error: &RedisError) -> bool {
-        error.is_unrecoverable_error()
+    /// An `Ok` result never leads to the connection being detached.
+    /// A [`RedisError`] leads to the connection being detached if it is unrecoverable.
+    fn should_be_detached<T>(result: Result<T, &RedisError>) -> bool {
+        result.is_err_and(|error| error.is_unrecoverable_error())
     }
 }
 
@@ -48,9 +52,8 @@ impl<C: redis::aio::ConnectionLike + Send> redis::aio::ConnectionLike for Tracke
         async move {
             let result = self.connection.req_packed_command(cmd).await;
 
-            if let Err(error) = &result {
-                self.detach = self.detach || Self::should_be_detached(error);
-            }
+            self.detach |= Self::should_be_detached(result.as_ref());
+            emit_metrics(result.as_ref());
 
             result
         }
@@ -69,9 +72,8 @@ impl<C: redis::aio::ConnectionLike + Send> redis::aio::ConnectionLike for Tracke
                 .req_packed_commands(cmd, offset, count)
                 .await;
 
-            if let Err(error) = &result {
-                self.detach = self.detach || Self::should_be_detached(error);
-            }
+            self.detach |= Self::should_be_detached(result.as_ref());
+            emit_metrics(result.as_ref());
 
             result
         }
@@ -286,3 +288,16 @@ impl From<Object<CustomSingleManager>> for CustomSingleConnection {
         Self(conn)
     }
 }
+
+fn emit_metrics<T>(result: Result<T, &RedisError>) {
+    let result = match result {
+        Ok(_) => "ok",
+        Err(e) if e.is_timeout() => "timeout",
+        Err(_) => "error",
+    };
+
+    metric!(
+        counter(RedisCounters::CommandExecuted) += 1,
+        result = result,
+    );
+}

relay-redis/src/pool.rs

@@ -0,0 +1,17 @@
+use relay_statsd::CounterMetric;
+
+pub enum RedisCounters {
+    /// Incremented every time a Redis command or pipeline is run.
+    ///
+    /// This metric is tagged with:
+    /// - `result`: The outcome (`ok`, `error`, `timeout`).
+    CommandExecuted,
+}
+
+impl CounterMetric for RedisCounters {
+    fn name(&self) -> &'static str {
+        match self {
+            Self::CommandExecuted => "redis.command_executed",
+        }
+    }
+}