feat(redis): Make connection pool configurable [INGEST-1557]

CHANGELOG.md

@@ -6,6 +6,10 @@
 
 - Refactor tokio-based Addr from healthcheck to be generic. ([#1405](https://github.com/relay/pull/1405))
 
+**Features**:
+
+- Make Redis connection pool configurable. ([#1413](https://github.com/getsentry/relay/pull/1413))
+
 ## 22.8.0
 
 **Features**:

relay-quotas/src/redis.rs

@@ -249,7 +249,7 @@ mod tests {
     use std::time::{SystemTime, UNIX_EPOCH};
 
     use relay_common::{ProjectId, ProjectKey};
-    use relay_redis::redis::Commands;
+    use relay_redis::{redis::Commands, RedisConfigOptions};
 
     use crate::quota::{DataCategories, DataCategory, ReasonCode, Scoping};
     use crate::rate_limit::RateLimitScope;
@@ -261,7 +261,7 @@ mod tests {
             .unwrap_or_else(|_| "redis://127.0.0.1:6379".to_owned());
 
         RedisRateLimiter {
-            pool: RedisPool::single(&url).unwrap(),
+            pool: RedisPool::single(&url, &RedisConfigOptions::default()).unwrap(),
             script: Arc::new(load_lua_script()),
             max_limit: None,
         }

relay-redis/Cargo.toml

@@ -18,3 +18,6 @@ serde = { version = "1.0.114", features = ["derive"] }
 [features]
 default = []
 impl = ["r2d2", "redis"]
+
+[dev-dependencies]
+serde_yaml = "0.8.13"

relay-redis/src/config.rs

@@ -1,5 +1,25 @@
 use serde::{Deserialize, Serialize};
 
+const fn default_max_connections() -> u32 {
+    24
+}
+
+/// Additional configuration options for a redis client.
+#[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
+pub struct RedisConfigOptions {
+    /// Maximum number of connections managed by the pool.
+    #[serde(default = "default_max_connections")]
+    pub max_connections: u32,
+}
+
+impl Default for RedisConfigOptions {
+    fn default() -> Self {
+        Self {
+            max_connections: default_max_connections(),
+        }
+    }
+}
+
 /// Configuration for connecting a redis client.
 #[derive(Clone, Debug, Serialize, Deserialize, Eq, PartialEq)]
 #[serde(untagged)]
@@ -10,10 +30,87 @@ pub enum RedisConfig {
         ///
         /// This can also be a single node which is configured in cluster mode.
         cluster_nodes: Vec<String>,
+
+        /// Additional configuration options for the redis client and a connections pool.
+        #[serde(flatten)]
+        options: RedisConfigOptions,
     },
 
     /// Connect to a single Redis instance.
     ///
     /// Contains the `redis://` url to the node.
     Single(String),
+
+    /// Connect to a single Redis instance.
+    ///
+    /// Allows to provide more configuration options, e.g. `max_connections`.
+    SingleWithOpts {
+        /// Containes the `redis://` url to the node.
+        server: String,
+
+        /// Additional configuration options for the redis client and a connections pool.
+        #[serde(flatten)]
+        options: RedisConfigOptions,
+    },
+}
+
+#[cfg(test)]
+mod tests {
+    use super::*;
+
+    #[test]
+    fn test_redis_single_opts() {
+        let yaml = r###"
+server: "redis://127.0.0.1:6379"
+max_connections: 42
+"###;
+
+        let config: RedisConfig = serde_yaml::from_str(yaml)
+            .expect("Parsed processing redis config: single with options");
+
+        match config {
+            RedisConfig::SingleWithOpts { server, options } => {
+                assert_eq!(options.max_connections, 42);
+                assert_eq!(server, "redis://127.0.0.1:6379");
+            }
+            e => panic!("Expected RedisConfig::SingleWithOpts but got {:?}", e),
+        }
+    }
+
+    #[test]
+    fn test_redis_single_opts_default() {
+        let yaml = r###"
+server: "redis://127.0.0.1:6379"
+"###;
+
+        let config: RedisConfig = serde_yaml::from_str(yaml)
+            .expect("Parsed processing redis config: single with options");
+
+        match config {
+            RedisConfig::SingleWithOpts { options, .. } => {
+                // check if all the defaults are correctly set
+                assert_eq!(options.max_connections, 24);
+            }
+            e => panic!("Expected RedisConfig::SingleWithOpts but got {:?}", e),
+        }
+    }
+
+    // To make sure that we have backwards compatibility and still support the redis configuration
+    // when the single `redis://...` address is provided.
+    #[test]
+    fn test_redis_single() {
+        let yaml = r###"
+"redis://127.0.0.1:6379"
+"###;
+
+        let config: RedisConfig = serde_yaml::from_str(yaml)
+            .expect("Parsed processing redis config: single with options");
+
+        match config {
+            RedisConfig::Single(server) => {
+                assert_eq!(server, "redis://127.0.0.1:6379");
+            }
+            e => panic!("Expected RedisConfig::Single but got {:?}", e),
+        }
+    }
 }

relay-redis/src/real.rs

@@ -2,7 +2,7 @@ use failure::Fail;
 use r2d2::{Pool, PooledConnection};
 use redis::ConnectionLike;
 
-use crate::config::RedisConfig;
+use crate::config::{RedisConfig, RedisConfigOptions};
 
 pub use redis;
 
@@ -115,18 +115,25 @@ impl RedisPool {
     /// Creates a `RedisPool` from configuration.
     pub fn new(config: &RedisConfig) -> Result<Self, RedisError> {
         match config {
-            RedisConfig::Cluster { ref cluster_nodes } => {
+            RedisConfig::Cluster {
+                ref cluster_nodes,
+                ref options,
+            } => {
                 let servers = cluster_nodes.iter().map(String::as_str).collect();
-                Self::cluster(servers)
+                Self::cluster(servers, options)
             }
-            RedisConfig::Single(ref server) => Self::single(server),
+            RedisConfig::Single(ref server) => Self::single(server, &RedisConfigOptions::default()),
+            RedisConfig::SingleWithOpts {
+                ref server,
+                ref options,
+            } => Self::single(server, options),
         }
     }
 
     /// Creates a `RedisPool` in cluster configuration.
-    pub fn cluster(servers: Vec<&str>) -> Result<Self, RedisError> {
+    pub fn cluster(servers: Vec<&str>, opts: &RedisConfigOptions) -> Result<Self, RedisError> {
         let pool = Pool::builder()
-            .max_size(24)
+            .max_size(opts.max_connections)
             .test_on_check_out(false)
             .build(redis::cluster::ClusterClient::open(servers).map_err(RedisError::Redis)?)
             .map_err(RedisError::Pool)?;
@@ -136,9 +143,9 @@ impl RedisPool {
     }
 
     /// Creates a `RedisPool` in single-node configuration.
-    pub fn single(server: &str) -> Result<Self, RedisError> {
+    pub fn single(server: &str, opts: &RedisConfigOptions) -> Result<Self, RedisError> {
         let pool = Pool::builder()
-            .max_size(24)
+            .max_size(opts.max_connections)
             .test_on_check_out(false)
             .build(redis::Client::open(server).map_err(RedisError::Redis)?)
             .map_err(RedisError::Pool)?;