feat(redis): Allow configuring Redis pools individually

relay-config/src/config.rs

@@ -16,15 +16,14 @@ use relay_kafka::{
 };
 use relay_metrics::aggregator::{AggregatorConfig, FlushBatching};
 use relay_metrics::MetricNamespace;
-use relay_redis::RedisConfigOptions;
 use serde::de::{DeserializeOwned, Unexpected, Visitor};
 use serde::{Deserialize, Deserializer, Serialize, Serializer};
 use uuid::Uuid;
 
 use crate::aggregator::{AggregatorServiceConfig, ScopedAggregatorConfig};
 use crate::byte_size::ByteSize;
 use crate::upstream::UpstreamDescriptor;
-use crate::{RedisConfig, RedisConnection};
+use crate::{RedisConfig, RedisConfigs, RedisPools};
 
 const DEFAULT_NETWORK_OUTAGE_GRACE_PERIOD: u64 = 10;
 
@@ -1024,7 +1023,7 @@ pub struct Processing {
     pub kafka_validate_topics: bool,
     /// Redis hosts to connect to for storing state for rate limits.
     #[serde(default)]
-    pub redis: Option<RedisConfig>,
+    pub redis: Option<RedisConfigs>,
     /// Maximum chunk size of attachments for Kafka.
     #[serde(default = "default_chunk_size")]
     pub attachment_chunk_size: ByteSize,
@@ -1577,8 +1576,9 @@ impl Config {
             }
         }
 
+        // TODO: Should this be generalized to being able to override individual redis pool settings?
         if let Some(redis) = overrides.redis_url {
-            processing.redis = Some(RedisConfig::single(redis))
+            processing.redis = Some(RedisConfigs::Unified(RedisConfig::single(redis)))
         }
 
         if let Some(kafka_url) = overrides.kafka_url {
@@ -2273,25 +2273,12 @@ impl Config {
     }
 
     /// Redis servers to connect to, for rate limiting.
-    pub fn redis(&self) -> Option<(&RedisConnection, RedisConfigOptions)> {
-        let cpu_concurrency = self.cpu_concurrency();
-
-        let redis = self.values.processing.redis.as_ref()?;
-
-        let options = RedisConfigOptions {
-            max_connections: redis
-                .options
-                .max_connections
-                .unwrap_or(cpu_concurrency as u32 * 2)
-                .min(crate::redis::DEFAULT_MIN_MAX_CONNECTIONS),
-            connection_timeout: redis.options.connection_timeout,
-            max_lifetime: redis.options.max_lifetime,
-            idle_timeout: redis.options.idle_timeout,
-            read_timeout: redis.options.read_timeout,
-            write_timeout: redis.options.write_timeout,
+    pub fn redis(&self) -> RedisPools {
+        let Some(redis_configs) = self.values.processing.redis.as_ref() else {
+            return RedisPools::default();
         };
 
-        Some((&redis.connection, options))
+        RedisPools::from_configs(redis_configs, self.cpu_concurrency())
     }
 
     /// Chunk size of attachments in bytes.