feat: report db pool metrics

autoconnect/src/main.rs

@@ -4,7 +4,7 @@
 #[macro_use]
 extern crate slog_scope;
 
-use std::{env, vec::Vec};
+use std::{env, time::Duration, vec::Vec};
 
 use actix_http::HttpService;
 use actix_server::Server;
@@ -16,6 +16,7 @@ use serde::Deserialize;
 use autoconnect_settings::{AppState, Settings};
 use autoconnect_web::{build_app, config, config_router};
 use autopush_common::{
+    db::spawn_pool_periodic_reporter,
     errors::{ApcErrorKind, Result},
     logging,
 };
@@ -79,12 +80,17 @@ async fn main() -> Result<()> {
     let actix_workers = settings.actix_workers;
     let app_state = AppState::from_settings(settings)?;
     app_state.init_and_spawn_megaphone_updater().await?;
+    spawn_pool_periodic_reporter(
+        Duration::from_secs(10),
+        app_state.db.clone(),
+        app_state.metrics.clone(),
+    );
 
     info!(
-        "Starting autoconnect on port: {} router_port: {} (available_parallelism: {:?})",
+        "Starting autoconnect on port: {} router_port: {} ({})",
         port,
         router_port,
-        std::thread::available_parallelism()
+        logging::parallelism_banner()
     );
 
     let router_app_state = app_state.clone();

autoendpoint/src/main.rs

@@ -59,9 +59,9 @@ async fn main() -> Result<(), Box<dyn Error>> {
         .await
         .expect("Could not start server");
     info!(
-        "Starting autoendpoint on port: {} (available_parallelism: {:?})",
+        "Starting autoendpoint on port: {} ({})",
         host_port,
-        std::thread::available_parallelism()
+        logging::parallelism_banner()
     );
     server.await?;
 

autoendpoint/src/server.rs

@@ -7,19 +7,18 @@ use actix_cors::Cors;
 use actix_web::{
     dev, http::StatusCode, middleware::ErrorHandlers, web, web::Data, App, HttpServer,
 };
-#[cfg(feature = "bigtable")]
-use autopush_common::db::bigtable::BigTableClientImpl;
-#[cfg(feature = "dual")]
-use autopush_common::db::dual::DualClientImpl;
 use cadence::StatsdClient;
 use fernet::MultiFernet;
 use serde_json::json;
 
+#[cfg(feature = "bigtable")]
+use autopush_common::db::bigtable::BigTableClientImpl;
+#[cfg(feature = "dual")]
+use autopush_common::db::dual::DualClientImpl;
 #[cfg(feature = "dynamodb")]
 use autopush_common::db::dynamodb::DdbClientImpl;
-
 use autopush_common::{
-    db::{client::DbClient, DbSettings, StorageType},
+    db::{client::DbClient, spawn_pool_periodic_reporter, DbSettings, StorageType},
     middleware::sentry::SentryWrapper,
 };
 
@@ -130,6 +129,12 @@ impl Server {
             adm_router,
         };
 
+        spawn_pool_periodic_reporter(
+            Duration::from_secs(10),
+            app_state.db.clone(),
+            app_state.metrics.clone(),
+        );
+
         let server = HttpServer::new(move || {
             // These have a bad habit of being reset. Specify them explicitly.
             let cors = Cors::default()

autopush-common/Cargo.toml

@@ -59,6 +59,7 @@ async-trait = "0.1"
 deadpool = { version = "0.10", features = ["managed", "rt_tokio_1"] }
 gethostname = "0.4"
 futures-backoff = "0.1.0"
+num_cpus = "1.16"
 woothee = "0.13"
 
 # #[cfg(bigtable)] for this section.

autopush-common/src/db/bigtable/bigtable_client/mod.rs

@@ -1199,6 +1199,10 @@ impl DbClient for BigTableClientImpl {
     fn name(&self) -> String {
         "Bigtable".to_owned()
     }
+
+    fn pool_status(&self) -> Option<deadpool::Status> {
+        Some(self.pool.pool.status())
+    }
 }
 
 #[cfg(all(test, feature = "emulator"))]

autopush-common/src/db/client.rs

@@ -109,12 +109,17 @@ pub trait DbClient: Send + Sync {
     #[allow(clippy::needless_lifetimes)]
     fn rotating_message_table<'a>(&'a self) -> Option<&'a str>;
 
-    fn box_clone(&self) -> Box<dyn DbClient>;
-
     /// Provide the module name.
     /// This was added for simple dual mode testing, but may be useful in
     /// other situations.
     fn name(&self) -> String;
+
+    /// Return the current deadpool Status (if using deadpool)
+    fn pool_status(&self) -> Option<deadpool::Status> {
+        None
+    }
+
+    fn box_clone(&self) -> Box<dyn DbClient>;
 }
 
 impl Clone for Box<dyn DbClient> {

autopush-common/src/db/dual/mod.rs

@@ -459,6 +459,10 @@ impl DbClient for DualClientImpl {
     fn name(&self) -> String {
         "Dual".to_owned()
     }
+
+    fn pool_status(&self) -> Option<deadpool::Status> {
+        self.primary.pool_status()
+    }
 }
 
 #[cfg(all(test, feature = "bigtable", feature = "dynamodb"))]

autopush-common/src/db/mod.rs

@@ -31,12 +31,15 @@ pub mod dual;
 pub mod dynamodb;
 pub mod error;
 pub mod models;
+pub mod reporter;
 pub mod routing;
 mod util;
 
 // used by integration testing
 pub mod mock;
 
+pub use reporter::spawn_pool_periodic_reporter;
+
 use crate::errors::{ApcErrorKind, Result};
 use crate::notification::{Notification, STANDARD_NOTIFICATION_PREFIX, TOPIC_NOTIFICATION_PREFIX};
 use crate::util::timing::{ms_since_epoch, sec_since_epoch};

autopush-common/src/db/reporter.rs

@@ -0,0 +1,43 @@
+use std::{sync::Arc, time::Duration};
+
+use actix_web::rt;
+use cadence::{Gauged, StatsdClient};
+use gethostname::gethostname;
+
+use super::client::DbClient;
+
+/// Emit db pool (deadpool) metrics periodically
+pub fn spawn_pool_periodic_reporter(
+    interval: Duration,
+    db: Box<dyn DbClient>,
+    metrics: Arc<StatsdClient>,
+) {
+    let hostname = gethostname().to_string_lossy().to_string();
+    rt::spawn(async move {
+        loop {
+            pool_periodic_reporter(&*db, &metrics, &hostname);
+            rt::time::sleep(interval).await;
+        }
+    });
+}
+
+fn pool_periodic_reporter(db: &dyn DbClient, metrics: &StatsdClient, hostname: &str) {
+    let Some(status) = db.pool_status() else {
+        return;
+    };
+    metrics
+        .gauge_with_tags(
+            "database.pool.active",
+            (status.size - status.available) as u64,
+        )
+        .with_tag("hostname", hostname)
+        .send();
+    metrics
+        .gauge_with_tags("database.pool.idle", status.available as u64)
+        .with_tag("hostname", hostname)
+        .send();
+    metrics
+        .gauge_with_tags("database.pool.waiting", status.waiting as u64)
+        .with_tag("hostname", hostname)
+        .send();
+}

autopush-common/src/logging.rs

@@ -83,3 +83,13 @@ fn get_ec2_instance_id() -> reqwest::Result<String> {
         .error_for_status()?
         .text()
 }
+
+/// Return parallelism/number of CPU information to log at startup
+pub fn parallelism_banner() -> String {
+    format!(
+        "available_parallelism: {:?} num_cpus: {} num_cpus (phys): {}",
+        std::thread::available_parallelism(),
+        num_cpus::get(),
+        num_cpus::get_physical()
+    )
+}