PYTHON-5846 Address PR review comments on telemetry refactor

- Fix race condition in _CmapTelemetry: remove shared _conn_created_start
  and _checkout_start slots; connection_ready now takes creation_time from
  AsyncConnection.creation_time, checkout_started returns the start time
  and checkout_succeeded/failed accept it as a parameter
- Move SRV monitor failure log into _telemetry.py as log_srv_monitor_failure
- Add _log_retry helper to _ClientConnectionRetryable to deduplicate
  log_command_retry call sites
- Fix _ServerSelectionTelemetry.failed to accept live topology_description
  at failure time instead of using the stale snapshot from construction

Author: blink1073

pymongo/_telemetry.py

@@ -204,9 +204,7 @@ class _CmapTelemetry:
 
     __slots__ = (
         "_address",
-        "_checkout_start",
         "_client_id",
-        "_conn_created_start",
         "_listeners",
         "_log",
         "_publish",
@@ -283,22 +281,19 @@ def pool_closed(self) -> None:
             self._listeners.publish_pool_closed(self._address)
 
     def connection_created(self, conn_id: int) -> None:
-        # Always record start time: logging or publishing may be enabled by the time
-        # connection_ready is called to compute the duration.
-        self._conn_created_start = time.monotonic()
         # Log before publishing to prevent potential listener preemption in tests.
         if self._should_log:
             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) -> None:
+    def connection_ready(self, conn_id: int, creation_time: float) -> None:
         should_log = self._should_log
         should_publish = self._should_publish
         if not should_log and not should_publish:
             return
-        duration = max(0.0, time.monotonic() - self._conn_created_start)
+        duration = max(0.0, time.monotonic() - creation_time)
         # Log before publishing to prevent potential listener preemption in tests.
         if should_log:
             self._emit_log(
@@ -324,24 +319,22 @@ def connection_closed(self, conn_id: int, reason: str) -> None:
                 error=reason,
             )
 
-    def checkout_started(self) -> None:
-        should_log = self._should_log
-        should_publish = self._should_publish
-        # Always record start time: logging or publishing may be enabled by the time
-        # checkout_succeeded or checkout_failed is called to compute the duration.
-        self._checkout_start = time.monotonic()
-        if should_publish:
+    def checkout_started(self) -> float:
+        """Emit the checkout started event/log and return the start time for duration tracking."""
+        start = time.monotonic()
+        if self._should_publish:
             assert self._listeners is not None
             self._listeners.publish_connection_check_out_started(self._address)
-        if should_log:
+        if self._should_log:
             self._emit_log(_ConnectionStatusMessage.CHECKOUT_STARTED)
+        return start
 
-    def checkout_succeeded(self, conn_id: int) -> None:
+    def checkout_succeeded(self, conn_id: int, start: float) -> None:
         should_log = self._should_log
         should_publish = self._should_publish
         if not should_log and not should_publish:
             return
-        duration = max(0.0, time.monotonic() - self._checkout_start)
+        duration = max(0.0, time.monotonic() - start)
         if should_publish:
             assert self._listeners is not None
             self._listeners.publish_connection_checked_out(self._address, conn_id, duration)
@@ -352,12 +345,12 @@ def checkout_succeeded(self, conn_id: int) -> None:
                 durationMS=duration * 1000,
             )
 
-    def checkout_failed(self, reason: str, error: str) -> None:
+    def checkout_failed(self, reason: str, error: str, start: float) -> None:
         should_log = self._should_log
         should_publish = self._should_publish
         if not should_log and not should_publish:
             return
-        duration = max(0.0, time.monotonic() - self._checkout_start)
+        duration = max(0.0, time.monotonic() - start)
         if should_publish:
             assert self._listeners is not None
             self._listeners.publish_connection_check_out_failed(self._address, error, duration)
