Add initial support for redis-sentinel

Author: ds

relay-config/src/redis.rs

@@ -92,6 +92,19 @@ enum RedisConfigFromFile {
         #[serde(flatten)]
         options: PartialRedisConfigOptions,
     },
+
+    /// Connect to a Sentinel cluster.
+    Sentinel {
+        // List of redis:// urls of sentinel nodes.
+        sentinel_nodes: Vec<String>,
+
+        /// Name of a monitored master from sentinel cluster.
+        master_name: String,
+
+        /// Additional configuration options for the redis client and a connections pool.
+        #[serde(flatten)]
+        options: PartialRedisConfigOptions,
+    },
 }
 
 /// Redis configuration.
@@ -108,6 +121,17 @@ pub enum RedisConfig {
     },
     /// Connect to a single Redis instance.
     Single(SingleRedisConfig),
+
+    /// Connect to a Sentinel cluster.
+    Sentinel {
+        /// List of redis:// urls of sentinel nodes.
+        sentinel_nodes: Vec<String>,
+        /// Name of a monitored master from sentinel cluster.
+        master_name: String,
+        /// Options of the Redis config.
+        #[serde(flatten)]
+        options: PartialRedisConfigOptions,
+    },
 }
 
 /// Struct that can serialize a string to a single Redis connection.
@@ -155,6 +179,15 @@ impl From<RedisConfigFromFile> for RedisConfig {
             RedisConfigFromFile::SingleWithOpts { server, options } => {
                 Self::Single(SingleRedisConfig::Detailed { server, options })
             }
+            RedisConfigFromFile::Sentinel {
+                sentinel_nodes,
+                master_name,
+                options,
+            } => Self::Sentinel {
+                sentinel_nodes,
+                master_name,
+                options,
+            },
         }
     }
 }
@@ -193,6 +226,15 @@ pub enum RedisConfigRef<'a> {
         /// Options of the Redis config.
         options: RedisConfigOptions,
     },
+    /// Connect to a Sentinel cluster.
+    Sentinel {
+        /// Reference to the Redis nodes urls of the cluster.
+        sentinel_nodes: &'a [String],
+        /// Name of a monitored master from sentinel cluster.
+        master_name: &'a str,
+        /// Options of the Redis config.
+        options: RedisConfigOptions,
+    },
 }
 
 /// Helper struct bundling connections and options for the various Redis pools.
@@ -253,6 +295,15 @@ pub(super) fn build_redis_config(
             server,
             options: Default::default(),
         },
+        RedisConfig::Sentinel {
+            sentinel_nodes,
+            master_name,
+            options,
+        } => RedisConfigRef::Sentinel {
+            sentinel_nodes,
+            master_name,
+            options: build_redis_config_options(options, default_connections),
+        },
     }
 }
 
@@ -521,6 +572,35 @@ max_connections: 20
         );
     }
 
+    #[test]
+    fn test_redis_sentinel_nodes_opts() {
+        let yaml = r#"
+sentinel_nodes:
+    - "redis://127.0.0.1:26379"
+    - "redis://127.0.0.2:26379"
+master_name: sentry-redis
+max_connections: 10
+"#;
+
+        let config: RedisConfig = serde_yaml::from_str(yaml)
+            .expect("Parsed processing redis config: sentinel with options");
+
+        assert_eq!(
+            config,
+            RedisConfig::Sentinel {
+                sentinel_nodes: vec![
+                    "redis://127.0.0.1:26379".to_owned(),
+                    "redis://127.0.0.2:26379".to_owned()
+                ],
+                master_name: "sentry-redis".to_owned(),
+                options: PartialRedisConfigOptions {
+                    max_connections: Some(10),
+                    ..Default::default()
+                },
+            }
+        );
+    }
+
     #[test]
     fn test_redis_cluster_serialize() {
         let config = RedisConfig::Cluster {
@@ -634,4 +714,34 @@ max_connections: 20
         }
         "###);
     }
+
+    #[test]
+    fn test_redis_sentinel_serialize() {
+        let config = RedisConfig::Sentinel {
+            sentinel_nodes: vec![
+                "redis://127.0.0.1:26379".to_owned(),
+                "redis://127.0.0.2:26379".to_owned(),
+            ],
+            master_name: "sentry-redis".to_owned(),
+            options: PartialRedisConfigOptions {
+                max_connections: Some(42),
+                ..Default::default()
+            },
+        };
+
+        assert_json_snapshot!(config, @r###"
+        {
+          "sentinel_nodes": [
+            "redis://127.0.0.1:26379",
+            "redis://127.0.0.2:26379"
+          ],
+          "master_name": "sentry-redis",
+          "max_connections": 42,
+          "idle_timeout": 60,
+          "create_timeout": 3,
+          "recycle_timeout": 2,
+          "recycle_check_frequency": 100
+        }
+        "###);
+    }
 }

