PYTHON-5846 Consolidate server selection logging and retry logging into _telemetry.py

Author: blink1073

pymongo/_telemetry.py

@@ -27,10 +27,12 @@
     _COMMAND_LOGGER,
     _CONNECTION_LOGGER,
     _SDAM_LOGGER,
+    _SERVER_SELECTION_LOGGER,
     _CommandStatusMessage,
     _ConnectionStatusMessage,
     _debug_log,
     _SDAMStatusMessage,
+    _ServerSelectionStatusMessage,
     _verbose_connection_error_reason,
 )
 from pymongo.pool_shared import _ConnectionTelemetryInfo
@@ -543,3 +545,91 @@ def server_closed(self, address: _Address) -> None:
             serverHost=address[0],
             serverPort=address[1],
         )
+
+
+class _ServerSelectionTelemetry:
+    """Structured logging for server selection events.
+
+    The server selection spec defines only log entries, not APM events, so this
+    class has no publish methods.
+
+    Construct once per :meth:`select_server` call.
+    """
+
+    __slots__ = (
+        "_operation",
+        "_operation_id",
+        "_selector",
+        "_topology_description",
+        "_topology_id",
+    )
+
+    def __init__(
+        self,
+        topology_id: Any,
+        selector: Any,
+        operation: str,
+        operation_id: Optional[int],
+        topology_description: Any,
+    ) -> None:
+        self._topology_id = topology_id
+        self._selector = selector
+        self._operation = operation
+        self._operation_id = operation_id
+        self._topology_description = topology_description
+
+    @property
+    def _should_log(self) -> bool:
+        return _SERVER_SELECTION_LOGGER.isEnabledFor(logging.DEBUG)
+
+    def _emit_log(self, message: _ServerSelectionStatusMessage, **extra: Any) -> None:
+        if self._should_log:
+            _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,
+                **extra,
+            )
+
+    def started(self) -> None:
+        """Emit the server selection STARTED log entry."""
+        self._emit_log(_ServerSelectionStatusMessage.STARTED)
+
+    def waiting(self, remaining_time_ms: int) -> None:
+        """Emit the server selection WAITING log entry."""
+        self._emit_log(_ServerSelectionStatusMessage.WAITING, remainingTimeMS=remaining_time_ms)
+
+    def failed(self, failure: str) -> None:
+        """Emit the server selection FAILED log entry."""
+        self._emit_log(_ServerSelectionStatusMessage.FAILED, failure=failure)
+
+    def succeeded(self, server_host: str, server_port: Optional[int]) -> None:
+        """Emit the server selection SUCCEEDED log entry."""
+        self._emit_log(
+            _ServerSelectionStatusMessage.SUCCEEDED,
+            serverHost=server_host,
+            serverPort=server_port,
+        )
+
+
+def log_command_retry(
+    topology_id: Any,
+    command_name: str,
+    operation_id: Optional[int],
+    attempt_number: int,
+    is_write: bool,
+) -> None:
+    """Emit a command-retry log entry."""
+    if _COMMAND_LOGGER.isEnabledFor(logging.DEBUG):
+        op = "write" if is_write else "read"
+        _debug_log(
+            _COMMAND_LOGGER,
+            message=f"Retrying {op} attempt number {attempt_number}",
+            clientId=topology_id,
+            commandName=command_name,
+            operationId=operation_id,
+        )

pymongo/asynchronous/mongo_client.py

@@ -57,6 +57,7 @@
 from bson.codec_options import DEFAULT_CODEC_OPTIONS, CodecOptions, TypeRegistry
 from bson.timestamp import Timestamp
 from pymongo import _csot, common, helpers_shared, periodic_executor
+from pymongo._telemetry import log_command_retry
 from pymongo.asynchronous import client_session, database, uri_parser
 from pymongo.asynchronous.change_stream import AsyncChangeStream, AsyncClusterChangeStream
 from pymongo.asynchronous.client_bulk import _AsyncClientBulk
@@ -90,8 +91,6 @@
 )
 from pymongo.logger import (
     _CLIENT_LOGGER,
-    _COMMAND_LOGGER,
-    _debug_log,
     _log_client_error,
     _log_or_warn,
 )
@@ -3014,12 +3013,12 @@ async def _write(self) -> T:
                     self._check_last_error()
                     self._retryable = False
                 if self._retrying:
-                    _debug_log(
-                        _COMMAND_LOGGER,
-                        message=f"Retrying write attempt number {self._attempt_number}",
-                        clientId=self._client._topology_id,
-                        commandName=self._operation,
-                        operationId=self._operation_id,
+                    log_command_retry(
+                        self._client._topology_id,
+                        self._operation,
+                        self._operation_id,
+                        self._attempt_number,
+                        is_write=True,
                     )
                 return await self._func(self._session, conn, self._retryable)  # type: ignore
         except PyMongoError as exc:
@@ -3043,12 +3042,12 @@ async def _read(self) -> T:
             if self._retrying and not self._retryable and not self._always_retryable:
                 self._check_last_error()
             if self._retrying:
-                _debug_log(
-                    _COMMAND_LOGGER,
-                    message=f"Retrying read attempt number {self._attempt_number}",
-                    clientId=self._client._topology_settings._topology_id,
-                    commandName=self._operation,
-                    operationId=self._operation_id,
+                log_command_retry(
+                    self._client._topology_settings._topology_id,
+                    self._operation,
+                    self._operation_id,
+                    self._attempt_number,
+                    is_write=False,
                 )
             return await self._func(self._session, self._server, conn, read_pref)  # type: ignore
 

pymongo/asynchronous/topology.py

@@ -17,7 +17,6 @@
 from __future__ import annotations
 
 import asyncio
-import logging
 import os
 import queue
 import random
@@ -30,7 +29,7 @@
 from typing import TYPE_CHECKING, Any, Callable, Optional, cast
 
 from pymongo import _csot, common, helpers_shared, periodic_executor
-from pymongo._telemetry import _SdamTelemetry
+from pymongo._telemetry import _SdamTelemetry, _ServerSelectionTelemetry
 from pymongo.asynchronous.client_session import _ServerSession, _ServerSessionPool
 from pymongo.asynchronous.monitor import MonitorBase, SrvMonitor
 from pymongo.asynchronous.pool import Pool
@@ -52,11 +51,6 @@
     _async_create_condition,
     _async_create_lock,
 )
-from pymongo.logger import (
-    _SERVER_SELECTION_LOGGER,
-    _debug_log,
-    _ServerSelectionStatusMessage,
-)
 from pymongo.pool_options import PoolOptions
 from pymongo.server_description import ServerDescription
 from pymongo.server_selectors import (
@@ -107,14 +101,14 @@ class Topology:
     def __init__(self, topology_settings: TopologySettings):
         self._topology_id = topology_settings._topology_id
         self._listeners = topology_settings._pool_options._event_listeners
-        self._publish_server = self._listeners is not None and self._listeners.enabled_for_server
-        self._publish_tp = self._listeners is not None and self._listeners.enabled_for_topology
 
         # Create events queue if there are publishers.
         self._events: queue.Queue[Any] | None = None
         self.__events_executor: Any = None
 
-        if self._publish_server or self._publish_tp:
+        publish_server = self._listeners is not None and self._listeners.enabled_for_server
+        publish_tp = self._listeners is not None and self._listeners.enabled_for_topology
+        if publish_server or publish_tp:
             self._events = queue.Queue(maxsize=100)
 
         self._sdam = _SdamTelemetry(self._topology_id, self._listeners, self._events)
@@ -152,7 +146,7 @@ def __init__(self, topology_settings: TopologySettings):
         self._max_cluster_time: Optional[ClusterTime] = None
         self._session_pool = _ServerSessionPool()
 
-        if self._publish_server or self._publish_tp:
+        if self._sdam._publish_server or self._sdam._publish_tp:
             assert self._events is not None
             weak: weakref.ReferenceType[queue.Queue[Any]]
 
@@ -285,17 +279,10 @@ async def _select_servers_loop(
         now = time.monotonic()
         end_time = now + timeout
         logged_waiting = False
-
-        if _SERVER_SELECTION_LOGGER.isEnabledFor(logging.DEBUG):
-            _debug_log(
-                _SERVER_SELECTION_LOGGER,
-                message=_ServerSelectionStatusMessage.STARTED,
-                selector=selector,
-                operation=operation,
-                operationId=operation_id,
-                topologyDescription=self.description,
-                clientId=self.description._topology_settings._topology_id,
-            )
+        ss = _ServerSelectionTelemetry(
+            self._topology_id, selector, operation, operation_id, self.description
+        )
+        ss.started()
 
         server_descriptions = self._description.apply_selector(
             selector,
@@ -309,32 +296,13 @@ async def _select_servers_loop(
         while not server_descriptions:
             # No suitable servers.
             if timeout == 0 or now > end_time:
-                if _SERVER_SELECTION_LOGGER.isEnabledFor(logging.DEBUG):
-                    _debug_log(
-                        _SERVER_SELECTION_LOGGER,
-                        message=_ServerSelectionStatusMessage.FAILED,
-                        selector=selector,
-                        operation=operation,
-                        operationId=operation_id,
-                        topologyDescription=self.description,
-                        clientId=self.description._topology_settings._topology_id,
-                        failure=self._error_message(selector),
-                    )
+                ss.failed(self._error_message(selector))
                 raise ServerSelectionTimeoutError(
                     f"{self._error_message(selector)}, Timeout: {timeout}s, Topology Description: {self.description!r}"
                 )
 
             if not logged_waiting:
-                _debug_log(
-                    _SERVER_SELECTION_LOGGER,
-                    message=_ServerSelectionStatusMessage.WAITING,
-                    selector=selector,
-                    operation=operation,
-                    operationId=operation_id,
-                    topologyDescription=self.description,
-                    clientId=self.description._topology_settings._topology_id,
-                    remainingTimeMS=int(1000 * (end_time - time.monotonic())),
-                )
+                ss.waiting(int(1000 * (end_time - time.monotonic())))
                 logged_waiting = True
 
             await self._ensure_opened()
@@ -399,18 +367,9 @@ async def select_server(
         )
         if _csot.get_timeout():
             _csot.set_rtt(server.description.min_round_trip_time)
-        if _SERVER_SELECTION_LOGGER.isEnabledFor(logging.DEBUG):
-            _debug_log(
-                _SERVER_SELECTION_LOGGER,
-                message=_ServerSelectionStatusMessage.SUCCEEDED,
-                selector=selector,
-                operation=operation,
-                operationId=operation_id,
-                topologyDescription=self.description,
-                clientId=self.description._topology_settings._topology_id,
-                serverHost=server.description.address[0],
-                serverPort=server.description.address[1],
-            )
+        _ServerSelectionTelemetry(
+            self._topology_id, selector, operation, operation_id, self.description
+        ).succeeded(server.description.address[0], server.description.address[1])
         return server
 
     async def select_server_by_address(
@@ -670,7 +629,7 @@ async def close(self) -> None:
             self._closed = True
 
         # Publish only after releasing the lock.
-        if self._publish_tp:
+        if self._sdam._publish_tp:
             self._description = TopologyDescription(
                 TOPOLOGY_TYPE.Unknown,
                 {},
@@ -681,7 +640,7 @@ async def close(self) -> None:
             )
         self._sdam.topology_closed(old_td, self._description)
 
-        if self._publish_server or self._publish_tp:
+        if self._sdam._publish_server or self._sdam._publish_tp:
             # Make sure the events executor thread is fully closed before publishing the remaining events
             self.__events_executor.close()
             await self.__events_executor.join(1)
@@ -722,7 +681,7 @@ async def _ensure_opened(self) -> None:
             await self._update_servers()
 
             # Start or restart the events publishing thread.
-            if self._publish_tp or self._publish_server:
+            if self._sdam._publish_tp or self._sdam._publish_server:
                 self.__events_executor.open()
 
             # Start the SRV polling thread.
@@ -861,7 +820,7 @@ async def _update_servers(self) -> None:
                 )
 
                 weak = None
-                if self._publish_server and self._events is not None:
+                if self._sdam._publish_server and self._events is not None:
                     weak = weakref.ref(self._events)
                 server = Server(
                     server_description=sd,

pymongo/synchronous/mongo_client.py

@@ -57,6 +57,7 @@
 from bson.codec_options import DEFAULT_CODEC_OPTIONS, CodecOptions, TypeRegistry
 from bson.timestamp import Timestamp
 from pymongo import _csot, common, helpers_shared, periodic_executor
+from pymongo._telemetry import log_command_retry
 from pymongo.client_options import ClientOptions
 from pymongo.driver_info import DriverInfo
 from pymongo.errors import (
@@ -80,8 +81,6 @@
 )
 from pymongo.logger import (
     _CLIENT_LOGGER,
-    _COMMAND_LOGGER,
-    _debug_log,
     _log_client_error,
     _log_or_warn,
 )
@@ -3005,12 +3004,12 @@ def _write(self) -> T:
                     self._check_last_error()
                     self._retryable = False
                 if self._retrying:
-                    _debug_log(
-                        _COMMAND_LOGGER,
-                        message=f"Retrying write attempt number {self._attempt_number}",
-                        clientId=self._client._topology_id,
-                        commandName=self._operation,
-                        operationId=self._operation_id,
+                    log_command_retry(
+                        self._client._topology_id,
+                        self._operation,
+                        self._operation_id,
+                        self._attempt_number,
+                        is_write=True,
                     )
                 return self._func(self._session, conn, self._retryable)  # type: ignore
         except PyMongoError as exc:
@@ -3034,12 +3033,12 @@ def _read(self) -> T:
             if self._retrying and not self._retryable and not self._always_retryable:
                 self._check_last_error()
             if self._retrying:
-                _debug_log(
-                    _COMMAND_LOGGER,
-                    message=f"Retrying read attempt number {self._attempt_number}",
-                    clientId=self._client._topology_settings._topology_id,
-                    commandName=self._operation,
-                    operationId=self._operation_id,
+                log_command_retry(
+                    self._client._topology_settings._topology_id,
+                    self._operation,
+                    self._operation_id,
+                    self._attempt_number,
+                    is_write=False,
                 )
             return self._func(self._session, self._server, conn, read_pref)  # type: ignore