PYTHON-5846 Consolidate CMAP, heartbeat, and SDAM telemetry into _telemetry.py (#2907)

Co-authored-by: Noah Stapp <noah@noahstapp.com>
Co-authored-by: Copilot Autofix powered by AI <175728472+Copilot@users.noreply.github.com>

Author: 3 people

pymongo/_telemetry.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,
 )
@@ -2980,6 +2979,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
 
@@ -3003,13 +3011,7 @@ 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,
-                    )
+                    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:
@@ -3032,13 +3034,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:
-                _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,
-                )
+                self._log_retry(is_write=False)
             return await self._func(self._session, self._server, conn, read_pref)  # type: ignore
 
 

pymongo/asynchronous/mongo_client.py

@@ -18,18 +18,17 @@
 
 import asyncio
 import atexit
-import logging
 import time
 import weakref
 from typing import TYPE_CHECKING, Any, Optional
 
 from pymongo import common, periodic_executor
 from pymongo._csot import MovingMinimum
+from pymongo._telemetry import _HeartbeatTelemetry, _monotonic_duration, 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, _SDAMStatusMessage
 from pymongo.periodic_executor import _shutdown_executors
 from pymongo.pool_options import _is_faas
 from pymongo.read_preferences import MovingAverage
@@ -54,15 +53,6 @@ def _sanitize(error: Exception) -> None:
     error.__cause__ = None
 
 
-def _monotonic_duration(start: float) -> float:
-    """Return the duration since the given start time.
-
-    Accounts for buggy platforms where time.monotonic() is not monotonic.
-    See PYTHON-4600.
-    """
-    return max(0.0, time.monotonic() - start)
-
-
 class MonitorBase:
     def __init__(self, topology: Topology, name: str, interval: int, min_interval: float):
         """Base class to do periodic work on a background thread.
@@ -151,9 +141,10 @@ def __init__(
         self._pool = pool
         self._settings = topology_settings
         self._listeners = self._settings._pool_options._event_listeners
-        self._publish = self._listeners is not None and self._listeners.enabled_for_server_heartbeat
         self._cancel_context: Optional[_CancellationContext] = None
         self._conn_id: Optional[int] = None
+        self._current_hb: Optional[_HeartbeatTelemetry] = None
+        self._awaited: bool = False
         self._rtt_monitor = _RttMonitor(
             topology,
             topology_settings,
@@ -257,32 +248,17 @@ async def _check_server(self) -> ServerDescription:
         Returns a ServerDescription.
         """
         self._conn_id = None
-        start = time.monotonic()
+        self._current_hb = None
+        self._awaited = False
         try:
             return await self._check_once()
         except ReferenceError:
             raise
         except Exception as error:
             _sanitize(error)
-            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:
-                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,
-                    message=_SDAMStatusMessage.HEARTBEAT_FAIL,
-                    topologyId=self._topology._topology_id,
-                    serverHost=address[0],
-                    serverPort=address[1],
-                    awaited=awaited,
-                    durationMS=duration * 1000,
-                    failure=error,
-                    driverConnectionId=self._conn_id,
-                )
+            address = self._server_description.address
+            if self._current_hb is not None:
+                self._current_hb.failed(error, self._conn_id, self._awaited)
             await self._reset_connection()
             if isinstance(error, _OperationCancelled):
                 raise
@@ -300,28 +276,17 @@ async def _check_once(self) -> ServerDescription:
 
         # XXX: "awaited" could be incorrectly set to True in the rare case
         # the pool checkout closes and recreates a connection.
-        awaited = bool(
+        self._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
-            self._listeners.publish_server_heartbeat_started(address, awaited)
+        hb = _HeartbeatTelemetry(self._topology._topology_id, address, self._listeners)
+        self._current_hb = hb
+        hb.started(self._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,
-                    message=_SDAMStatusMessage.HEARTBEAT_START,
-                    topologyId=self._topology._topology_id,
-                    driverConnectionId=conn.id,
-                    serverConnectionId=conn.server_connection_id,
-                    serverHost=address[0],
-                    serverPort=address[1],
-                    awaited=awaited,
-                )
-
+            hb.emit_started_log(conn.id, conn.server_connection_id, self._awaited)
             self._cancel_context = conn.cancel_context
             # Record the connection id so we can later attach it to the failed log message.
             self._conn_id = conn.id
@@ -331,24 +296,7 @@ async def _check_once(self) -> ServerDescription:
 
             avg_rtt, min_rtt = await self._rtt_monitor.get()
             sd = ServerDescription(address, response, avg_rtt, min_round_trip_time=min_rtt)
-            if self._publish:
-                assert self._listeners is not None
-                self._listeners.publish_server_heartbeat_succeeded(
-                    address, round_trip_time, response, response.awaitable
-                )
-            if _SDAM_LOGGER.isEnabledFor(logging.DEBUG):
-                _debug_log(
-                    _SDAM_LOGGER,
-                    message=_SDAMStatusMessage.HEARTBEAT_SUCCESS,
-                    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,
-                )
+            hb.succeeded(round_trip_time, response, conn.id, conn.server_connection_id)
             return sd
 
     async def _check_with_socket(self, conn: AsyncConnection) -> tuple[Hello, float]:  # type: ignore[type-arg]
@@ -429,7 +377,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))