relay-redis/src/pool.rs

@@ -1,6 +1,7 @@
 use deadpool::managed::{Manager, Metrics, Object, Pool, RecycleError, RecycleResult};
 use deadpool_redis::Manager as SingleManager;
 use deadpool_redis::cluster::Manager as ClusterManager;
+use deadpool_redis::sentinel::{Manager as SentinelManager, SentinelServerType};
 use futures::FutureExt;
 use std::ops::{Deref, DerefMut};
 
@@ -17,6 +18,9 @@ pub type CustomClusterPool = Pool<CustomClusterManager, CustomClusterConnection>
 /// A connection pool for single Redis instance deployments.
 pub type CustomSinglePool = Pool<CustomSingleManager, CustomSingleConnection>;
 
+/// A connection pool for single Redis sentinel deployments.
+pub type CustomSentinelPool = Pool<CustomSentinelManager, CustomSentinelConnection>;
+
 /// A wrapper for a connection that can be tracked with metadata.
 ///
 /// A connection is considered detached as soon as it is marked as detached and it can't be
@@ -280,3 +284,93 @@ impl From<Object<CustomSingleManager>> for CustomSingleConnection {
         Self(conn)
     }
 }
+
+/// Managed connection to a Redis-master instance
+///
+/// This manager handles the creation and recycling of Redis connections,
+/// ensuring proper connection health through periodic PING checks. It supports
+/// multiplexed connections for efficient handling of multiple operations.
+pub struct CustomSentinelConnection(Object<CustomSentinelManager>);
+
+impl redis::aio::ConnectionLike for CustomSentinelConnection {
+    fn req_packed_command<'a>(&'a mut self, cmd: &'a Cmd) -> RedisFuture<'a, Value> {
+        self.0.req_packed_command(cmd)
+    }
+
+    fn req_packed_commands<'a>(
+        &'a mut self,
+        cmd: &'a Pipeline,
+        offset: usize,
+        count: usize,
+    ) -> RedisFuture<'a, Vec<Value>> {
+        self.0.req_packed_commands(cmd, offset, count)
+    }
+
+    fn get_db(&self) -> i64 {
+        self.0.get_db()
+    }
+}
+
+/// Manages Redis-master connections and their lifecycle.
+///
+/// This manager handles the creation and recycling of Redis connections,
+/// ensuring proper connection health through periodic PING checks. It supports
+/// multiplexed connections for efficient handling of multiple operations.
+pub struct CustomSentinelManager {
+    manager: SentinelManager,
+    recycle_check_frequency: usize,
+}
+
+impl CustomSentinelManager {
+    /// Creates a new Sentinel manager for the specified Sentinel nodes.
+    ///
+    /// The manager will attempt to connect to each node in the provided list and
+    /// maintain connections to the Redis cluster.
+    pub fn new<T: IntoConnectionInfo>(
+        params: Vec<T>,
+        master_name: String,
+        recycle_check_frequency: usize,
+    ) -> RedisResult<Self> {
+        Ok(Self {
+            manager: SentinelManager::new(params, master_name, None, SentinelServerType::Master)?,
+            recycle_check_frequency,
+        })
+    }
+}
+
+impl Manager for CustomSentinelManager {
+    type Type = TrackedConnection<MultiplexedConnection>;
+    type Error = RedisError;
+
+    async fn create(&self) -> Result<TrackedConnection<MultiplexedConnection>, RedisError> {
+        self.manager.create().await.map(TrackedConnection::from)
+    }
+
+    async fn recycle(
+        &self,
+        conn: &mut TrackedConnection<MultiplexedConnection>,
+        metrics: &Metrics,
+    ) -> RecycleResult<RedisError> {
+        // If the connection is marked to be detached, we return and error, signaling that this
+        // connection must be detached from the pool.
+        if conn.detach {
+            return Err(RecycleError::Message(
+                "the tracked connection was marked as detached".into(),
+            ));
+        }
+
+        // If the interval has been reached, we optimistically assume the connection is active
+        // without doing an actual `PING`.
+        if metrics.recycle_count % self.recycle_check_frequency != 0 {
+            return Ok(());
+        }
+
+        self.manager.recycle(conn, metrics).await
+    }
+}
+
+impl From<Object<CustomSentinelManager>> for CustomSentinelConnection {
+    fn from(conn: Object<CustomSentinelManager>) -> Self {
+        Self(conn)
+    }
+}

relay-redis/src/real.rs

