feat: emit Db pool metrics periodically (#605)

kill the mislabeled pool get metric for now (fixes it repeatedly
emitting)

Closes #555, #406

(cherry picked from commit c3d6946)

src/db/mock.rs

@@ -18,6 +18,10 @@ impl DbPool for MockDbPool {
         Box::pin(future::ok(Box::new(MockDb::new()) as Box<dyn Db>))
     }
 
+    fn state(&self) -> results::PoolState {
+        results::PoolState::default()
+    }
+
     fn box_clone(&self) -> Box<dyn DbPool> {
         Box::new(self.clone())
     }

src/db/mod.rs

@@ -10,8 +10,9 @@ pub mod spanner;
 mod tests;
 pub mod util;
 
-use std::fmt::Debug;
+use std::{fmt::Debug, time::Duration};
 
+use cadence::{Gauged, StatsdClient};
 use futures::future::{self, LocalBoxFuture, TryFutureExt};
 use lazy_static::lazy_static;
 use serde::Deserialize;
@@ -61,6 +62,9 @@ type DbFuture<T> = LocalBoxFuture<'static, Result<T, ApiError>>;
 
 pub trait DbPool: Sync + Send + Debug {
     fn get(&self) -> DbFuture<Box<dyn Db>>;
+
+    fn state(&self) -> results::PoolState;
+
     fn box_clone(&self) -> Box<dyn DbPool>;
 }
 
@@ -259,3 +263,29 @@ pub fn pool_from_settings(
         _ => Err(DbErrorKind::InvalidUrl(settings.database_url.to_owned()))?,
     })
 }
+
+/// Emit DbPool metrics periodically
+pub fn spawn_pool_periodic_reporter(
+    interval: Duration,
+    metrics: StatsdClient,
+    pool: Box<dyn DbPool>,
+) {
+    actix_rt::spawn(async move {
+        loop {
+            let results::PoolState {
+                connections,
+                idle_connections,
+            } = pool.state();
+            metrics
+                .gauge(
+                    "storage.pool.connections.active",
+                    (connections - idle_connections) as u64,
+                )
+                .ok();
+            metrics
+                .gauge("storage.pool.connections.idle", idle_connections as u64)
+                .ok();
+            actix_rt::time::delay_for(interval).await;
+        }
+    });
+}

src/db/mysql/pool.rs

@@ -17,7 +17,7 @@ use diesel::{
 use super::models::{MysqlDb, Result};
 #[cfg(test)]
 use super::test::TestTransactionCustomizer;
-use crate::db::{error::DbError, Db, DbFuture, DbPool, STD_COLLS};
+use crate::db::{error::DbError, results, Db, DbFuture, DbPool, STD_COLLS};
 use crate::server::metrics::Metrics;
 use crate::settings::Settings;
 
@@ -92,6 +92,10 @@ impl DbPool for MysqlDbPool {
         )
     }
 
+    fn state(&self) -> results::PoolState {
+        self.pool.state().into()
+    }
+
     fn box_clone(&self) -> Box<dyn DbPool> {
         Box::new(self.clone())
     }

src/db/results.rs

@@ -69,6 +69,22 @@ pub struct PostBsos {
     pub failed: HashMap<String, String>,
 }
 
+#[derive(Debug, Default)]
+/// A mockable r2d2::State
+pub struct PoolState {
+    pub connections: u32,
+    pub idle_connections: u32,
+}
+
+impl From<diesel::r2d2::State> for PoolState {
+    fn from(state: diesel::r2d2::State) -> PoolState {
+        PoolState {
+            connections: state.connections,
+            idle_connections: state.idle_connections,
+        }
+    }
+}
+
 #[cfg(test)]
 pub type GetCollectionId = i32;
 

src/db/spanner/pool.rs

@@ -15,7 +15,7 @@ use scheduled_thread_pool::ScheduledThreadPool;
 use super::models::Result;
 #[cfg(test)]
 use super::test_util::SpannerTestTransactionCustomizer;
-use crate::db::{error::DbError, Db, DbFuture, DbPool, STD_COLLS};
+use crate::db::{error::DbError, results, Db, DbFuture, DbPool, STD_COLLS};
 use crate::server::metrics::Metrics;
 use crate::settings::Settings;
 
@@ -63,8 +63,6 @@ impl SpannerDbPool {
         let builder = r2d2::Pool::builder()
             .max_size(max_size)
             .thread_pool(Arc::new(ScheduledThreadPool::new(r2d2_thread_pool_size)));
-        let mut metrics = metrics.clone();
-        metrics.start_timer("storage.spanner.pool.get", None);
 
         #[cfg(test)]
         let builder = if settings.database_use_test_transactions {
@@ -76,7 +74,7 @@ impl SpannerDbPool {
         Ok(Self {
             pool: builder.build(manager)?,
             coll_cache: Default::default(),
-            metrics,
+            metrics: metrics.clone(),
         })
     }
 
@@ -102,6 +100,10 @@ impl DbPool for SpannerDbPool {
         )
     }
 
+    fn state(&self) -> results::PoolState {
+        self.pool.state().into()
+    }
+
     fn box_clone(&self) -> Box<dyn DbPool> {
         Box::new(self.clone())
     }

src/server/mod.rs

@@ -1,8 +1,8 @@
 //! Main application server
 
-use std::sync::Arc;
+use std::{sync::Arc, time::Duration};
 
-use crate::db::{pool_from_settings, DbPool};
+use crate::db::{pool_from_settings, spawn_pool_periodic_reporter, DbPool};
 use crate::error::ApiError;
 use crate::server::metrics::Metrics;
 use crate::settings::{Secrets, ServerLimits, Settings};
@@ -158,6 +158,8 @@ impl Server {
         let secrets = Arc::new(settings.master_secret);
         let port = settings.port;
 
+        spawn_pool_periodic_reporter(Duration::from_secs(10), metrics.clone(), db_pool.clone());
+
         let server = HttpServer::new(move || {
             // Setup the server state
             let state = ServerState {