Adds new statistics for tracking connections created and closed

bb8/src/api.rs

@@ -99,6 +99,20 @@ pub struct Statistics {
     pub get_waited: u64,
     /// Total gets performed that timed out while waiting for a connection.
     pub get_timed_out: u64,
+    /// Total time accumulated waiting for a connection.
+    pub get_waited_time_micro: u64,
+    /// Total connections created.
+    pub connections_created: u64,
+    /// Total connections that were closed due to be in broken state.
+    pub connections_broken_closed: u64,
+    /// Total connections that were closed due to be considered invalid.
+    pub connections_invalid_closed: u64,
+    /// Total connections that were closed because they reached the max
+    /// lifetime.
+    pub connections_max_lifetime_closed: u64,
+    /// Total connections that were closed because they reached the max
+    /// idle timeout.
+    pub connections_max_idle_timeout_closed: u64,
 }
 
 /// A builder for a connection pool.

bb8/src/inner.rs

@@ -1,6 +1,7 @@
 use std::cmp::{max, min};
 use std::fmt;
 use std::future::Future;
+use std::sync::atomic::Ordering;
 use std::sync::{Arc, Weak};
 use std::time::{Duration, Instant};
 
@@ -86,6 +87,7 @@ where
 
     pub(crate) async fn get(&self) -> Result<PooledConnection<'_, M>, RunError<M::Error>> {
         let mut kind = StatsKind::Direct;
