PYTHON-5846 Refine _CmapTelemetry and _HeartbeatTelemetry API

- Rename _CmapTelemetry._log -> _emit_log for consistency with _CommandTelemetry
- Rename _HeartbeatTelemetry.apm_started -> started to match started/succeeded/failed lifecycle
- Rename _HeartbeatTelemetry.log_started -> emit_started_log to signal it is the deferred log-only half
- Replace is_sdam flag with separate publish/log bool parameters on _CmapTelemetry

Author: blink1073

pymongo/_telemetry.py

@@ -205,15 +205,16 @@ def __init__(
         client_id: Optional[ObjectId],
         address: _Address,
         listeners: Optional[_EventListeners],
-        is_sdam: bool,
+        publish: bool,
+        log: bool,
     ) -> None:
         self._client_id = client_id
         self._address = address
         self._listeners = listeners
-        self._should_publish = not is_sdam and listeners is not None and listeners.enabled_for_cmap
-        self._should_log = not is_sdam
+        self._should_publish = publish and listeners is not None and listeners.enabled_for_cmap
+        self._should_log = log
 
-    def _log(self, message: _ConnectionStatusMessage, **extra: Any) -> None:
+    def _emit_log(self, message: _ConnectionStatusMessage, **extra: Any) -> None:
         if self._should_log and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
             _debug_log(
                 _CONNECTION_LOGGER,
@@ -226,21 +227,21 @@ def _log(self, message: _ConnectionStatusMessage, **extra: Any) -> None:
 
     def pool_created(self, non_default_options: dict[str, Any]) -> None:
         # Log before publishing to prevent potential listener preemption in tests.
-        self._log(_ConnectionStatusMessage.POOL_CREATED, **non_default_options)
+        self._emit_log(_ConnectionStatusMessage.POOL_CREATED, **non_default_options)
         if self._should_publish:
             assert self._listeners is not None
             self._listeners.publish_pool_created(self._address, non_default_options)
 
     def pool_ready(self) -> None:
         # Log before publishing to prevent potential listener preemption in tests.
-        self._log(_ConnectionStatusMessage.POOL_READY)
+        self._emit_log(_ConnectionStatusMessage.POOL_READY)
         if self._should_publish:
             assert self._listeners is not None
             self._listeners.publish_pool_ready(self._address)
 
     def pool_cleared(self, service_id: Optional[ObjectId], interrupt_connections: bool) -> None:
         # Log before publishing to prevent potential listener preemption in tests.
-        self._log(_ConnectionStatusMessage.POOL_CLEARED, serviceId=service_id)
+        self._emit_log(_ConnectionStatusMessage.POOL_CLEARED, serviceId=service_id)
         if self._should_publish:
             assert self._listeners is not None
             self._listeners.publish_pool_cleared(
@@ -251,21 +252,21 @@ def pool_cleared(self, service_id: Optional[ObjectId], interrupt_connections: bo
 
     def pool_closed(self) -> None:
         # Log before publishing to prevent potential listener preemption in tests.
-        self._log(_ConnectionStatusMessage.POOL_CLOSED)
+        self._emit_log(_ConnectionStatusMessage.POOL_CLOSED)
         if self._should_publish:
             assert self._listeners is not None
             self._listeners.publish_pool_closed(self._address)
 
     def connection_created(self, conn_id: int) -> None:
         # Log before publishing to prevent potential listener preemption in tests.
-        self._log(_ConnectionStatusMessage.CONN_CREATED, driverConnectionId=conn_id)
+        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, duration: float) -> None:
         # Log before publishing to prevent potential listener preemption in tests.
-        self._log(
+        self._emit_log(
             _ConnectionStatusMessage.CONN_READY,
             driverConnectionId=conn_id,
             durationMS=duration,
@@ -278,7 +279,7 @@ def connection_closed(self, conn_id: int, reason: str) -> None:
         if self._should_publish:
             assert self._listeners is not None
             self._listeners.publish_connection_closed(self._address, conn_id, reason)
-        self._log(
+        self._emit_log(
             _ConnectionStatusMessage.CONN_CLOSED,
             driverConnectionId=conn_id,
             reason=_verbose_connection_error_reason(reason),
@@ -289,13 +290,13 @@ def checkout_started(self) -> None:
         if self._should_publish:
             assert self._listeners is not None
             self._listeners.publish_connection_check_out_started(self._address)
-        self._log(_ConnectionStatusMessage.CHECKOUT_STARTED)
+        self._emit_log(_ConnectionStatusMessage.CHECKOUT_STARTED)
 
     def checkout_succeeded(self, conn_id: int, duration: float) -> None:
         if self._should_publish:
             assert self._listeners is not None
             self._listeners.publish_connection_checked_out(self._address, conn_id, duration)
-        self._log(
+        self._emit_log(
             _ConnectionStatusMessage.CHECKOUT_SUCCEEDED,
             driverConnectionId=conn_id,
             durationMS=duration,
@@ -305,7 +306,7 @@ def checkout_failed(self, reason: str, error: str, duration: float) -> None:
         if self._should_publish:
             assert self._listeners is not None
             self._listeners.publish_connection_check_out_failed(self._address, error, duration)
-        self._log(
+        self._emit_log(
             _ConnectionStatusMessage.CHECKOUT_FAILED,
             reason=reason,
             error=error,
@@ -316,7 +317,7 @@ def checked_in(self, conn_id: int) -> None:
         if self._should_publish:
             assert self._listeners is not None
             self._listeners.publish_connection_checked_in(self._address, conn_id)
-        self._log(_ConnectionStatusMessage.CHECKEDIN, driverConnectionId=conn_id)
+        self._emit_log(_ConnectionStatusMessage.CHECKEDIN, driverConnectionId=conn_id)
 
 
 class _HeartbeatTelemetry:
@@ -344,13 +345,13 @@ def __init__(
         self._publish = publish
         self._awaited = awaited
 
-    def apm_started(self) -> None:
+    def started(self) -> None:
         """Publish the APM heartbeat-started event (before connection checkout)."""
         if self._publish:
             assert self._listeners is not None
             self._listeners.publish_server_heartbeat_started(self._address, self._awaited)
 
-    def log_started(self, conn_id: int, server_conn_id: Optional[int]) -> None:
+    def emit_started_log(self, conn_id: int, server_conn_id: Optional[int]) -> None:
         """Emit the log entry for heartbeat started (after connection checkout)."""
         if _SDAM_LOGGER.isEnabledFor(logging.DEBUG):
             _debug_log(

pymongo/asynchronous/monitor.py

@@ -294,12 +294,12 @@ async def _check_once(self) -> ServerDescription:
             self._topology._topology_id, address, self._listeners, self._publish, awaited
         )
         self._current_hb = hb
-        hb.apm_started()
+        hb.started()
 
         if self._cancel_context and self._cancel_context.cancelled:
             await self._reset_connection()
         async with self._pool.checkout() as conn:
-            hb.log_started(conn.id, conn.server_connection_id)
+            hb.emit_started_log(conn.id, conn.server_connection_id)
             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

pymongo/asynchronous/pool.py

@@ -683,7 +683,9 @@ def __init__(
         self._ssl_session_cache: Optional[list[Any]] = (
             [None] if self.opts._ssl_context is not None else None
         )
-        self._telemetry = _CmapTelemetry(client_id, address, options._event_listeners, is_sdam)
+        self._telemetry = _CmapTelemetry(
+            client_id, address, options._event_listeners, publish=not is_sdam, log=not is_sdam
+        )
         self._telemetry.pool_created(self.opts.non_default_options)
         # Similar to active_sockets but includes threads in the wait queue.
         self.operation_count: int = 0

pymongo/synchronous/monitor.py

@@ -292,12 +292,12 @@ def _check_once(self) -> ServerDescription:
             self._topology._topology_id, address, self._listeners, self._publish, awaited
         )
         self._current_hb = hb
-        hb.apm_started()
+        hb.started()
 
         if self._cancel_context and self._cancel_context.cancelled:
             self._reset_connection()
         with self._pool.checkout() as conn:
-            hb.log_started(conn.id, conn.server_connection_id)
+            hb.emit_started_log(conn.id, conn.server_connection_id)
             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

pymongo/synchronous/pool.py

@@ -681,7 +681,9 @@ def __init__(
         self._ssl_session_cache: Optional[list[Any]] = (
             [None] if self.opts._ssl_context is not None else None
         )
-        self._telemetry = _CmapTelemetry(client_id, address, options._event_listeners, is_sdam)
+        self._telemetry = _CmapTelemetry(
+            client_id, address, options._event_listeners, publish=not is_sdam, log=not is_sdam
+        )
         self._telemetry.pool_created(self.opts.non_default_options)
         # Similar to active_sockets but includes threads in the wait queue.
         self.operation_count: int = 0