RUST-1707 Add durations to connection pool events (#976)

abr-egn · web-flow · commit 135c9f0a4527 · 2023-10-17T14:53:06.000-04:00
diff --git a/src/cmap.rs b/src/cmap.rs
@@ -9,6 +9,8 @@ pub(crate) mod options;
 mod status;
 mod worker;
 
+use std::time::Instant;
+
 use derivative::Derivative;
 
 pub use self::conn::ConnectionInfo;
@@ -119,6 +121,7 @@ impl ConnectionPool {
     /// front of the wait queue, and then will block again if no available connections are in the
     /// pool and the total number of connections is not less than the max pool size.
     pub(crate) async fn check_out(&self) -> Result<Connection> {
+        let time_started = Instant::now();
         self.event_emitter.emit_event(|| {
             ConnectionCheckoutStartedEvent {
                 address: self.address.clone(),
@@ -142,25 +145,17 @@ impl ConnectionPool {
         match conn {
             Ok(ref conn) => {
                 self.event_emitter
-                    .emit_event(|| conn.checked_out_event().into());
-            }
-            #[cfg(feature = "tracing-unstable")]
-            Err(ref err) => {
-                self.event_emitter.emit_event(|| {
-                    ConnectionCheckoutFailedEvent {
-                        address: self.address.clone(),
-                        reason: ConnectionCheckoutFailedReason::ConnectionError,
-                        error: Some(err.clone()),
-                    }
-                    .into()
-                });
+                    .emit_event(|| conn.checked_out_event(time_started).into());
             }
-            #[cfg(not(feature = "tracing-unstable"))]
-            Err(_) => {
+
+            Err(ref _err) => {
                 self.event_emitter.emit_event(|| {
                     ConnectionCheckoutFailedEvent {
                         address: self.address.clone(),
                         reason: ConnectionCheckoutFailedReason::ConnectionError,
+                        #[cfg(feature = "tracing-unstable")]
+                        error: Some(_err.clone()),
+                        duration: Instant::now() - time_started,
                     }
                     .into()
                 });
diff --git a/src/cmap/conn.rs b/src/cmap/conn.rs
@@ -74,6 +74,8 @@ pub(crate) struct Connection {
 
     pub(crate) generation: ConnectionGeneration,
 
+    pub(crate) time_created: Instant,
+
     /// The cached StreamDescription from the connection's handshake.
     pub(super) stream_description: Option<StreamDescription>,
 
@@ -127,11 +129,13 @@ impl Connection {
         stream: AsyncStream,
         id: u32,
         generation: ConnectionGeneration,
+        time_created: Instant,
     ) -> Self {
         Self {
             id,
             server_id: None,
             generation,
+            time_created,
             pool_manager: None,
             command_executing: false,
             ready_and_available_time: None,
@@ -159,6 +163,7 @@ impl Connection {
             stream,
             pending_connection.id,
             generation,
+            pending_connection.time_created,
         );
         conn.event_emitter = Some(pending_connection.event_emitter);
         conn
@@ -167,7 +172,13 @@ impl Connection {
     /// Create a connection intended for monitoring purposes.
     /// TODO: RUST-1454 Rename this to just `new`, drop the pooling-specific data.
     pub(crate) fn new_monitoring(address: ServerAddress, stream: AsyncStream, id: u32) -> Self {
-        Self::new(address, stream, id, ConnectionGeneration::Monitoring)
+        Self::new(
+            address,
+            stream,
+            id,
+            ConnectionGeneration::Monitoring,
+            Instant::now(),
+        )
     }
 
     pub(crate) fn info(&self) -> ConnectionInfo {
@@ -226,10 +237,11 @@ impl Connection {
     }
 
     /// Helper to create a `ConnectionCheckedOutEvent` for the connection.
-    pub(super) fn checked_out_event(&self) -> ConnectionCheckedOutEvent {
+    pub(super) fn checked_out_event(&self, time_started: Instant) -> ConnectionCheckedOutEvent {
         ConnectionCheckedOutEvent {
             address: self.address.clone(),
             connection_id: self.id,
+            duration: Instant::now() - time_started,
         }
     }
 
@@ -246,6 +258,7 @@ impl Connection {
         ConnectionReadyEvent {
             address: self.address.clone(),
             connection_id: self.id,
+            duration: Instant::now() - self.time_created,
         }
     }
 
@@ -422,6 +435,7 @@ impl Connection {
             server_id: self.server_id,
             address: self.address.clone(),
             generation: self.generation,
+            time_created: self.time_created,
             stream: std::mem::replace(&mut self.stream, BufStream::new(AsyncStream::Null)),
             event_emitter: self.event_emitter.take(),
             stream_description: self.stream_description.take(),
@@ -599,6 +613,7 @@ pub(crate) struct PendingConnection {
     pub(crate) address: ServerAddress,
     pub(crate) generation: PoolGeneration,
     pub(crate) event_emitter: CmapEventEmitter,
+    pub(crate) time_created: Instant,
 }
 
 impl PendingConnection {
diff --git a/src/cmap/test/event.rs b/src/cmap/test/event.rs
@@ -1,4 +1,7 @@
-use std::sync::{Arc, RwLock};
+use std::{
+    sync::{Arc, RwLock},
+    time::Duration,
+};
 
 use serde::{de::Unexpected, Deserialize, Deserializer, Serialize};
 
@@ -268,5 +271,6 @@ where
         reason,
         #[cfg(feature = "tracing-unstable")]
         error: None,
+        duration: Duration::ZERO,
     })
 }
diff --git a/src/cmap/worker.rs b/src/cmap/worker.rs
@@ -38,7 +38,7 @@ use crate::{
 
 use std::{
     collections::{HashMap, VecDeque},
-    time::Duration,
+    time::{Duration, Instant},
 };
 
 const DEFAULT_MAX_CONNECTING: u32 = 2;
@@ -468,6 +468,7 @@ impl ConnectionPoolWorker {
             address: self.address.clone(),
             generation: self.generation.clone(),
             event_emitter: self.event_emitter.clone(),
+            time_created: Instant::now(),
         };
         self.next_connection_id += 1;
         self.event_emitter
diff --git a/src/event/cmap.rs b/src/event/cmap.rs
@@ -135,6 +135,10 @@ pub struct ConnectionReadyEvent {
     /// to identify other events related to this connection.
     #[serde(default = "default_connection_id")]
     pub connection_id: u32,
+
+    /// The time it took to establish the connection.
+    #[serde(default = "Duration::default")]
+    pub duration: Duration,
 }
 
 /// Event emitted when a connection is closed.
@@ -216,6 +220,10 @@ pub struct ConnectionCheckoutFailedEvent {
     #[serde(skip)]
     #[derivative(PartialEq = "ignore")]
     pub(crate) error: Option<crate::error::Error>,
+
+    /// See [ConnectionCheckedOutEvent::duration].
+    #[serde(default = "Duration::default")]
+    pub duration: Duration,
 }
 
 /// The reasons a connection may not be able to be checked out.
@@ -245,6 +253,10 @@ pub struct ConnectionCheckedOutEvent {
     /// to identify other events related to this connection.
     #[serde(default = "default_connection_id")]
     pub connection_id: u32,
+
+    /// The time it took to check out the connection.
+    #[serde(default = "Duration::default")]
+    pub duration: Duration,
 }
 
 /// Event emitted when a connection is checked back into a connection pool.
diff --git a/src/test/spec/json/connection-monitoring-and-pooling/logging/connection-logging.json b/src/test/spec/json/connection-monitoring-and-pooling/logging/connection-logging.json
@@ -140,6 +140,13 @@
                     "int",
                     "long"
                   ]
+                },
+                "durationMS": {
+                  "$$type": [
+                    "double",
+                    "int",
+                    "long"
+                  ]
                 }
               }
             },
@@ -162,6 +169,13 @@
                     "int",
                     "long"
                   ]
+                },
+                "durationMS": {
+                  "$$type": [
+                    "double",
+                    "int",
+                    "long"
+                  ]
                 }
               }
             },
@@ -424,6 +438,13 @@
                 "reason": "An error occurred while trying to establish a new connection",
                 "error": {
                   "$$exists": true
+                },
+                "durationMS": {
+                  "$$type": [
+                    "double",
+                    "int",
+                    "long"
+                  ]
                 }
               }
             }
diff --git a/src/test/spec/json/connection-monitoring-and-pooling/logging/connection-logging.yml b/src/test/spec/json/connection-monitoring-and-pooling/logging/connection-logging.yml
@@ -66,14 +66,16 @@ tests:
               driverConnectionId: { $$type: [int, long] }
               serverHost: { $$type: string }
               serverPort: { $$type: [int, long] }
-          
+              durationMS: { $$type: [double, int, long] }
+
           - level: debug
             component: connection
             data:
               message: "Connection checked out"
               driverConnectionId: { $$type: [int, long] }
               serverHost: { $$type: string }
               serverPort: { $$type: [int, long] }
+              durationMS: { $$type: [double, int, long] }
 
           - level: debug
             component: connection
@@ -194,3 +196,4 @@ tests:
               serverPort: { $$type: [int, long] }
               reason: "An error occurred while trying to establish a new connection"
               error: { $$exists: true }
+              durationMS: { $$type: [double, int, long] }
diff --git a/src/test/spec/json/connection-monitoring-and-pooling/logging/connection-pool-options.json b/src/test/spec/json/connection-monitoring-and-pooling/logging/connection-pool-options.json
@@ -1,5 +1,5 @@
 {
-  "description": "connection-logging",
+  "description": "connection-pool-options",
   "schemaVersion": "1.13",
   "runOnRequirements": [
     {
@@ -128,6 +128,13 @@
                     "int",
                     "long"
                   ]
+                },
+                "durationMS": {
+                  "$$type": [
+                    "double",
+                    "int",
+                    "long"
+                  ]
                 }
               }
             }
diff --git a/src/test/spec/json/connection-monitoring-and-pooling/logging/connection-pool-options.yml b/src/test/spec/json/connection-monitoring-and-pooling/logging/connection-pool-options.yml
@@ -1,4 +1,4 @@
-description: "connection-logging"
+description: "connection-pool-options"
 
 schemaVersion: "1.13"
 
@@ -71,6 +71,7 @@ tests:
               driverConnectionId: { $$type: [int, long] }
               serverHost: { $$type: string }
               serverPort: { $$type: [int, long] }
+              durationMS: { $$type: [double, int, long] }
 
   # Drivers who have not done DRIVERS-1943 will need to skip this test.
   - description: "maxConnecting should be included in connection pool created message when specified"
diff --git a/src/trace/connection.rs b/src/trace/connection.rs
@@ -95,6 +95,7 @@ impl CmapEventHandler for ConnectionTracingEventEmitter {
             serverHost = event.address.host().as_ref(),
             serverPort = event.address.port_tracing_representation(),
             driverConnectionId = event.connection_id,
+            durationMS = event.duration.as_millis(),
             "Connection ready",
         );
     }
@@ -130,6 +131,7 @@ impl CmapEventHandler for ConnectionTracingEventEmitter {
             serverPort = event.address.port_tracing_representation(),
             reason = event.reason.tracing_representation(),
             error = event.error.map(|e| e.tracing_representation()),
+            durationMS = event.duration.as_millis(),
             "Connection checkout failed",
         );
     }
@@ -141,6 +143,7 @@ impl CmapEventHandler for ConnectionTracingEventEmitter {
             serverHost = event.address.host().as_ref(),
             serverPort = event.address.port_tracing_representation(),
             driverConnectionId = event.connection_id,
+            durationMS = event.duration.as_millis(),
             "Connection checked out",
         );
     }

Original file line number	Diff line number	Diff line change
`@@ -140,6 +140,13 @@`
`140`	`140`	`"int",`
`141`	`141`	`"long"`
`142`	`142`	`]`
	`143`	`+ },`
	`144`	`+ "durationMS": {`
	`145`	`+ "$$type": [`
	`146`	`+ "double",`
	`147`	`+ "int",`
	`148`	`+ "long"`
	`149`	`+ ]`
`143`	`150`	`}`
`144`	`151`	`}`
`145`	`152`	`},`
`@@ -162,6 +169,13 @@`
`162`	`169`	`"int",`
`163`	`170`	`"long"`
`164`	`171`	`]`
	`172`	`+ },`
	`173`	`+ "durationMS": {`
	`174`	`+ "$$type": [`
	`175`	`+ "double",`
	`176`	`+ "int",`
	`177`	`+ "long"`
	`178`	`+ ]`
`165`	`179`	`}`
`166`	`180`	`}`
`167`	`181`	`},`
`@@ -424,6 +438,13 @@`
`424`	`438`	`"reason": "An error occurred while trying to establish a new connection",`
`425`	`439`	`"error": {`
`426`	`440`	`"$$exists": true`
	`441`	`+ },`
	`442`	`+ "durationMS": {`
	`443`	`+ "$$type": [`
	`444`	`+ "double",`
	`445`	`+ "int",`
	`446`	`+ "long"`
	`447`	`+ ]`
`427`	`448`	`}`
`428`	`449`	`}`
`429`	`450`	`}`
Original file line number	Diff line number	Diff line change
`@@ -1,5 +1,5 @@`
`1`	`1`	`{`
`2`		`- "description": "connection-logging",`
	`2`	`+ "description": "connection-pool-options",`
`3`	`3`	`"schemaVersion": "1.13",`
`4`	`4`	`"runOnRequirements": [`
`5`	`5`	`{`
`@@ -128,6 +128,13 @@`
`128`	`128`	`"int",`
`129`	`129`	`"long"`
`130`	`130`	`]`
	`131`	`+ },`
	`132`	`+ "durationMS": {`
	`133`	`+ "$$type": [`
	`134`	`+ "double",`
	`135`	`+ "int",`
	`136`	`+ "long"`
	`137`	`+ ]`
`131`	`138`	`}`
`132`	`139`	`}`
`133`	`140`	`}`