@@ -79,6 +79,8 @@ pub enum AsyncRedisClient {
     Cluster(pool::CustomClusterPool),
     /// Contains a connection pool to a single Redis instance.
     Single(pool::CustomSinglePool),
+    /// Contains a connection pool to a Redis-master instance.
+    Sentinel(pool::CustomSentinelPool),
 }
 
 impl AsyncRedisClient {
@@ -126,6 +128,37 @@ impl AsyncRedisClient {
         Ok(AsyncRedisClient::Single(pool))
     }
 
+    /// Creates a new connection client for a Redis-master instance.
+    ///
+    /// This method initializes a connection client that can communicate with multiple Sentinel nodes
+    /// to retrieve connection to a current Redis-master instance.
+    /// The client is configured with the specified sentinel servers, master name and options.
+    ///
+    /// The client uses a custom Sentinel manager that implements a specific connection recycling
+    /// strategy, ensuring optimal performance and reliability in single-instance environments.
+    pub fn sentinel<'a>(
+        sentinels: impl IntoIterator<Item = &'a str>,
+        master_name: &str,
+        opts: &RedisConfigOptions,
+    ) -> Result<Self, RedisError> {
+        let sentinels = sentinels
+            .into_iter()
+            .map(|s| s.to_owned())
+            .collect::<Vec<_>>();
+        // We use our custom single manager which performs recycling in a different way from the
+        // default manager.
+        let manager = pool::CustomSentinelManager::new(
+            sentinels,
+            master_name.to_owned(),
+            opts.recycle_check_frequency,
+        )
+        .map_err(RedisError::Redis)?;
+
+        let pool = Self::build_pool(manager, opts)?;
+
+        Ok(AsyncRedisClient::Sentinel(pool))
+    }
+
     /// Acquires a connection from the pool.
     ///
     /// Returns a new [`AsyncRedisConnection`] that can be used to execute Redis commands.
@@ -138,6 +171,9 @@ impl AsyncRedisClient {
             Self::Single(pool) => {
                 AsyncRedisConnection::Single(pool.get().await.map_err(RedisError::Pool)?)
             }
+            Self::Sentinel(pool) => {
+                AsyncRedisConnection::Sentinel(pool.get().await.map_err(RedisError::Pool)?)
+            }
         };
 
         Ok(connection)
@@ -151,6 +187,7 @@ impl AsyncRedisClient {
         let status = match self {
             Self::Cluster(pool) => pool.status(),
             Self::Single(pool) => pool.status(),
+            Self::Sentinel(pool) => pool.status(),
         };
 
         RedisClientStats {
@@ -172,6 +209,9 @@ impl AsyncRedisClient {
             Self::Single(pool) => {
                 pool.retain(|_, metrics| predicate(metrics));
             }
+            Self::Sentinel(pool) => {
+                pool.retain(|_, metrics| predicate(metrics));
+            }
         }
     }
 
@@ -210,6 +250,7 @@ impl std::fmt::Debug for AsyncRedisClient {
         match self {
             AsyncRedisClient::Cluster(_) => write!(f, "AsyncRedisPool::Cluster"),
             AsyncRedisClient::Single(_) => write!(f, "AsyncRedisPool::Single"),
+            AsyncRedisClient::Sentinel(_) => write!(f, "AsyncRedisPool::Sentinel"),
         }
     }
 }
@@ -225,13 +266,16 @@ pub enum AsyncRedisConnection {
     Cluster(pool::CustomClusterConnection),
     /// A connection to a single Redis instance.
     Single(pool::CustomSingleConnection),
+    /// A connection to a single Redis-master instance
+    Sentinel(pool::CustomSentinelConnection),
 }
 
 impl std::fmt::Debug for AsyncRedisConnection {
     fn fmt(&self, f: &mut std::fmt::Formatter<'_>) -> std::fmt::Result {
         let name = match self {
             Self::Cluster(_) => "Cluster",
             Self::Single(_) => "Single",
+            Self::Sentinel(_) => "Sentinel",
         };
         f.debug_tuple(name).finish()
     }
@@ -242,6 +286,7 @@ impl redis::aio::ConnectionLike for AsyncRedisConnection {
         match self {
             Self::Cluster(conn) => conn.req_packed_command(cmd),
             Self::Single(conn) => conn.req_packed_command(cmd),
+            Self::Sentinel(conn) => conn.req_packed_command(cmd),
         }
     }
 
@@ -254,13 +299,15 @@ impl redis::aio::ConnectionLike for AsyncRedisConnection {
         match self {
             Self::Cluster(conn) => conn.req_packed_commands(cmd, offset, count),
             Self::Single(conn) => conn.req_packed_commands(cmd, offset, count),
+            Self::Sentinel(conn) => conn.req_packed_commands(cmd, offset, count),
         }
     }
 
     fn get_db(&self) -> i64 {
         match self {
             Self::Cluster(conn) => conn.get_db(),
             Self::Single(conn) => conn.get_db(),
+            Self::Sentinel(conn) => conn.get_db(),
         }
     }
 }