+        let mut wait_time_start = None;
 
         let future = async {
             loop {
@@ -98,6 +100,7 @@ where
                 let mut conn = match conn {
                     Some(conn) => PooledConnection::new(self, conn),
                     None => {
+                        wait_time_start = Some(Instant::now());
                         kind = StatsKind::Waited;
                         self.inner.notify.notified().await;
                         continue;
@@ -111,6 +114,10 @@ where
                 match self.inner.manager.is_valid(&mut conn).await {
                     Ok(()) => return Ok(conn),
                     Err(e) => {
+                        self.inner
+                            .statistics
+                            .connections_invalid_closed
+                            .fetch_add(1, Ordering::SeqCst);
                         self.inner.forward_error(e);
                         conn.state = ConnectionState::Invalid;
                         continue;
@@ -127,7 +134,16 @@ where
             }
         };
 
-        self.inner.statistics.record(kind);
+        self.inner.statistics.record_get(kind);
+
+        if let Some(wait_time_start) = wait_time_start {
+            let wait_time = Instant::now() - wait_time_start;
+            self.inner
+                .statistics
+                .get_waited_time_micro
+                .fetch_add(wait_time.as_micros() as u64, Ordering::SeqCst);
+        }
+
         result
     }
 
@@ -147,7 +163,13 @@ where
         let mut locked = self.inner.internals.lock();
         match (state, self.inner.manager.has_broken(&mut conn.conn)) {
             (ConnectionState::Present, false) => locked.put(conn, None, self.inner.clone()),
-            (_, _) => {
+            (_, is_broken) => {
+                if is_broken {
+                    self.inner
+                        .statistics
+                        .connections_broken_closed
+                        .fetch_add(1, Ordering::SeqCst);
+                }
                 let approvals = locked.dropped(1, &self.inner.statics);
                 self.spawn_replenishing_approvals(approvals);
                 self.inner.notify.notify_waiters();
@@ -190,6 +212,10 @@ where
                         .internals
                         .lock()
                         .put(conn, Some(approval), self.inner.clone());
+                    self.inner
+                        .statistics
+                        .connections_created
+                        .fetch_add(1, Ordering::SeqCst);
                     return Ok(());
                 }
                 Err(e) => {

bb8/src/internals.rs

@@ -49,7 +49,15 @@ where
 
     pub(crate) fn reap(&self) -> ApprovalIter {
         let mut locked = self.internals.lock();
-        locked.reap(&self.statics)
+        let (iter, max_idle_timeout_closed, max_lifetime_closed) = locked.reap(&self.statics);
+        drop(locked);
+        self.statistics
+            .connections_max_idle_timeout_closed
+            .fetch_add(max_idle_timeout_closed, Ordering::SeqCst);
+        self.statistics
+            .connections_max_lifetime_closed
+            .fetch_add(max_lifetime_closed, Ordering::SeqCst);
+        iter
     }
 
     pub(crate) fn forward_error(&self, err: M::Error) {
@@ -139,22 +147,34 @@ where
         ApprovalIter { num: num as usize }
     }
 
-    pub(crate) fn reap(&mut self, config: &Builder<M>) -> ApprovalIter {
+    pub(crate) fn reap(&mut self, config: &Builder<M>) -> (ApprovalIter, u64, u64) {
+        let mut max_lifetime_closed: u64 = 0;
+        let mut max_idle_timeout_closed: u64 = 0;
         let now = Instant::now();
         let before = self.conns.len();
 
         self.conns.retain(|conn| {
             let mut keep = true;
             if let Some(timeout) = config.idle_timeout {
-                keep &= now - conn.idle_start < timeout;
+                if now - conn.idle_start >= timeout {
+                    max_idle_timeout_closed += 1;
+                    keep &= false;
+                }
             }
             if let Some(lifetime) = config.max_lifetime {
-                keep &= now - conn.conn.birth < lifetime;
+                if now - conn.conn.birth >= lifetime {
+                    max_lifetime_closed += 1;
+                    keep &= false;
+                }
             }
             keep
         });
 
-        self.dropped((before - self.conns.len()) as u32, config)
+        (
+            self.dropped((before - self.conns.len()) as u32, config),
+            max_idle_timeout_closed,
+            max_lifetime_closed,
+        )
     }
 
     pub(crate) fn state(&self, statistics: Statistics) -> State {
@@ -255,10 +275,16 @@ pub(crate) struct AtomicStatistics {
     pub(crate) get_direct: AtomicU64,
     pub(crate) get_waited: AtomicU64,
     pub(crate) get_timed_out: AtomicU64,
+    pub(crate) get_waited_time_micro: AtomicU64,
+    pub(crate) connections_created: AtomicU64,
+    pub(crate) connections_broken_closed: AtomicU64,
+    pub(crate) connections_invalid_closed: AtomicU64,
+    pub(crate) connections_max_lifetime_closed: AtomicU64,
+    pub(crate) connections_max_idle_timeout_closed: AtomicU64,
 }
 
 impl AtomicStatistics {
-    pub(crate) fn record(&self, kind: StatsKind) {
+    pub(crate) fn record_get(&self, kind: StatsKind) {
         match kind {
             StatsKind::Direct => self.get_direct.fetch_add(1, Ordering::SeqCst),
             StatsKind::Waited => self.get_waited.fetch_add(1, Ordering::SeqCst),
@@ -273,6 +299,16 @@ impl From<&AtomicStatistics> for Statistics {
             get_direct: item.get_direct.load(Ordering::SeqCst),
             get_waited: item.get_waited.load(Ordering::SeqCst),
             get_timed_out: item.get_timed_out.load(Ordering::SeqCst),
+            get_waited_time_micro: item.get_waited_time_micro.load(Ordering::SeqCst),
+            connections_created: item.connections_created.load(Ordering::SeqCst),
+            connections_broken_closed: item.connections_broken_closed.load(Ordering::SeqCst),
+            connections_invalid_closed: item.connections_invalid_closed.load(Ordering::SeqCst),
+            connections_max_lifetime_closed: item
+                .connections_max_lifetime_closed
+                .load(Ordering::SeqCst),
+            connections_max_idle_timeout_closed: item
+                .connections_max_idle_timeout_closed
+                .load(Ordering::SeqCst),
         }
     }
 }

bb8/tests/test.rs

@@ -246,6 +246,8 @@ async fn test_drop_on_broken() {
     }
 
     assert!(DROPPED.load(Ordering::SeqCst));
+
+    assert_eq!(pool.state().statistics.connections_broken_closed, 1);
 }
 
 #[tokio::test]
@@ -453,6 +455,71 @@ async fn test_now_invalid() {
     // Now try to get a new connection.
     let r = pool.get().await;
     assert!(r.is_err());
+
+    // both connections in the pool were considered invalid
+    assert_eq!(pool.state().statistics.connections_invalid_closed, 2);
+}
+
+#[tokio::test]
+async fn test_idle_timeout() {
+    static DROPPED: AtomicUsize = AtomicUsize::new(0);
+
+    #[derive(Default)]
+    struct Connection;
+    impl Drop for Connection {
+        fn drop(&mut self) {
+            DROPPED.fetch_add(1, Ordering::SeqCst);
+        }
+    }
+
+    let manager = NthConnectionFailManager::<Connection>::new(5);
+    let pool = Pool::builder()
+        .idle_timeout(Some(Duration::from_secs(1)))
+        .connection_timeout(Duration::from_secs(1))
+        .reaper_rate(Duration::from_secs(1))
+        .max_size(5)
+        .min_idle(Some(5))
+        .build(manager)
+        .await
+        .unwrap();
+
+    let (tx1, rx1) = oneshot::channel();
+    let (tx2, rx2) = oneshot::channel();
+    let clone = pool.clone();
+    tokio::spawn(async move {
+        let conn = clone.get().await.unwrap();
+        tx1.send(()).unwrap();
+        // NB: If we sleep here we'll block this thread's event loop, and the
+        // reaper can't run.
+        let _ = rx2
+            .map(|r| match r {
+                Ok(v) => Ok((v, conn)),
+                Err(_) => Err((Error, conn)),
+            })
+            .await;
+    });
+
+    rx1.await.unwrap();
+
+    // And wait.
+    assert!(timeout(Duration::from_secs(2), pending::<()>())
+        .await
+        .is_err());
+    assert_eq!(DROPPED.load(Ordering::SeqCst), 4);
+
+    tx2.send(()).unwrap();
+
+    // And wait some more.
+    assert!(timeout(Duration::from_secs(3), pending::<()>())
+        .await
+        .is_err());
+    assert_eq!(DROPPED.load(Ordering::SeqCst), 5);
+
+    // all 5 idle connections were closed due to max idle time
+    assert_eq!(
+        pool.state().statistics.connections_max_idle_timeout_closed,
+        5
+    );
 }
 
 #[tokio::test]
@@ -509,6 +576,9 @@ async fn test_max_lifetime() {
         .await
         .is_err());
     assert_eq!(DROPPED.load(Ordering::SeqCst), 5);
+
+    // all 5 connections were closed due to max lifetime
+    assert_eq!(pool.state().statistics.connections_max_lifetime_closed, 5);
 }
 
 #[tokio::test]
@@ -924,4 +994,25 @@ async fn test_state_get_contention() {
     let statistics = pool.state().statistics;
     assert_eq!(statistics.get_direct, 1);
     assert_eq!(statistics.get_waited, 1);
+    assert!(statistics.get_waited_time_micro > 0);
+}
+
+#[tokio::test]
+async fn test_statistics_connections_created() {
+    let pool = Pool::builder()
+        .max_size(1)
+        .min_idle(1)
+        .build(OkManager::<FakeConnection>::new())
+        .await
+        .unwrap();
+    let (tx1, rx1) = oneshot::channel();
+    let clone = pool.clone();
+    tokio::spawn(async move {
+        let _ = clone.get().await.unwrap();
+        tx1.send(()).unwrap();
+    });
+    // wait until finished.
+    rx1.await.unwrap();
+
+    assert_eq!(pool.state().statistics.connections_created, 1);
 }