@@ -632,43 +625,62 @@ def __init__(
         # logging level is stable for its lifetime.
         self._should_log = _SERVER_SELECTION_LOGGER.isEnabledFor(logging.DEBUG)
 
-    def _emit_log(self, message: _ServerSelectionStatusMessage, **extra: Any) -> None:
+    def _emit_log(
+        self, message: _ServerSelectionStatusMessage, topology_description: Any, **extra: Any
+    ) -> None:
         _debug_log(
             _SERVER_SELECTION_LOGGER,
             message=message,
             clientId=self._topology_id,
             selector=self._selector,
             operation=self._operation,
             operationId=self._operation_id,
-            topologyDescription=self._topology_description,
+            topologyDescription=topology_description,
             **extra,
         )
 
     def started(self) -> None:
         """Emit the server selection STARTED log entry."""
         if self._should_log:
-            self._emit_log(_ServerSelectionStatusMessage.STARTED)
+            self._emit_log(_ServerSelectionStatusMessage.STARTED, self._topology_description)
 
     def waiting(self, remaining_time_ms: int) -> None:
         """Emit the server selection WAITING log entry."""
         if self._should_log:
-            self._emit_log(_ServerSelectionStatusMessage.WAITING, remainingTimeMS=remaining_time_ms)
+            self._emit_log(
+                _ServerSelectionStatusMessage.WAITING,
+                self._topology_description,
+                remainingTimeMS=remaining_time_ms,
+            )
 
-    def failed(self, failure: str) -> None:
-        """Emit the server selection FAILED log entry."""
+    def failed(self, failure: str, topology_description: Any = None) -> None:
+        """Emit the server selection FAILED log entry with the current topology description."""
         if self._should_log:
-            self._emit_log(_ServerSelectionStatusMessage.FAILED, failure=failure)
+            self._emit_log(
+                _ServerSelectionStatusMessage.FAILED,
+                topology_description
+                if topology_description is not None
+                else self._topology_description,
+                failure=failure,
+            )
 
     def succeeded(self, server_host: str, server_port: Optional[int]) -> None:
         """Emit the server selection SUCCEEDED log entry."""
         if self._should_log:
             self._emit_log(
                 _ServerSelectionStatusMessage.SUCCEEDED,
+                self._topology_description,
                 serverHost=server_host,
                 serverPort=server_port,
             )
 
 
+def log_srv_monitor_failure(failure: Exception) -> None:
+    """Emit a log entry when the SRV monitor fails to poll DNS records."""
+    if _SDAM_LOGGER.isEnabledFor(logging.DEBUG):
+        _debug_log(_SDAM_LOGGER, message="SRV monitor check failed", failure=repr(failure))
+
+
 def log_command_retry(
     topology_id: Any,
     command_name: str,

pymongo/asynchronous/mongo_client.py

@@ -2990,6 +2990,15 @@ async def _get_server(self) -> Server:
             operation_id=self._operation_id,
         )
 
+    def _log_retry(self, is_write: bool) -> None:
+        log_command_retry(
+            self._client._topology_id,
+            self._operation,
+            self._operation_id,
+            self._attempt_number,
+            is_write,
+        )
+
     async def _write(self) -> T:
         """Wrapper method for write-type retryable client executions
 
@@ -3013,13 +3022,7 @@ async def _write(self) -> T:
                     self._check_last_error()
                     self._retryable = False
                 if self._retrying:
-                    log_command_retry(
-                        self._client._topology_id,
-                        self._operation,
-                        self._operation_id,
-                        self._attempt_number,
-                        is_write=True,
-                    )
+                    self._log_retry(is_write=True)
                 return await self._func(self._session, conn, self._retryable)  # type: ignore
         except PyMongoError as exc:
             if not self._retryable:
@@ -3042,13 +3045,7 @@ async def _read(self) -> T:
             if self._retrying and not self._retryable and not self._always_retryable:
                 self._check_last_error()
             if self._retrying:
-                log_command_retry(
-                    self._client._topology_id,
-                    self._operation,
-                    self._operation_id,
-                    self._attempt_number,
-                    is_write=False,
-                )
+                self._log_retry(is_write=False)
             return await self._func(self._session, self._server, conn, read_pref)  # type: ignore
 
 

pymongo/asynchronous/monitor.py

@@ -24,12 +24,11 @@
 
 from pymongo import common, periodic_executor
 from pymongo._csot import MovingMinimum
-from pymongo._telemetry import _HeartbeatTelemetry
+from pymongo._telemetry import _HeartbeatTelemetry, log_srv_monitor_failure
 from pymongo.asynchronous.srv_resolver import _SrvResolver
 from pymongo.errors import NetworkTimeout, _OperationCancelled
 from pymongo.hello import Hello
 from pymongo.lock import _async_create_lock
-from pymongo.logger import _SDAM_LOGGER, _debug_log
 from pymongo.periodic_executor import _shutdown_executors
 from pymongo.pool_options import _is_faas
 from pymongo.read_preferences import MovingAverage
@@ -385,7 +384,7 @@ async def _get_seedlist(self) -> Optional[list[tuple[str, Any]]]:
             # - SRV records must be rescanned every heartbeatFrequencyMS
             # - Topology must be left unchanged
             self.request_check()
-            _debug_log(_SDAM_LOGGER, message="SRV monitor check failed", failure=repr(exc))
+            log_srv_monitor_failure(exc)
             return None
         else:
             self._executor.update_interval(max(ttl, common.MIN_SRV_RESCAN_INTERVAL))

pymongo/asynchronous/pool.py

@@ -128,6 +128,7 @@ def __init__(
         self.id = id
         self.is_sdam = is_sdam
         self.closed = False
+        self.creation_time = time.monotonic()
         self.last_checkin_time = time.monotonic()
         self.performed_handshake = False
         self.is_writable: bool = False
@@ -466,7 +467,7 @@ async def authenticate(self, reauthenticate: bool = False) -> None:
 
                 await auth.authenticate(creds, self, reauthenticate=reauthenticate)
             self.ready = True
-            self._telemetry.connection_ready(self.id)
+            self._telemetry.connection_ready(self.id, self.creation_time)
 
     def validate_session(
         self, client: Optional[AsyncMongoClient[Any]], session: Optional[AsyncClientSession]
@@ -976,11 +977,11 @@ async def checkout(
 
         :param handler: A _MongoClientErrorHandler.
         """
-        self._telemetry.checkout_started()
+        start = self._telemetry.checkout_started()
 
-        conn = await self._get_conn(handler=handler)
+        conn = await self._get_conn(handler=handler, checkout_start=start)
 
-        self._telemetry.checkout_succeeded(conn.id)
+        self._telemetry.checkout_succeeded(conn.id, start)
         try:
             async with self.lock:
                 self.active_contexts.add(conn.cancel_context)
@@ -1011,12 +1012,13 @@ async def checkout(
         elif conn.active:
             await self.checkin(conn)
 
-    def _raise_if_not_ready(self, emit_event: bool) -> None:
+    def _raise_if_not_ready(self, emit_event: bool, checkout_start: float) -> None:
         if self.state != PoolState.READY:
             if emit_event:
                 self._telemetry.checkout_failed(
                     "An error occurred while trying to establish a new connection",
                     ConnectionCheckOutFailedReason.CONN_ERROR,
+                    checkout_start,
                 )
 
             details = _get_timeout_details(self.opts)
@@ -1025,7 +1027,7 @@ def _raise_if_not_ready(self, emit_event: bool) -> None:
             )
 
     async def _get_conn(
-        self, handler: Optional[_MongoClientErrorHandler] = None
+        self, handler: Optional[_MongoClientErrorHandler] = None, checkout_start: float = 0.0
     ) -> AsyncConnection:
         """Get or create a AsyncConnection. Can raise ConnectionFailure."""
         # We use the pid here to avoid issues with fork / multiprocessing.
@@ -1038,6 +1040,7 @@ async def _get_conn(
             self._telemetry.checkout_failed(
                 "Connection pool was closed",
                 ConnectionCheckOutFailedReason.POOL_CLOSED,
+                checkout_start,
             )
             raise _PoolClosedError(
                 "Attempted to check out a connection from closed connection pool"
@@ -1055,16 +1058,16 @@ async def _get_conn(
             deadline = None
 
         async with self.size_cond:
-            self._raise_if_not_ready(emit_event=True)
+            self._raise_if_not_ready(emit_event=True, checkout_start=checkout_start)
             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()
-                self._raise_if_not_ready(emit_event=True)
+                    self._raise_wait_queue_timeout(checkout_start)
+                self._raise_if_not_ready(emit_event=True, checkout_start=checkout_start)
             self.requests += 1
 
         # We've now acquired the semaphore and must release it on error.
@@ -1079,7 +1082,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(emit_event=False)
+                    self._raise_if_not_ready(emit_event=False, checkout_start=checkout_start)
                     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):
@@ -1088,8 +1091,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()
-                        self._raise_if_not_ready(emit_event=False)
+                            self._raise_wait_queue_timeout(checkout_start)
+                        self._raise_if_not_ready(emit_event=False, checkout_start=checkout_start)
 
                     try:
                         conn = self.conns.popleft()
@@ -1121,6 +1124,7 @@ async def _get_conn(
                 self._telemetry.checkout_failed(
                     "An error occurred while trying to establish a new connection",
                     ConnectionCheckOutFailedReason.CONN_ERROR,
+                    checkout_start,
                 )
             raise
 
@@ -1212,10 +1216,11 @@ async def _perished(self, conn: AsyncConnection) -> bool:
 
         return False
 
-    def _raise_wait_queue_timeout(self) -> NoReturn:
+    def _raise_wait_queue_timeout(self, checkout_start: float) -> NoReturn:
         self._telemetry.checkout_failed(
             "Wait queue timeout elapsed without a connection becoming available",
             ConnectionCheckOutFailedReason.TIMEOUT,
+            checkout_start,
         )
         timeout = _csot.get_timeout() or self.opts.wait_queue_timeout
         if self.opts.load_balanced:

pymongo/asynchronous/topology.py

@@ -296,7 +296,7 @@ async def _select_servers_loop(
         while not server_descriptions:
             # No suitable servers.
             if timeout == 0 or now > end_time:
-                ss.failed(self._error_message(selector))
+                ss.failed(self._error_message(selector), self.description)
                 raise ServerSelectionTimeoutError(
                     f"{self._error_message(selector)}, Timeout: {timeout}s, Topology Description: {self.description!r}"
                 )