Use deadpool fork with tracing (#5276)

* Use deadpool fork with tracing

* Implement minimum Redis connections

* fix typos maybe

* address pr comments

apps/labrinth/src/database/redis/mod.rs

@@ -3,6 +3,7 @@ use ariadne::ids::base62_impl::{parse_base62, to_base62};
 use chrono::{TimeZone, Utc};
 use dashmap::DashMap;
 use deadpool_redis::{Config, Runtime};
+use futures::TryStreamExt;
 use futures::future::Either;
 use futures::stream::{FuturesUnordered, StreamExt};
 use prometheus::{IntGauge, Registry};
@@ -14,7 +15,7 @@ use std::fmt::{Debug, Display};
 use std::future::Future;
 use std::hash::Hash;
 use std::time::Duration;
-use tracing::{Instrument, info_span};
+use tracing::{Instrument, info, info_span};
 use util::{cmd, redis_pipe};
 
 pub mod util;
@@ -25,7 +26,7 @@ const ACTUAL_EXPIRY: i64 = 60 * 30; // 30 minutes
 #[derive(Clone)]
 pub struct RedisPool {
     pub url: String,
-    pub pool: util::InstrumentedPool,
+    pub pool: deadpool_redis::Pool,
     cache_list: DashMap<String, util::CacheSubscriber>,
     meta_namespace: String,
 }
@@ -69,11 +70,35 @@ impl RedisPool {
 
         let pool = RedisPool {
             url,
-            pool: util::InstrumentedPool::new(pool),
+            pool,
             cache_list: DashMap::with_capacity(2048),
             meta_namespace: meta_namespace.unwrap_or("".to_string()),
         };
 
+        let redis_min_connections = dotenvy::var("REDIS_MIN_CONNECTIONS")
+            .ok()
+            .and_then(|x| x.parse::<usize>().ok())
+            .unwrap_or(0);
+        let spawn_min_connections = (0..redis_min_connections)
+            .map(|_| {
+                let pool = pool.clone();
+                tokio::spawn(async move { pool.pool.get().await })
+            })
+            .collect::<FuturesUnordered<_>>();
+        tokio::spawn({
+            let pool = pool.clone();
+            async move {
+                // collect the connections into a buffer while we're spawning them,
+                // to make sure that we're not `get`ing any connections we previously took
+                let _connections =
+                    spawn_min_connections.try_collect::<Vec<_>>().await;
+                info!(
+                    pool_status = ?pool.pool.status(),
+                    "Finished getting {redis_min_connections} initial Redis connections"
+                );
+            }
+        });
+
         let interval = Duration::from_secs(30);
         let max_age = Duration::from_secs(5 * 60); // 5 minutes
         let pool_ref = pool.clone();

apps/labrinth/src/database/redis/util.rs

@@ -2,7 +2,6 @@ use std::fmt::Debug;
 use std::sync::Arc;
 use std::sync::atomic::{AtomicBool, Ordering};
 
-use deadpool_redis::PoolError;
 use derive_more::{Deref, DerefMut};
 use redis::{FromRedisValue, RedisResult, ToRedisArgs};
 use tokio::sync::Notify;
@@ -11,24 +10,6 @@ use tracing::{Instrument, info_span};
 
 use crate::database::models::DatabaseError;
 
-#[derive(Debug, Clone, Deref, DerefMut)]
-pub struct InstrumentedPool {
-    inner: deadpool_redis::Pool,
-}
-
-impl InstrumentedPool {
-    pub fn new(inner: deadpool_redis::Pool) -> Self {
-        Self { inner }
-    }
-
-    pub async fn get(&self) -> Result<deadpool_redis::Connection, PoolError> {
-        self.inner
-            .get()
-            .instrument(info_span!("get redis connection"))
-            .await
-    }
-}
-
 pub fn redis_pipe() -> InstrumentedPipeline {
     InstrumentedPipeline {
         inner: redis::pipe(),
@@ -55,7 +36,7 @@ impl InstrumentedPipeline {
     ) -> RedisResult<T> {
         self.inner
             .query_async(con)
-            .instrument(info_span!("execute redis pipeline"))
+            .instrument(info_span!("pipeline.query_async"))
             .await
     }
 }
@@ -88,7 +69,7 @@ impl InstrumentedCmd {
         con: &mut impl redis::aio::ConnectionLike,
     ) -> RedisResult<T> {
         let span = info_span!(
-            "query_async",
+            "cmd.query_async",
             // <https://opentelemetry.io/docs/specs/semconv/db/redis/>
             db.system.name = "redis",
             db.operation.name = self.name,