PYTHON-5846 Move duration tracking into telemetry classes

_CmapTelemetry now owns connection-creation and checkout durations:
connection_created() starts the clock, connection_ready() computes it;
checkout_started() starts the clock, checkout_succeeded/failed() compute it.

_HeartbeatTelemetry.started() starts the clock; failed() computes its
own duration instead of receiving it as a parameter.

Removes checkout_started_time from Pool._get_conn, _raise_if_not_ready,
and _raise_wait_queue_timeout, and removes AsyncConnection.creation_time.

Author: blink1073

pymongo/_telemetry.py

@@ -19,6 +19,7 @@
 import datetime
 import logging
 import queue
+import time
 from collections.abc import MutableMapping
 from typing import TYPE_CHECKING, Any, Optional
 
@@ -198,7 +199,15 @@ def failed(
 class _CmapTelemetry:
     """Combines CMAP structured logging and APM event publishing for pool and connection events."""
 
-    __slots__ = ("_address", "_client_id", "_listeners", "_should_log", "_should_publish")
+    __slots__ = (
+        "_address",
+        "_checkout_start",
+        "_client_id",
+        "_conn_created_start",
+        "_listeners",
+        "_should_log",
+        "_should_publish",
+    )
 
     def __init__(
         self,
@@ -258,13 +267,15 @@ def pool_closed(self) -> None:
             self._listeners.publish_pool_closed(self._address)
 
     def connection_created(self, conn_id: int) -> None:
+        self._conn_created_start = time.monotonic()
         # Log before publishing to prevent potential listener preemption in tests.
         self._emit_log(_ConnectionStatusMessage.CONN_CREATED, driverConnectionId=conn_id)
         if self._should_publish:
             assert self._listeners is not None
             self._listeners.publish_connection_created(self._address, conn_id)
 
-    def connection_ready(self, conn_id: int, duration: float) -> None:
+    def connection_ready(self, conn_id: int) -> None:
+        duration = max(0.0, time.monotonic() - self._conn_created_start)
         # Log before publishing to prevent potential listener preemption in tests.
         self._emit_log(
             _ConnectionStatusMessage.CONN_READY,
@@ -287,12 +298,14 @@ def connection_closed(self, conn_id: int, reason: str) -> None:
         )
 
     def checkout_started(self) -> None:
+        self._checkout_start = time.monotonic()
         if self._should_publish:
             assert self._listeners is not None
             self._listeners.publish_connection_check_out_started(self._address)
         self._emit_log(_ConnectionStatusMessage.CHECKOUT_STARTED)
 
-    def checkout_succeeded(self, conn_id: int, duration: float) -> None:
+    def checkout_succeeded(self, conn_id: int) -> None:
+        duration = max(0.0, time.monotonic() - self._checkout_start)
         if self._should_publish:
             assert self._listeners is not None
             self._listeners.publish_connection_checked_out(self._address, conn_id, duration)
@@ -302,7 +315,8 @@ def checkout_succeeded(self, conn_id: int, duration: float) -> None:
             durationMS=duration,
         )
 
-    def checkout_failed(self, reason: str, error: str, duration: float) -> None:
+    def checkout_failed(self, reason: str, error: str) -> None:
+        duration = max(0.0, time.monotonic() - self._checkout_start)
         if self._should_publish:
             assert self._listeners is not None
             self._listeners.publish_connection_check_out_failed(self._address, error, duration)
@@ -329,7 +343,7 @@ class _HeartbeatTelemetry:
     context, then :meth:`succeeded` or :meth:`failed` when the outcome is known.
     """
 
-    __slots__ = ("_address", "_awaited", "_listeners", "_publish", "_topology_id")
+    __slots__ = ("_address", "_awaited", "_listeners", "_publish", "_start", "_topology_id")
 
     def __init__(
         self,
@@ -347,6 +361,7 @@ def __init__(
 
     def started(self) -> None:
         """Publish the APM heartbeat-started event (before connection checkout)."""
+        self._start = time.monotonic()
         if self._publish:
             assert self._listeners is not None
             self._listeners.publish_server_heartbeat_started(self._address, self._awaited)
@@ -392,8 +407,9 @@ def succeeded(
                 reply=response.document,
             )
 
-    def failed(self, duration: float, error: Exception, conn_id: Optional[int]) -> None:
+    def failed(self, error: Exception, conn_id: Optional[int]) -> None:
         """Emit the FAILED log entry and APM event."""
+        duration = max(0.0, time.monotonic() - self._start)
         if self._publish:
             assert self._listeners is not None
             self._listeners.publish_server_heartbeat_failed(

pymongo/asynchronous/monitor.py

@@ -259,17 +259,15 @@ async def _check_server(self) -> ServerDescription:
         """
         self._conn_id = None
         self._current_hb = None
-        start = time.monotonic()
         try:
             return await self._check_once()
         except ReferenceError:
             raise
         except Exception as error:
             _sanitize(error)
             address = self._server_description.address
-            duration = _monotonic_duration(start)
             if self._current_hb is not None:
-                self._current_hb.failed(duration, error, self._conn_id)
+                self._current_hb.failed(error, self._conn_id)
             await self._reset_connection()
             if isinstance(error, _OperationCancelled):
                 raise

pymongo/asynchronous/pool.py

@@ -166,7 +166,6 @@ def __init__(
         self.active = False
         self.last_timeout = self.opts.socket_timeout
         self.connect_rtt = 0.0
-        self.creation_time = time.monotonic()
         # For gossiping $clusterTime from the connection handshake to the client.
         self._cluster_time = None
 
@@ -467,8 +466,7 @@ async def authenticate(self, reauthenticate: bool = False) -> None:
 
                 await auth.authenticate(creds, self, reauthenticate=reauthenticate)
             self.ready = True
-            duration = time.monotonic() - self.creation_time
-            self._telemetry.connection_ready(self.id, duration)
+            self._telemetry.connection_ready(self.id)
 
     def validate_session(
         self, client: Optional[AsyncMongoClient[Any]], session: Optional[AsyncClientSession]
@@ -978,13 +976,11 @@ async def checkout(
 
         :param handler: A _MongoClientErrorHandler.
         """
-        checkout_started_time = time.monotonic()
         self._telemetry.checkout_started()
 
-        conn = await self._get_conn(checkout_started_time, handler=handler)
+        conn = await self._get_conn(handler=handler)
 
-        duration = time.monotonic() - checkout_started_time
-        self._telemetry.checkout_succeeded(conn.id, duration)
+        self._telemetry.checkout_succeeded(conn.id)
         try:
             async with self.lock:
                 self.active_contexts.add(conn.cancel_context)
@@ -1015,14 +1011,12 @@ async def checkout(
         elif conn.active:
             await self.checkin(conn)
 
-    def _raise_if_not_ready(self, checkout_started_time: float, emit_event: bool) -> None:
+    def _raise_if_not_ready(self, emit_event: bool) -> None:
         if self.state != PoolState.READY:
             if emit_event:
-                duration = time.monotonic() - checkout_started_time
                 self._telemetry.checkout_failed(
                     "An error occurred while trying to establish a new connection",
                     ConnectionCheckOutFailedReason.CONN_ERROR,
-                    duration,
                 )
 
             details = _get_timeout_details(self.opts)
@@ -1031,7 +1025,7 @@ def _raise_if_not_ready(self, checkout_started_time: float, emit_event: bool) ->
             )
 
     async def _get_conn(
-        self, checkout_started_time: float, handler: Optional[_MongoClientErrorHandler] = None
+        self, handler: Optional[_MongoClientErrorHandler] = None
     ) -> AsyncConnection:
         """Get or create a AsyncConnection. Can raise ConnectionFailure."""
         # We use the pid here to avoid issues with fork / multiprocessing.
@@ -1041,11 +1035,9 @@ async def _get_conn(
             await self.reset_without_pause()
 
         if self.closed:
-            duration = time.monotonic() - checkout_started_time
             self._telemetry.checkout_failed(
                 "Connection pool was closed",
                 ConnectionCheckOutFailedReason.POOL_CLOSED,
-                duration,
             )
             raise _PoolClosedError(
                 "Attempted to check out a connection from closed connection pool"
@@ -1063,16 +1055,16 @@ async def _get_conn(
             deadline = None
 
         async with self.size_cond:
-            self._raise_if_not_ready(checkout_started_time, emit_event=True)
+            self._raise_if_not_ready(emit_event=True)
             while not (self.requests < self.max_pool_size):
                 timeout = deadline - time.monotonic() if deadline else None
                 if not await _async_cond_wait(self.size_cond, timeout):
                     # Timed out, notify the next thread to ensure a
                     # timeout doesn't consume the condition.
                     if self.requests < self.max_pool_size:
                         self.size_cond.notify()
-                    self._raise_wait_queue_timeout(checkout_started_time)
-                self._raise_if_not_ready(checkout_started_time, emit_event=True)
+                    self._raise_wait_queue_timeout()
+                self._raise_if_not_ready(emit_event=True)
             self.requests += 1
 
         # We've now acquired the semaphore and must release it on error.
@@ -1087,7 +1079,7 @@ async def _get_conn(
                 # CMAP: we MUST wait for either maxConnecting OR for a socket
                 # to be checked back into the pool.
                 async with self._max_connecting_cond:
-                    self._raise_if_not_ready(checkout_started_time, emit_event=False)
+                    self._raise_if_not_ready(emit_event=False)
                     while not (self.conns or self._pending < self._max_connecting):
                         timeout = deadline - time.monotonic() if deadline else None
                         if not await _async_cond_wait(self._max_connecting_cond, timeout):
@@ -1096,8 +1088,8 @@ async def _get_conn(
                             if self.conns or self._pending < self._max_connecting:
                                 self._max_connecting_cond.notify()
                             emitted_event = True
-                            self._raise_wait_queue_timeout(checkout_started_time)
-                        self._raise_if_not_ready(checkout_started_time, emit_event=False)
+                            self._raise_wait_queue_timeout()
+                        self._raise_if_not_ready(emit_event=False)
 
                     try:
                         conn = self.conns.popleft()
@@ -1126,11 +1118,9 @@ async def _get_conn(
                 self.size_cond.notify()
 
             if not emitted_event:
-                duration = time.monotonic() - checkout_started_time
                 self._telemetry.checkout_failed(
                     "An error occurred while trying to establish a new connection",
                     ConnectionCheckOutFailedReason.CONN_ERROR,
-                    duration,
                 )
             raise
 
@@ -1222,12 +1212,10 @@ async def _perished(self, conn: AsyncConnection) -> bool:
 
         return False
 
-    def _raise_wait_queue_timeout(self, checkout_started_time: float) -> NoReturn:
-        duration = time.monotonic() - checkout_started_time
+    def _raise_wait_queue_timeout(self) -> NoReturn:
         self._telemetry.checkout_failed(
             "Wait queue timeout elapsed without a connection becoming available",
             ConnectionCheckOutFailedReason.TIMEOUT,
-            duration,
         )
         timeout = _csot.get_timeout() or self.opts.wait_queue_timeout
         if self.opts.load_balanced:

pymongo/synchronous/monitor.py

@@ -257,17 +257,15 @@ def _check_server(self) -> ServerDescription:
         """
         self._conn_id = None
         self._current_hb = None
-        start = time.monotonic()
         try:
             return self._check_once()
         except ReferenceError:
             raise
         except Exception as error:
             _sanitize(error)
             address = self._server_description.address
-            duration = _monotonic_duration(start)
             if self._current_hb is not None:
-                self._current_hb.failed(duration, error, self._conn_id)
+                self._current_hb.failed(error, self._conn_id)
             self._reset_connection()
             if isinstance(error, _OperationCancelled):
                 raise