More client/server integration tests (#173)

* more client tests

* Count healthchecks, fix assignment overcounting

* remove diff

* diff

* let the os handle the port

Author: levkk

pgdog/src/admin/show_servers.rs

@@ -68,7 +68,7 @@ impl Command for ShowServers {
                 .add(server.stats.total.queries)
                 .add(server.stats.total.rollbacks)
                 .add(server.stats.total.prepared_statements)
-                .add(server.stats.healthchecks)
+                .add(server.stats.total.healthchecks)
                 .add(server.stats.total.errors)
                 .add(server.stats.total.bytes_received)
                 .add(server.stats.total.bytes_sent)

pgdog/src/backend/pool/inner.rs

@@ -222,7 +222,7 @@ impl Inner {
     /// Place connection back into the pool
     /// or give it to a waiting client.
     #[inline]
-    pub(super) fn put(&mut self, conn: Box<Server>) {
+    pub(super) fn put(&mut self, conn: Box<Server>, now: Instant) {
         // Try to give it to a client that's been waiting, if any.
         let id = *conn.id();
         if let Some(waiter) = self.waiting.pop_front() {
@@ -233,6 +233,8 @@ impl Inner {
                     server: id,
                     client: waiter.request.id,
                 });
+                self.stats.counts.server_assignment_count += 1;
+                self.stats.counts.wait_time += now.duration_since(waiter.request.created_at);
             }
         } else {
             self.conns.push(conn);
@@ -324,7 +326,7 @@ impl Inner {
         // Finally, if the server is ok,
         // place the connection back into the idle list.
         if server.can_check_in() {
-            self.put(server);
+            self.put(server, now);
         } else {
             self.out_of_sync += 1;
         }

pgdog/src/backend/pool/monitor.rs

@@ -242,7 +242,7 @@ impl Monitor {
                 let server = Box::new(conn);
 
                 let mut guard = self.pool.lock();
-                guard.put(server);
+                guard.put(server, Instant::now());
             }
 
             Ok(Err(err)) => {

pgdog/src/backend/pool/pool_impl.rs

@@ -273,6 +273,7 @@ impl Pool {
     pub(crate) fn move_conns_to(&self, destination: &Pool) {
         // Ensure no deadlock.
         assert!(self.inner.id != destination.id());
+        let now = Instant::now();
 
         {
             let mut from_guard = self.lock();
@@ -281,7 +282,7 @@ impl Pool {
             from_guard.online = false;
             let (idle, taken) = from_guard.move_conns_to(destination);
             for server in idle {
-                to_guard.put(server);
+                to_guard.put(server, now);
             }
             to_guard.set_taken(taken);
         }

pgdog/src/backend/pool/stats.rs

@@ -21,6 +21,7 @@ pub struct Counts {
     pub parse_count: usize,
     pub bind_count: usize,
     pub rollbacks: usize,
+    pub healthchecks: usize,
 }
 
 impl Sub for Counts {
@@ -41,6 +42,7 @@ impl Sub for Counts {
             parse_count: self.parse_count.saturating_sub(rhs.parse_count),
             bind_count: self.parse_count.saturating_sub(rhs.bind_count),
             rollbacks: self.rollbacks.saturating_sub(rhs.rollbacks),
+            healthchecks: self.healthchecks.saturating_add(rhs.healthchecks),
         }
     }
 }
@@ -61,6 +63,7 @@ impl Div<usize> for Counts {
             parse_count: self.parse_count.saturating_div(rhs),
             bind_count: self.parse_count.saturating_div(rhs),
             rollbacks: self.rollbacks.saturating_div(rhs),
+            healthchecks: self.healthchecks.saturating_div(rhs),
         }
     }
 }
@@ -72,7 +75,7 @@ impl Add<BackendCounts> for Counts {
         Counts {
             xact_count: self.xact_count + rhs.transactions,
             query_count: self.query_count + rhs.queries,
-            server_assignment_count: self.server_assignment_count + 1,
+            server_assignment_count: self.server_assignment_count,
             received: self.received + rhs.bytes_received,
             sent: self.sent + rhs.bytes_sent,
             query_time: self.query_time + rhs.query_time,
@@ -81,6 +84,7 @@ impl Add<BackendCounts> for Counts {
             parse_count: self.parse_count + rhs.parse,
             bind_count: self.parse_count + rhs.bind,
             rollbacks: self.rollbacks + rhs.rollbacks,
+            healthchecks: self.healthchecks + rhs.healthchecks,
         }
     }
 }
@@ -114,6 +118,7 @@ impl Add for Counts {
             parse_count: self.parse_count.saturating_add(rhs.parse_count),
             bind_count: self.parse_count.saturating_add(rhs.bind_count),
             rollbacks: self.rollbacks.saturating_add(rhs.rollbacks),
+            healthchecks: self.healthchecks.saturating_add(rhs.healthchecks),
         }
     }
 }

pgdog/src/backend/stats.rs

@@ -60,6 +60,7 @@ pub struct Counts {
     pub transaction_time: Duration,
     pub parse: usize,
     pub bind: usize,
+    pub healthchecks: usize,
 }
 
 impl Add for Counts {
@@ -80,6 +81,7 @@ impl Add for Counts {
             transaction_time: self.query_time.saturating_add(rhs.transaction_time),
             parse: self.parse.saturating_add(rhs.parse),
             bind: self.bind.saturating_add(rhs.bind),
+            healthchecks: self.healthchecks.saturating_add(rhs.healthchecks),
         }
     }
 }
@@ -88,8 +90,6 @@ impl Add for Counts {
 #[derive(Copy, Clone, Debug)]
 pub struct Stats {
     pub id: BackendKeyData,
-    /// Number of bytes sent.
-    pub healthchecks: usize,
     pub state: State,
     pub last_used: Instant,
     pub last_healthcheck: Option<Instant>,
@@ -107,7 +107,6 @@ impl Stats {
         let now = Instant::now();
         let stats = Stats {
             id,
-            healthchecks: 0,
             state: State::Idle,
             last_used: now,
             last_healthcheck: None,
@@ -244,7 +243,8 @@ impl Stats {
 
     /// Track healtchecks.
     pub fn healthcheck(&mut self) {
-        self.healthchecks += 1;
+        self.total.healthchecks += 1;
+        self.last_checkout.healthchecks += 1;
         self.last_healthcheck = Some(Instant::now());
         self.update();
     }

pgdog/src/config/mod.rs

@@ -964,6 +964,37 @@ pub mod test {
         init();
     }
 
+    pub fn load_test_replicas() {
+        let mut config = ConfigAndUsers::default();
+        config.config.databases = vec![
+            Database {
+                name: "pgdog".into(),
+                host: "127.0.0.1".into(),
+                port: 5432,
+                role: Role::Primary,
+                ..Default::default()
+            },
+            Database {
+                name: "pgdog".into(),
+                host: "127.0.0.1".into(),
+                port: 5432,
+                role: Role::Replica,
+                read_only: Some(true),
+                ..Default::default()
+            },
+        ];
+        config.config.general.load_balancing_strategy = LoadBalancingStrategy::RoundRobin;
+        config.users.users = vec![User {
+            name: "pgdog".into(),
+            database: "pgdog".into(),
+            password: Some("pgdog".into()),
+            ..Default::default()
+        }];
+
+        set(config).unwrap();
+        init();
+    }
+
     #[test]
     fn test_basic() {
         let source = r#"