PYTHON-5846 Address further PR review comments

- Restore original _raise_if_not_ready and _get_conn parameter names/order
- Add docstrings to all public methods on _CmapTelemetry and _SdamTelemetry
- Remove redundant _publish/_listener/_events attrs from Server.__init__
- Avoid creating two _ServerSelectionTelemetry per select_server call by
  threading ss through select_servers and _select_server
- Make _ServerSelectionTelemetry.failed topology_description a required arg
- Move SRV monitor failure log to telemetry.py (log_srv_monitor_failure)

Author: blink1073

pymongo/_telemetry.py

@@ -245,6 +245,7 @@ def _emit_log(self, message: _ConnectionStatusMessage, **extra: Any) -> None:
         )
 
     def pool_created(self, non_default_options: dict[str, Any]) -> None:
+        """Emit the pool created log entry and APM event."""
         # Log before publishing to prevent potential listener preemption in tests.
         if self._should_log:
             self._emit_log(_ConnectionStatusMessage.POOL_CREATED, **non_default_options)
@@ -253,6 +254,7 @@ def pool_created(self, non_default_options: dict[str, Any]) -> None:
             self._listeners.publish_pool_created(self._address, non_default_options)
 
     def pool_ready(self) -> None:
+        """Emit the pool ready log entry and APM event."""
         # Log before publishing to prevent potential listener preemption in tests.
         if self._should_log:
             self._emit_log(_ConnectionStatusMessage.POOL_READY)
@@ -261,6 +263,7 @@ def pool_ready(self) -> None:
             self._listeners.publish_pool_ready(self._address)
 
     def pool_cleared(self, service_id: Optional[ObjectId], interrupt_connections: bool) -> None:
+        """Emit the pool cleared log entry and APM event."""
         # Log before publishing to prevent potential listener preemption in tests.
         if self._should_log:
             self._emit_log(_ConnectionStatusMessage.POOL_CLEARED, serviceId=service_id)
@@ -273,6 +276,7 @@ def pool_cleared(self, service_id: Optional[ObjectId], interrupt_connections: bo
             )
 
     def pool_closed(self) -> None:
+        """Emit the pool closed log entry and APM event."""
         # Log before publishing to prevent potential listener preemption in tests.
         if self._should_log:
             self._emit_log(_ConnectionStatusMessage.POOL_CLOSED)
@@ -281,6 +285,7 @@ def pool_closed(self) -> None:
             self._listeners.publish_pool_closed(self._address)
 
     def connection_created(self, conn_id: int) -> None:
+        """Emit the connection created log entry and APM event."""
         # Log before publishing to prevent potential listener preemption in tests.
         if self._should_log:
             self._emit_log(_ConnectionStatusMessage.CONN_CREATED, driverConnectionId=conn_id)
@@ -289,6 +294,7 @@ def connection_created(self, conn_id: int) -> None:
             self._listeners.publish_connection_created(self._address, conn_id)
 
     def connection_ready(self, conn_id: int, creation_time: float) -> None:
+        """Emit the connection ready log entry and APM event."""
         should_log = self._should_log
         should_publish = self._should_publish
         if not should_log and not should_publish:
@@ -306,6 +312,7 @@ def connection_ready(self, conn_id: int, creation_time: float) -> None:
             self._listeners.publish_connection_ready(self._address, conn_id, duration)
 
     def connection_closed(self, conn_id: int, reason: str) -> None:
+        """Emit the connection closed log entry and APM event."""
         should_log = self._should_log
         should_publish = self._should_publish
         if should_publish:
@@ -330,6 +337,7 @@ def checkout_started(self) -> float:
         return start
 
     def checkout_succeeded(self, conn_id: int, start: float) -> None:
+        """Emit the checkout succeeded log entry and APM event."""
         should_log = self._should_log
         should_publish = self._should_publish
         if not should_log and not should_publish:
@@ -346,6 +354,7 @@ def checkout_succeeded(self, conn_id: int, start: float) -> None:
             )
 
     def checkout_failed(self, reason: str, error: str, start: float) -> None:
+        """Emit the checkout failed log entry and APM event."""
         should_log = self._should_log
         should_publish = self._should_publish
         if not should_log and not should_publish:
@@ -363,6 +372,7 @@ def checkout_failed(self, reason: str, error: str, start: float) -> None:
             )
 
     def checked_in(self, conn_id: int) -> None:
+        """Emit the connection checked-in log entry and APM event."""
         if self._should_publish:
             assert self._listeners is not None
             self._listeners.publish_connection_checked_in(self._address, conn_id)
@@ -522,13 +532,15 @@ def _emit_log(self, message: _SDAMStatusMessage, **extra: Any) -> None:
         )
 
     def topology_opened(self) -> None:
+        """Emit the topology opened log entry and APM event."""
         if self._should_log:
             self._emit_log(_SDAMStatusMessage.START_TOPOLOGY)
         if self._publish_tp:
             assert self._listeners is not None
             self._enqueue(self._listeners.publish_topology_opened, (self._topology_id,))
 
     def topology_description_changed(self, old_td: Any, new_td: Any) -> None:
+        """Emit the topology description changed APM event and log entry."""
         if self._publish_tp:
             assert self._listeners is not None
             self._enqueue(
@@ -560,6 +572,7 @@ def topology_closed(self, old_td: Any, new_td: Any) -> None:
             self._emit_log(_SDAMStatusMessage.STOP_TOPOLOGY)
 
     def server_opened(self, address: _Address) -> None:
+        """Emit the server opened log entry and APM event."""
         if self._publish_server:
             assert self._listeners is not None
             self._enqueue(self._listeners.publish_server_opened, (address, self._topology_id))
@@ -571,6 +584,7 @@ def server_opened(self, address: _Address) -> None:
             )
 
     def server_description_changed(self, sd_old: Any, sd_new: Any, address: _Address) -> None:
+        """Emit the server description changed APM event."""
         if self._publish_server:
             assert self._listeners is not None
             self._enqueue(
@@ -579,6 +593,7 @@ def server_description_changed(self, sd_old: Any, sd_new: Any, address: _Address
             )
 
     def server_closed(self, address: _Address) -> None:
+        """Emit the server closed log entry and APM event."""
         if self._publish_server:
             assert self._listeners is not None
             self._enqueue(self._listeners.publish_server_closed, (address, self._topology_id))
@@ -653,14 +668,12 @@ def waiting(self, remaining_time_ms: int) -> None:
                 remainingTimeMS=remaining_time_ms,
             )
 
-    def failed(self, failure: str, topology_description: Any = None) -> None:
+    def failed(self, failure: str, topology_description: Any) -> None:
         """Emit the server selection FAILED log entry with the current topology description."""
         if self._should_log:
             self._emit_log(
                 _ServerSelectionStatusMessage.FAILED,
-                topology_description
-                if topology_description is not None
-                else self._topology_description,
+                topology_description,
                 failure=failure,
             )
 

pymongo/asynchronous/pool.py

@@ -977,11 +977,11 @@ async def checkout(
 
         :param handler: A _MongoClientErrorHandler.
         """
-        start = self._telemetry.checkout_started()
+        checkout_started_time = self._telemetry.checkout_started()
 
-        conn = await self._get_conn(handler=handler, checkout_start=start)
+        conn = await self._get_conn(checkout_started_time, handler=handler)
 
-        self._telemetry.checkout_succeeded(conn.id, start)
+        self._telemetry.checkout_succeeded(conn.id, checkout_started_time)
         try:
             async with self.lock:
                 self.active_contexts.add(conn.cancel_context)
@@ -1012,13 +1012,13 @@ async def checkout(
         elif conn.active:
             await self.checkin(conn)
 
-    def _raise_if_not_ready(self, emit_event: bool, checkout_start: float) -> None:
+    def _raise_if_not_ready(self, checkout_started_time: float, emit_event: bool) -> 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,
+                    checkout_started_time,
                 )
 
             details = _get_timeout_details(self.opts)
@@ -1027,7 +1027,7 @@ def _raise_if_not_ready(self, emit_event: bool, checkout_start: float) -> None:
             )
 
     async def _get_conn(
-        self, handler: Optional[_MongoClientErrorHandler] = None, checkout_start: float = 0.0
+        self, checkout_started_time: float, handler: Optional[_MongoClientErrorHandler] = None
     ) -> AsyncConnection:
         """Get or create a AsyncConnection. Can raise ConnectionFailure."""
         # We use the pid here to avoid issues with fork / multiprocessing.
@@ -1040,7 +1040,7 @@ async def _get_conn(
             self._telemetry.checkout_failed(
                 "Connection pool was closed",
                 ConnectionCheckOutFailedReason.POOL_CLOSED,
-                checkout_start,
+                checkout_started_time,
             )
             raise _PoolClosedError(
                 "Attempted to check out a connection from closed connection pool"
@@ -1058,16 +1058,16 @@ async def _get_conn(
             deadline = None
 
         async with self.size_cond:
-            self._raise_if_not_ready(emit_event=True, checkout_start=checkout_start)
+            self._raise_if_not_ready(checkout_started_time, 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_start)
-                self._raise_if_not_ready(emit_event=True, checkout_start=checkout_start)
+                    self._raise_wait_queue_timeout(checkout_started_time)
+                self._raise_if_not_ready(checkout_started_time, emit_event=True)
             self.requests += 1
 
         # We've now acquired the semaphore and must release it on error.
@@ -1082,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, checkout_start=checkout_start)
+                    self._raise_if_not_ready(checkout_started_time, 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):
@@ -1091,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(checkout_start)
-                        self._raise_if_not_ready(emit_event=False, checkout_start=checkout_start)
+                            self._raise_wait_queue_timeout(checkout_started_time)
+                        self._raise_if_not_ready(checkout_started_time, emit_event=False)
 
                     try:
                         conn = self.conns.popleft()
@@ -1124,7 +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,
+                    checkout_started_time,
                 )
             raise
 
@@ -1216,11 +1216,11 @@ async def _perished(self, conn: AsyncConnection) -> bool:
 
         return False
 
-    def _raise_wait_queue_timeout(self, checkout_start: float) -> NoReturn:
+    def _raise_wait_queue_timeout(self, checkout_started_time: float) -> NoReturn:
         self._telemetry.checkout_failed(
             "Wait queue timeout elapsed without a connection becoming available",
             ConnectionCheckOutFailedReason.TIMEOUT,
-            checkout_start,
+            checkout_started_time,
         )
         timeout = _csot.get_timeout() or self.opts.wait_queue_timeout
         if self.opts.load_balanced:

pymongo/asynchronous/server.py

@@ -64,12 +64,8 @@ def __init__(
         self._pool = pool
         self._monitor = monitor
         self._topology_id = topology_id
-        self._publish = listeners is not None and listeners.enabled_for_server
-        self._listener = listeners
-        self._events = None
-        if self._publish:
-            self._events = events()  # type: ignore[misc]
-        self._sdam = _SdamTelemetry(topology_id, listeners, self._events)  # type: ignore[arg-type]
+        _events = events() if listeners is not None and listeners.enabled_for_server else None  # type: ignore[misc]
+        self._sdam = _SdamTelemetry(topology_id, listeners, _events)  # type: ignore[arg-type]
 
     async def open(self) -> None:
         """Start monitoring, or restart after a fork.

pymongo/asynchronous/topology.py

@@ -227,7 +227,7 @@ async def select_servers(
         address: Optional[_Address] = None,
         operation_id: Optional[int] = None,
         deprioritized_servers: Optional[list[Server]] = None,
-    ) -> list[Server]:
+    ) -> tuple[list[Server], _ServerSelectionTelemetry]:
         """Return a list of Servers matching selector, or time out.
 
         :param selector: function that takes a list of Servers and returns
@@ -253,7 +253,7 @@ async def select_servers(
             await self.cleanup_monitors()
 
         async with self._lock:
-            server_descriptions = await self._select_servers_loop(
+            server_descriptions, ss = await self._select_servers_loop(
                 selector,
                 server_timeout,
                 operation,
@@ -264,7 +264,7 @@ async def select_servers(
 
             return [
                 cast(Server, self.get_server_by_address(sd.address)) for sd in server_descriptions
-            ]
+            ], ss
 
     async def _select_servers_loop(
         self,
@@ -274,7 +274,7 @@ async def _select_servers_loop(
         operation_id: Optional[int],
         address: Optional[_Address],
         deprioritized_servers: Optional[list[Server]] = None,
-    ) -> list[ServerDescription]:
+    ) -> tuple[list[ServerDescription], _ServerSelectionTelemetry]:
         """select_servers() guts. Hold the lock when calling this."""
         now = time.monotonic()
         end_time = now + timeout
@@ -320,7 +320,7 @@ async def _select_servers_loop(
             )
 
         self._description.check_compatible()
-        return server_descriptions
+        return server_descriptions, ss
 
     async def _select_server(
         self,
@@ -330,8 +330,8 @@ async def _select_server(
         address: Optional[_Address] = None,
         deprioritized_servers: Optional[list[Server]] = None,
         operation_id: Optional[int] = None,
-    ) -> Server:
-        servers = await self.select_servers(
+    ) -> tuple[Server, _ServerSelectionTelemetry]:
+        servers, ss = await self.select_servers(
             selector,
             operation,
             server_selection_timeout,
@@ -340,12 +340,12 @@ async def _select_server(
             deprioritized_servers,
         )
         if len(servers) == 1:
-            return servers[0]
+            return servers[0], ss
         server1, server2 = random.sample(servers, 2)
         if server1.pool.operation_count <= server2.pool.operation_count:
-            return server1
+            return server1, ss
         else:
-            return server2
+            return server2, ss
 
     async def select_server(
         self,
@@ -357,7 +357,7 @@ async def select_server(
         operation_id: Optional[int] = None,
     ) -> Server:
         """Like select_servers, but choose a random server if several match."""
-        server = await self._select_server(
+        server, ss = await self._select_server(
             selector,
             operation,
             server_selection_timeout,
@@ -367,9 +367,7 @@ async def select_server(
         )
         if _csot.get_timeout():
             _csot.set_rtt(server.description.min_round_trip_time)
-        _ServerSelectionTelemetry(
-            self._topology_id, selector, operation, operation_id, self.description
-        ).succeeded(server.description.address[0], server.description.address[1])
+        ss.succeeded(server.description.address[0], server.description.address[1])
         return server
 
     async def select_server_by_address(