PYTHON-3472 - Add log messages to SDAM spec (mongodb#1771)

Co-authored-by: Jib <[email protected]>

Author: NoahStapp

pymongo/asynchronous/monitor.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import atexit
+import logging
 import time
 import weakref
 from typing import TYPE_CHECKING, Any, Mapping, Optional, cast
@@ -28,6 +29,7 @@
 from pymongo.errors import NetworkTimeout, NotPrimaryError, OperationFailure, _OperationCancelled
 from pymongo.hello import Hello
 from pymongo.lock import _create_lock
+from pymongo.logger import _SDAM_LOGGER, _debug_log, _SDAMStatusMessage
 from pymongo.pool_options import _is_faas
 from pymongo.read_preferences import MovingAverage
 from pymongo.server_description import ServerDescription
@@ -257,10 +259,21 @@ async def _check_server(self) -> ServerDescription:
             sd = self._server_description
             address = sd.address
             duration = _monotonic_duration(start)
+            awaited = bool(self._stream and sd.is_server_type_known and sd.topology_version)
             if self._publish:
-                awaited = bool(self._stream and sd.is_server_type_known and sd.topology_version)
                 assert self._listeners is not None
                 self._listeners.publish_server_heartbeat_failed(address, duration, error, awaited)
+            if _SDAM_LOGGER.isEnabledFor(logging.DEBUG):
+                _debug_log(
+                    _SDAM_LOGGER,
+                    topologyId=self._topology._topology_id,
+                    serverHost=address[0],
+                    serverPort=address[1],
+                    awaited=awaited,
+                    durationMS=duration * 1000,
+                    failure=error,
+                    message=_SDAMStatusMessage.HEARTBEAT_FAIL,
+                )
             await self._reset_connection()
             if isinstance(error, _OperationCancelled):
                 raise
@@ -274,22 +287,32 @@ async def _check_once(self) -> ServerDescription:
         Returns a ServerDescription, or raises an exception.
         """
         address = self._server_description.address
+        sd = self._server_description
+
+        # XXX: "awaited" could be incorrectly set to True in the rare case
+        # the pool checkout closes and recreates a connection.
+        awaited = bool(
+            self._pool.conns and self._stream and sd.is_server_type_known and sd.topology_version
+        )
         if self._publish:
             assert self._listeners is not None
-            sd = self._server_description
-            # XXX: "awaited" could be incorrectly set to True in the rare case
-            # the pool checkout closes and recreates a connection.
-            awaited = bool(
-                self._pool.conns
-                and self._stream
-                and sd.is_server_type_known
-                and sd.topology_version
-            )
             self._listeners.publish_server_heartbeat_started(address, awaited)
 
         if self._cancel_context and self._cancel_context.cancelled:
             await self._reset_connection()
         async with self._pool.checkout() as conn:
+            if _SDAM_LOGGER.isEnabledFor(logging.DEBUG):
+                _debug_log(
+                    _SDAM_LOGGER,
+                    topologyId=self._topology._topology_id,
+                    driverConnectionId=conn.id,
+                    serverConnectionId=conn.server_connection_id,
+                    serverHost=address[0],
+                    serverPort=address[1],
+                    awaited=awaited,
+                    message=_SDAMStatusMessage.HEARTBEAT_START,
+                )
+
             self._cancel_context = conn.cancel_context
             response, round_trip_time = await self._check_with_socket(conn)
             if not response.awaitable:
@@ -302,6 +325,19 @@ async def _check_once(self) -> ServerDescription:
                 self._listeners.publish_server_heartbeat_succeeded(
                     address, round_trip_time, response, response.awaitable
                 )
+            if _SDAM_LOGGER.isEnabledFor(logging.DEBUG):
+                _debug_log(
+                    _SDAM_LOGGER,
+                    topologyId=self._topology._topology_id,
+                    driverConnectionId=conn.id,
+                    serverConnectionId=conn.server_connection_id,
+                    serverHost=address[0],
+                    serverPort=address[1],
+                    awaited=awaited,
+                    durationMS=round_trip_time * 1000,
+                    reply=response.document,
+                    message=_SDAMStatusMessage.HEARTBEAT_SUCCESS,
+                )
             return sd
 
     async def _check_with_socket(self, conn: AsyncConnection) -> tuple[Hello, float]:

pymongo/asynchronous/server.py

@@ -30,7 +30,13 @@
 from pymongo.asynchronous.helpers import _handle_reauth
 from pymongo.errors import NotPrimaryError, OperationFailure
 from pymongo.helpers_shared import _check_command_response
-from pymongo.logger import _COMMAND_LOGGER, _CommandStatusMessage, _debug_log
+from pymongo.logger import (
+    _COMMAND_LOGGER,
+    _SDAM_LOGGER,
+    _CommandStatusMessage,
+    _debug_log,
+    _SDAMStatusMessage,
+)
 from pymongo.message import _convert_exception, _GetMore, _OpMsg, _Query
 from pymongo.response import PinnedResponse, Response
 
@@ -99,6 +105,15 @@ async def close(self) -> None:
                     (self._description.address, self._topology_id),
                 )
             )
+        if _SDAM_LOGGER.isEnabledFor(logging.DEBUG):
+            _debug_log(
+                _SDAM_LOGGER,
+                topologyId=self._topology_id,
+                serverHost=self._description.address[0],
+                serverPort=self._description.address[1],
+                message=_SDAMStatusMessage.STOP_SERVER,
+            )
+
         await self._monitor.close()
         await self._pool.close()
 

pymongo/asynchronous/topology.py

@@ -46,8 +46,10 @@
 from pymongo.hello import Hello
 from pymongo.lock import _ACondition, _ALock, _create_lock
 from pymongo.logger import (
+    _SDAM_LOGGER,
     _SERVER_SELECTION_LOGGER,
     _debug_log,
+    _SDAMStatusMessage,
     _ServerSelectionStatusMessage,
 )
 from pymongo.pool_options import PoolOptions
@@ -110,6 +112,13 @@ def __init__(self, topology_settings: TopologySettings):
         if self._publish_server or self._publish_tp:
             self._events = queue.Queue(maxsize=100)
 
+        if _SDAM_LOGGER.isEnabledFor(logging.DEBUG):
+            _debug_log(
+                _SDAM_LOGGER,
+                topologyId=self._topology_id,
+                message=_SDAMStatusMessage.START_TOPOLOGY,
+            )
+
         if self._publish_tp:
             assert self._events is not None
             self._events.put((self._listeners.publish_topology_opened, (self._topology_id,)))
@@ -124,22 +133,38 @@ def __init__(self, topology_settings: TopologySettings):
         )
 
         self._description = topology_description
+        initial_td = TopologyDescription(
+            TOPOLOGY_TYPE.Unknown, {}, None, None, None, self._settings
+        )
         if self._publish_tp:
             assert self._events is not None
-            initial_td = TopologyDescription(
-                TOPOLOGY_TYPE.Unknown, {}, None, None, None, self._settings
-            )
             self._events.put(
                 (
                     self._listeners.publish_topology_description_changed,
                     (initial_td, self._description, self._topology_id),
                 )
             )
+        if _SDAM_LOGGER.isEnabledFor(logging.DEBUG):
+            _debug_log(
+                _SDAM_LOGGER,
+                topologyId=self._topology_id,
+                previousDescription=initial_td,
+                newDescription=self._description,
+                message=_SDAMStatusMessage.TOPOLOGY_CHANGE,
+            )
 
         for seed in topology_settings.seeds:
             if self._publish_server:
                 assert self._events is not None
                 self._events.put((self._listeners.publish_server_opened, (seed, self._topology_id)))
+            if _SDAM_LOGGER.isEnabledFor(logging.DEBUG):
+                _debug_log(
+                    _SDAM_LOGGER,
+                    topologyId=self._topology_id,
+                    serverHost=seed[0],
+                    serverPort=seed[1],
+                    message=_SDAMStatusMessage.START_SERVER,
+                )
 
         # Store the seed list to help diagnose errors in _error_message().
         self._seed_addresses = list(topology_description.server_descriptions())
@@ -472,6 +497,14 @@ async def _process_change(
                     (td_old, self._description, self._topology_id),
                 )
             )
+        if _SDAM_LOGGER.isEnabledFor(logging.DEBUG):
+            _debug_log(
+                _SDAM_LOGGER,
+                topologyId=self._topology_id,
+                previousDescription=td_old,
+                newDescription=self._description,
+                message=_SDAMStatusMessage.TOPOLOGY_CHANGE,
+            )
 
         # Shutdown SRV polling for unsupported cluster types.
         # This is only applicable if the old topology was Unknown, and the
@@ -530,6 +563,14 @@ async def _process_srv_update(self, seedlist: list[tuple[str, Any]]) -> None:
                     (td_old, self._description, self._topology_id),
                 )
             )
+        if _SDAM_LOGGER.isEnabledFor(logging.DEBUG):
+            _debug_log(
+                _SDAM_LOGGER,
+                topologyId=self._topology_id,
+                previousDescription=td_old,
+                newDescription=self._description,
+                message=_SDAMStatusMessage.TOPOLOGY_CHANGE,
+            )
 
     async def on_srv_update(self, seedlist: list[tuple[str, Any]]) -> None:
         """Process a new list of nodes obtained from scanning SRV records."""
@@ -684,6 +725,18 @@ async def close(self) -> None:
                 )
             )
             self._events.put((self._listeners.publish_topology_closed, (self._topology_id,)))
+        if _SDAM_LOGGER.isEnabledFor(logging.DEBUG):
+            _debug_log(
+                _SDAM_LOGGER,
+                topologyId=self._topology_id,
+                previousDescription=old_td,
+                newDescription=self._description,
+                message=_SDAMStatusMessage.TOPOLOGY_CHANGE,
+            )
+            _debug_log(
+                _SDAM_LOGGER, topologyId=self._topology_id, message=_SDAMStatusMessage.STOP_TOPOLOGY
+            )
+
         if self._publish_server or self._publish_tp:
             # Make sure the events executor thread is fully closed before publishing the remaining events
             self.__events_executor.close()

pymongo/logger.py

@@ -53,6 +53,17 @@ class _ConnectionStatusMessage(str, enum.Enum):
     CHECKEDIN = "Connection checked in"
 
 
+class _SDAMStatusMessage(str, enum.Enum):
+    START_TOPOLOGY = "Starting topology monitoring"
+    STOP_TOPOLOGY = "Stopped topology monitoring"
+    START_SERVER = "Starting server monitoring"
+    STOP_SERVER = "Stopped server monitoring"
+    TOPOLOGY_CHANGE = "Topology description changed"
+    HEARTBEAT_START = "Server heartbeat started"
+    HEARTBEAT_SUCCESS = "Server heartbeat succeeded"
+    HEARTBEAT_FAIL = "Server heartbeat failed"
+
+
 _DEFAULT_DOCUMENT_LENGTH = 1000
 _SENSITIVE_COMMANDS = [
     "authenticate",
@@ -73,6 +84,7 @@ class _ConnectionStatusMessage(str, enum.Enum):
 _CONNECTION_LOGGER = logging.getLogger("pymongo.connection")
 _SERVER_SELECTION_LOGGER = logging.getLogger("pymongo.serverSelection")
 _CLIENT_LOGGER = logging.getLogger("pymongo.client")
+_SDAM_LOGGER = logging.getLogger("pymongo.topology")
 _VERBOSE_CONNECTION_ERROR_REASONS = {
     ConnectionClosedReason.POOL_CLOSED: "Connection pool was closed",
     ConnectionCheckOutFailedReason.POOL_CLOSED: "Connection pool was closed",
@@ -129,7 +141,7 @@ def _is_sensitive(self, doc_name: str) -> bool:
         )
 
         is_sensitive_hello = (
-            self._kwargs["commandName"] in _HELLO_COMMANDS and is_speculative_authenticate
+            self._kwargs.get("commandName", None) in _HELLO_COMMANDS and is_speculative_authenticate
         )
 
         return is_sensitive_command or is_sensitive_hello

pymongo/synchronous/monitor.py

@@ -17,6 +17,7 @@
 from __future__ import annotations
 
 import atexit
+import logging
 import time
 import weakref
 from typing import TYPE_CHECKING, Any, Mapping, Optional, cast
@@ -26,6 +27,7 @@
 from pymongo.errors import NetworkTimeout, NotPrimaryError, OperationFailure, _OperationCancelled
 from pymongo.hello import Hello
 from pymongo.lock import _create_lock
+from pymongo.logger import _SDAM_LOGGER, _debug_log, _SDAMStatusMessage
 from pymongo.pool_options import _is_faas
 from pymongo.read_preferences import MovingAverage
 from pymongo.server_description import ServerDescription
@@ -257,10 +259,21 @@ def _check_server(self) -> ServerDescription:
             sd = self._server_description
             address = sd.address
             duration = _monotonic_duration(start)
+            awaited = bool(self._stream and sd.is_server_type_known and sd.topology_version)
             if self._publish:
-                awaited = bool(self._stream and sd.is_server_type_known and sd.topology_version)
                 assert self._listeners is not None
                 self._listeners.publish_server_heartbeat_failed(address, duration, error, awaited)
+            if _SDAM_LOGGER.isEnabledFor(logging.DEBUG):
+                _debug_log(
+                    _SDAM_LOGGER,
+                    topologyId=self._topology._topology_id,
+                    serverHost=address[0],
+                    serverPort=address[1],
+                    awaited=awaited,
+                    durationMS=duration * 1000,
+                    failure=error,
+                    message=_SDAMStatusMessage.HEARTBEAT_FAIL,
+                )
             self._reset_connection()
             if isinstance(error, _OperationCancelled):
                 raise
@@ -274,22 +287,32 @@ def _check_once(self) -> ServerDescription:
         Returns a ServerDescription, or raises an exception.
         """
         address = self._server_description.address
+        sd = self._server_description
+
+        # XXX: "awaited" could be incorrectly set to True in the rare case
+        # the pool checkout closes and recreates a connection.
+        awaited = bool(
+            self._pool.conns and self._stream and sd.is_server_type_known and sd.topology_version
+        )
         if self._publish:
             assert self._listeners is not None
-            sd = self._server_description
-            # XXX: "awaited" could be incorrectly set to True in the rare case
-            # the pool checkout closes and recreates a connection.
-            awaited = bool(
-                self._pool.conns
-                and self._stream
-                and sd.is_server_type_known
-                and sd.topology_version
-            )
             self._listeners.publish_server_heartbeat_started(address, awaited)
 
         if self._cancel_context and self._cancel_context.cancelled:
             self._reset_connection()
         with self._pool.checkout() as conn:
+            if _SDAM_LOGGER.isEnabledFor(logging.DEBUG):
+                _debug_log(
+                    _SDAM_LOGGER,
+                    topologyId=self._topology._topology_id,
+                    driverConnectionId=conn.id,
+                    serverConnectionId=conn.server_connection_id,
+                    serverHost=address[0],
+                    serverPort=address[1],
+                    awaited=awaited,
+                    message=_SDAMStatusMessage.HEARTBEAT_START,
+                )
+
             self._cancel_context = conn.cancel_context
             response, round_trip_time = self._check_with_socket(conn)
             if not response.awaitable:
@@ -302,6 +325,19 @@ def _check_once(self) -> ServerDescription:
                 self._listeners.publish_server_heartbeat_succeeded(
                     address, round_trip_time, response, response.awaitable
                 )
+            if _SDAM_LOGGER.isEnabledFor(logging.DEBUG):
+                _debug_log(
+                    _SDAM_LOGGER,
+                    topologyId=self._topology._topology_id,
+                    driverConnectionId=conn.id,
+                    serverConnectionId=conn.server_connection_id,
+                    serverHost=address[0],
+                    serverPort=address[1],
+                    awaited=awaited,
+                    durationMS=round_trip_time * 1000,
+                    reply=response.document,
+                    message=_SDAMStatusMessage.HEARTBEAT_SUCCESS,
+                )
             return sd
 
     def _check_with_socket(self, conn: Connection) -> tuple[Hello, float]: