diff --git a/pymongo/_telemetry.py b/pymongo/_telemetry.py
index 471a013eb9..14bb866316 100644
--- a/pymongo/_telemetry.py
+++ b/pymongo/_telemetry.py
@@ -18,16 +18,41 @@
 
 import datetime
 import logging
+import queue
+import time
 from collections.abc import MutableMapping
 from typing import TYPE_CHECKING, Any, Optional
 
-from pymongo.logger import _COMMAND_LOGGER, _CommandStatusMessage, _debug_log
+from pymongo.logger import (
+    _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
 
 if TYPE_CHECKING:
     from bson.objectid import ObjectId
+    from pymongo.hello import Hello
     from pymongo.monitoring import _EventListeners
-    from pymongo.typings import _DocumentOut
+    from pymongo.server_description import ServerDescription
+    from pymongo.topology_description import TopologyDescription
+    from pymongo.typings import _Address, _DocumentOut
+
+
+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 _CommandTelemetry:
@@ -183,3 +208,537 @@ def failed(
                 service_id=self._conn.service_id,
                 database_name=self._dbname,
             )
+
+
+class _CmapTelemetry:
+    """Combines CMAP structured logging and APM event publishing for pool and connection events."""
+
+    __slots__ = (
+        "_address",
+        "_client_id",
+        "_listeners",
+        "_log",
+        "_publish",
+    )
+
+    def __init__(
+        self,
+        client_id: Optional[ObjectId],
+        address: _Address,
+        listeners: Optional[_EventListeners],
+        publish: bool,
+        log: bool,
+    ) -> None:
+        self._client_id = client_id
+        self._address = address
+        self._listeners = listeners
+        self._publish = publish
+        self._log = log
+
+    @property
+    def _should_publish(self) -> bool:
+        """Computed per-call because listener registration can change while the pool is open."""
+        return self._publish and self._listeners is not None and self._listeners.enabled_for_cmap
+
+    @property
+    def _should_log(self) -> bool:
+        """Computed per-call because logging level can be reconfigured at runtime."""
+        return self._log and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG)
+
+    def _emit_log(self, message: _ConnectionStatusMessage, **extra: Any) -> None:
+        _debug_log(
+            _CONNECTION_LOGGER,
+            message=message,
+            clientId=self._client_id,
+            serverHost=self._address[0],
+            serverPort=self._address[1],
+            **extra,
+        )
+
+    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)
+        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:
+        """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)
+        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:
+        """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)
+        if self._should_publish:
+            assert self._listeners is not None
+            self._listeners.publish_pool_cleared(
+                self._address,
+                service_id=service_id,
+                interrupt_connections=interrupt_connections,
+            )
+
+    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)
+        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:
+        """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)
+        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, 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:
+            return
+        duration = _monotonic_duration(creation_time)
+        # Log before publishing to prevent potential listener preemption in tests.
+        if should_log:
+            self._emit_log(
+                _ConnectionStatusMessage.CONN_READY,
+                driverConnectionId=conn_id,
+                durationMS=duration * 1000,
+            )
+        if should_publish:
+            assert self._listeners is not 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:
+            assert self._listeners is not None
+            self._listeners.publish_connection_closed(self._address, conn_id, reason)
+        if should_log:
+            self._emit_log(
+                _ConnectionStatusMessage.CONN_CLOSED,
+                driverConnectionId=conn_id,
+                reason=_verbose_connection_error_reason(reason),
+                error=reason,
+            )
+
+    def checkout_started(self) -> float:
+        """Emit the checkout started event/log and return the start time for duration tracking."""
+        start = time.monotonic()
+        if self._should_publish:
+            assert self._listeners is not None
+            self._listeners.publish_connection_check_out_started(self._address)
+        if self._should_log:
+            self._emit_log(_ConnectionStatusMessage.CHECKOUT_STARTED)
+        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:
+            return
+        duration = _monotonic_duration(start)
+        if should_publish:
+            assert self._listeners is not None
+            self._listeners.publish_connection_checked_out(self._address, conn_id, duration)
+        if should_log:
+            self._emit_log(
+                _ConnectionStatusMessage.CHECKOUT_SUCCEEDED,
+                driverConnectionId=conn_id,
+                durationMS=duration * 1000,
+            )
+
+    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:
+            return
+        duration = _monotonic_duration(start)
+        if should_publish:
+            assert self._listeners is not None
+            self._listeners.publish_connection_check_out_failed(self._address, error, duration)
+        if should_log:
+            self._emit_log(
+                _ConnectionStatusMessage.CHECKOUT_FAILED,
+                reason=reason,
+                error=error,
+                durationMS=duration * 1000,
+            )
+
+    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)
+        if self._should_log:
+            self._emit_log(_ConnectionStatusMessage.CHECKEDIN, driverConnectionId=conn_id)
+
+
+class _HeartbeatTelemetry:
+    """Combines SDAM structured logging and APM event publishing for server heartbeats.
+
+    The APM started event is published before connection checkout (no conn_id yet);
+    the log entry for started is emitted after checkout once the conn_id is known.
+    Call :meth:`started` first, then :meth:`emit_started_log` inside the checkout
+    context, then :meth:`succeeded` or :meth:`failed` when the outcome is known.
+    """
+
+    __slots__ = (
+        "_address",
+        "_listeners",
+        "_should_log",
+        "_should_publish",
+        "_start",
+        "_topology_id",
+    )
+
+    def __init__(
+        self,
+        topology_id: ObjectId,
+        address: _Address,
+        listeners: Optional[_EventListeners],
+    ) -> None:
+        self._topology_id = topology_id
+        self._address = address
+        self._listeners = listeners
+        # Cached at construction: this object is short-lived (one heartbeat check) so
+        # listener registration and logging level are stable for its lifetime.
+        self._should_publish = listeners is not None and listeners.enabled_for_server_heartbeat
+        self._should_log = _SDAM_LOGGER.isEnabledFor(logging.DEBUG)
+        self._start: float = 0.0
+
+    def _emit_log(self, message: _SDAMStatusMessage, awaited: bool, **extra: Any) -> None:
+        _debug_log(
+            _SDAM_LOGGER,
+            message=message,
+            topologyId=self._topology_id,
+            serverHost=self._address[0],
+            serverPort=self._address[1],
+            awaited=awaited,
+            **extra,
+        )
+
+    def started(self, awaited: bool) -> None:
+        """Publish the APM heartbeat-started event (before connection checkout)."""
+        if self._should_publish or self._should_log:
+            self._start = time.monotonic()
+        if self._should_publish:
+            assert self._listeners is not None
+            self._listeners.publish_server_heartbeat_started(self._address, awaited)
+
+    def emit_started_log(self, conn_id: int, server_conn_id: Optional[int], awaited: bool) -> None:
+        """Emit the log entry for heartbeat started (after connection checkout)."""
+        if self._should_log:
+            self._emit_log(
+                _SDAMStatusMessage.HEARTBEAT_START,
+                awaited=awaited,
+                driverConnectionId=conn_id,
+                serverConnectionId=server_conn_id,
+            )
+
+    def succeeded(
+        self,
+        round_trip_time: float,
+        response: Hello[Any],
+        conn_id: int,
+        server_conn_id: Optional[int],
+    ) -> None:
+        """Emit the SUCCEEDED log entry and APM event."""
+        if self._should_publish:
+            assert self._listeners is not None
+            self._listeners.publish_server_heartbeat_succeeded(
+                self._address, round_trip_time, response, response.awaitable
+            )
+        if self._should_log:
+            self._emit_log(
+                _SDAMStatusMessage.HEARTBEAT_SUCCESS,
+                awaited=response.awaitable,
+                driverConnectionId=conn_id,
+                serverConnectionId=server_conn_id,
+                durationMS=round_trip_time * 1000,
+                reply=response.document,
+            )
+
+    def failed(self, error: Exception, conn_id: Optional[int], awaited: bool) -> None:
+        """Emit the FAILED log entry and APM event."""
+        should_publish = self._should_publish
+        should_log = self._should_log
+        if not should_publish and not should_log:
+            return
+        duration = _monotonic_duration(self._start)
+        if should_publish:
+            assert self._listeners is not None
+            self._listeners.publish_server_heartbeat_failed(self._address, duration, error, awaited)
+        if should_log:
+            self._emit_log(
+                _SDAMStatusMessage.HEARTBEAT_FAIL,
+                awaited=awaited,
+                durationMS=duration * 1000,
+                failure=error,
+                driverConnectionId=conn_id,
+            )
+
+
+class _SdamTelemetry:
+    """Combines SDAM structured logging and APM event publishing for topology and server events.
+
+    Topology events are queued for asynchronous delivery; log entries are emitted inline.
+    """
+
+    __slots__ = ("_events", "_listeners", "_topology_id")
+
+    def __init__(
+        self,
+        topology_id: ObjectId,
+        listeners: Optional[_EventListeners],
+        events: Optional[queue.Queue[Any]],
+    ) -> None:
+        self._topology_id = topology_id
+        self._listeners = listeners
+        self._events = events
+
+    @property
+    def _publish_server(self) -> bool:
+        """Computed per-call because listener registration can change while the topology is open."""
+        return self._listeners is not None and self._listeners.enabled_for_server
+
+    @property
+    def _publish_tp(self) -> bool:
+        """Computed per-call because listener registration can change while the topology is open."""
+        return self._listeners is not None and self._listeners.enabled_for_topology
+
+    @property
+    def _should_log(self) -> bool:
+        """Computed per-call because logging level can be reconfigured at runtime."""
+        return _SDAM_LOGGER.isEnabledFor(logging.DEBUG)
+
+    def _enqueue(self, fn: Any, args: tuple[Any, ...]) -> None:
+        if self._events is not None:
+            self._events.put((fn, args))
+
+    def _emit_log(self, message: _SDAMStatusMessage, **extra: Any) -> None:
+        _debug_log(
+            _SDAM_LOGGER,
+            message=message,
+            topologyId=self._topology_id,
+            **extra,
+        )
+
+    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: TopologyDescription, new_td: TopologyDescription
+    ) -> None:
+        """Emit the topology description changed APM event and log entry."""
+        if self._publish_tp:
+            assert self._listeners is not None
+            self._enqueue(
+                self._listeners.publish_topology_description_changed,
+                (old_td, new_td, self._topology_id),
+            )
+        if self._should_log:
+            self._emit_log(
+                _SDAMStatusMessage.TOPOLOGY_CHANGE,
+                previousDescription=repr(old_td),
+                newDescription=repr(new_td),
+            )
+
+    def topology_closed(self, old_td: TopologyDescription, new_td: TopologyDescription) -> None:
+        """Emit APM and log events for topology description change + topology closed."""
+        if self._publish_tp:
+            assert self._listeners is not None
+            self._enqueue(
+                self._listeners.publish_topology_description_changed,
+                (old_td, new_td, self._topology_id),
+            )
+            self._enqueue(self._listeners.publish_topology_closed, (self._topology_id,))
+        if self._should_log:
+            self._emit_log(
+                _SDAMStatusMessage.TOPOLOGY_CHANGE,
+                previousDescription=repr(old_td),
+                newDescription=repr(new_td),
+            )
+            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))
+        if self._should_log:
+            self._emit_log(
+                _SDAMStatusMessage.START_SERVER,
+                serverHost=address[0],
+                serverPort=address[1],
+            )
+
+    def server_description_changed(
+        self, sd_old: ServerDescription, sd_new: ServerDescription, address: _Address
+    ) -> None:
+        """Emit the server description changed APM event."""
+        if self._publish_server:
+            assert self._listeners is not None
+            self._enqueue(
+                self._listeners.publish_server_description_changed,
+                (sd_old, sd_new, address, self._topology_id),
+            )
+
+    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))
+        if self._should_log:
+            self._emit_log(
+                _SDAMStatusMessage.STOP_SERVER,
+                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",
+        "_should_log",
+        "_topology_description",
+        "_topology_id",
+    )
+
+    def __init__(
+        self,
+        topology_id: ObjectId,
+        selector: Any,
+        operation: str,
+        operation_id: Optional[int],
+        topology_description: TopologyDescription,
+    ) -> None:
+        self._topology_id = topology_id
+        self._selector = selector
+        self._operation = operation
+        self._operation_id = operation_id
+        self._topology_description = topology_description
+        # Cached at construction: this object is short-lived (one select_server call) so
+        # logging level is stable for its lifetime.
+        self._should_log = _SERVER_SELECTION_LOGGER.isEnabledFor(logging.DEBUG)
+
+    def _emit_log(
+        self,
+        message: _ServerSelectionStatusMessage,
+        topology_description: TopologyDescription,
+        **extra: Any,
+    ) -> None:
+        _debug_log(
+            _SERVER_SELECTION_LOGGER,
+            message=message,
+            clientId=self._topology_id,
+            selector=self._selector,
+            operation=self._operation,
+            operationId=self._operation_id,
+            topologyDescription=topology_description,
+            **extra,
+        )
+
+    def started(self) -> None:
+        """Emit the server selection STARTED log entry."""
+        if self._should_log:
+            self._emit_log(_ServerSelectionStatusMessage.STARTED, self._topology_description)
+
+    def waiting(self, remaining_time_ms: int) -> None:
+        """Emit the server selection WAITING log entry."""
+        if self._should_log:
+            self._emit_log(
+                _ServerSelectionStatusMessage.WAITING,
+                self._topology_description,
+                remainingTimeMS=remaining_time_ms,
+            )
+
+    def failed(self, failure: str, topology_description: TopologyDescription) -> None:
+        """Emit the server selection FAILED log entry with the current topology description."""
+        if self._should_log:
+            self._emit_log(
+                _ServerSelectionStatusMessage.FAILED,
+                topology_description,
+                failure=failure,
+            )
+
+
+def log_server_selection_succeeded(
+    topology_id: ObjectId,
+    selector: Any,
+    operation: str,
+    operation_id: Optional[int],
+    topology_description: TopologyDescription,
+    server_host: str,
+    server_port: Optional[int],
+) -> None:
+    """Emit the server selection SUCCEEDED log entry."""
+    if _SERVER_SELECTION_LOGGER.isEnabledFor(logging.DEBUG):
+        _debug_log(
+            _SERVER_SELECTION_LOGGER,
+            message=_ServerSelectionStatusMessage.SUCCEEDED,
+            clientId=topology_id,
+            selector=selector,
+            operation=operation,
+            operationId=operation_id,
+            topologyDescription=topology_description,
+            serverHost=server_host,
+            serverPort=server_port,
+        )
+
+
+def log_srv_monitor_failure(failure: Exception) -> None:
+    """Emit a log entry when the SRV monitor fails to poll DNS records."""
+    if _SDAM_LOGGER.isEnabledFor(logging.DEBUG):
+        _debug_log(_SDAM_LOGGER, message="SRV monitor check failed", failure=repr(failure))
+
+
+def log_command_retry(
+    topology_id: Optional[ObjectId],
+    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,
+        )
diff --git a/pymongo/asynchronous/mongo_client.py b/pymongo/asynchronous/mongo_client.py
index c745720ed9..2a53462f59 100644
--- a/pymongo/asynchronous/mongo_client.py
+++ b/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,
 )
@@ -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
 
 
diff --git a/pymongo/asynchronous/monitor.py b/pymongo/asynchronous/monitor.py
index 45c12b219f..537d9ea7f5 100644
--- a/pymongo/asynchronous/monitor.py
+++ b/pymongo/asynchronous/monitor.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))
diff --git a/pymongo/asynchronous/pool.py b/pymongo/asynchronous/pool.py
index 4ed3b85dbf..b3cfae8a45 100644
--- a/pymongo/asynchronous/pool.py
+++ b/pymongo/asynchronous/pool.py
@@ -17,7 +17,6 @@
 import asyncio
 import collections
 import contextlib
-import logging
 import os
 import socket
 import ssl
@@ -35,6 +34,7 @@
 
 from bson import DEFAULT_CODEC_OPTIONS
 from pymongo import _csot, helpers_shared
+from pymongo._telemetry import _CmapTelemetry
 from pymongo.asynchronous.client_session import _validate_session_write_concern
 from pymongo.asynchronous.command_runner import run_command
 from pymongo.asynchronous.helpers import _handle_reauth
@@ -65,12 +65,6 @@
     _async_create_condition,
     _async_create_lock,
 )
-from pymongo.logger import (
-    _CONNECTION_LOGGER,
-    _ConnectionStatusMessage,
-    _debug_log,
-    _verbose_connection_error_reason,
-)
 from pymongo.monitoring import (
     ConnectionCheckOutFailedReason,
     ConnectionClosedReason,
@@ -134,6 +128,7 @@ def __init__(
         self.id = id
         self.is_sdam = is_sdam
         self.closed = False
+        self.creation_time = time.monotonic()
         self.last_checkin_time = time.monotonic()
         self.performed_handshake = False
         self.is_writable: bool = False
@@ -145,8 +140,7 @@ def __init__(
         self.hello_ok: bool = False
         self.is_mongos = False
         self.listeners = pool.opts._event_listeners
-        self.enabled_for_cmap = pool.enabled_for_cmap
-        self.enabled_for_logging = pool.enabled_for_logging
+        self._telemetry = pool._telemetry
         self.compression_settings = pool.opts._compression_settings
         self.compression_context: Union[SnappyContext, ZlibContext, ZstdContext, None] = None
         self.socket_checker: SocketChecker = SocketChecker()
@@ -173,8 +167,6 @@ def __init__(
         self.active = False
         self.last_timeout = self.opts.socket_timeout
         self.connect_rtt = 0.0
-        self._client_id = pool._client_id
-        self.creation_time = time.monotonic()
         # For gossiping $clusterTime from the connection handshake to the client.
         self._cluster_time = None
 
@@ -475,21 +467,7 @@ async def authenticate(self, reauthenticate: bool = False) -> None:
 
                 await auth.authenticate(creds, self, reauthenticate=reauthenticate)
             self.ready = True
-            duration = time.monotonic() - self.creation_time
-            # Log before publishing event to prevent potential listener preemption in tests
-            if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-                _debug_log(
-                    _CONNECTION_LOGGER,
-                    message=_ConnectionStatusMessage.CONN_READY,
-                    clientId=self._client_id,
-                    serverHost=self.address[0],
-                    serverPort=self.address[1],
-                    driverConnectionId=self.id,
-                    durationMS=duration,
-                )
-            if self.enabled_for_cmap:
-                assert self.listeners is not None
-                self.listeners.publish_connection_ready(self.address, self.id, duration)
+            self._telemetry.connection_ready(self.id, self.creation_time)
 
     def validate_session(
         self, client: Optional[AsyncMongoClient[Any]], session: Optional[AsyncClientSession]
@@ -510,20 +488,7 @@ async def close_conn(self, reason: Optional[str]) -> None:
             return
         await self._close_conn()
         if reason:
-            if self.enabled_for_cmap:
-                assert self.listeners is not None
-                self.listeners.publish_connection_closed(self.address, self.id, reason)
-            if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-                _debug_log(
-                    _CONNECTION_LOGGER,
-                    message=_ConnectionStatusMessage.CONN_CLOSED,
-                    clientId=self._client_id,
-                    serverHost=self.address[0],
-                    serverPort=self.address[1],
-                    driverConnectionId=self.id,
-                    reason=_verbose_connection_error_reason(reason),
-                    error=reason,
-                )
+            self._telemetry.connection_closed(self.id, reason)
 
     async def _close_conn(self) -> None:
         """Close this connection."""
@@ -699,13 +664,6 @@ def __init__(
         self.address = address
         self.opts = options
         self.is_sdam = is_sdam
-        # Don't publish events or logs in Monitor pools.
-        self.enabled_for_cmap = (
-            not self.is_sdam
-            and self.opts._event_listeners is not None
-            and self.opts._event_listeners.enabled_for_cmap
-        )
-        self.enabled_for_logging = not self.is_sdam
 
         # The first portion of the wait queue.
         # Enforces: maxPoolSize
@@ -721,25 +679,13 @@ def __init__(
         self._max_connecting_cond = _async_create_condition(self.lock)
         self._pending = 0
         self._max_connecting = self.opts.max_connecting
-        self._client_id = client_id
         self._ssl_session_cache: Optional[list[Any]] = (
             [None] if self.opts._ssl_context is not None else None
         )
-        # Log before publishing event to prevent potential listener preemption in tests
-        if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-            _debug_log(
-                _CONNECTION_LOGGER,
-                message=_ConnectionStatusMessage.POOL_CREATED,
-                clientId=self._client_id,
-                serverHost=self.address[0],
-                serverPort=self.address[1],
-                **self.opts.non_default_options,
-            )
-        if self.enabled_for_cmap:
-            assert self.opts._event_listeners is not None
-            self.opts._event_listeners.publish_pool_created(
-                self.address, self.opts.non_default_options
-            )
+        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
         # Retain references to pinned connections to prevent the CPython GC
@@ -754,17 +700,7 @@ async def ready(self) -> None:
         async with self.lock:
             if self.state != PoolState.READY:
                 self.state = PoolState.READY
-                if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-                    _debug_log(
-                        _CONNECTION_LOGGER,
-                        message=_ConnectionStatusMessage.POOL_READY,
-                        clientId=self._client_id,
-                        serverHost=self.address[0],
-                        serverPort=self.address[1],
-                    )
-                if self.enabled_for_cmap:
-                    assert self.opts._event_listeners is not None
-                    self.opts._event_listeners.publish_pool_ready(self.address)
+                self._telemetry.pool_ready()
 
     @property
     def closed(self) -> bool:
@@ -812,23 +748,7 @@ async def _reset(
             # and free-threaded Python causes ConnectionCheckOutFailedEvent to
             # arrive before PoolClearedEvent (PYTHON-3519).
             if not close and old_state != PoolState.PAUSED:
-                _listeners = self.opts._event_listeners
-                if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-                    _debug_log(
-                        _CONNECTION_LOGGER,
-                        message=_ConnectionStatusMessage.POOL_CLEARED,
-                        clientId=self._client_id,
-                        serverHost=self.address[0],
-                        serverPort=self.address[1],
-                        serviceId=service_id,
-                    )
-                if self.enabled_for_cmap:
-                    assert _listeners is not None
-                    _listeners.publish_pool_cleared(
-                        self.address,
-                        service_id=service_id,
-                        interrupt_connections=interrupt_connections,
-                    )
+                self._telemetry.pool_cleared(service_id, interrupt_connections)
 
             # Clear the wait queue
             self._max_connecting_cond.notify_all()
@@ -838,7 +758,6 @@ async def _reset(
                 for context in self.active_contexts:
                     context.cancel()
 
-        listeners = self.opts._event_listeners
         # CMAP spec says that close() MUST close sockets before publishing the
         # PoolClosedEvent but that reset() SHOULD close sockets *after*
         # publishing the PoolClearedEvent.
@@ -851,17 +770,7 @@ async def _reset(
             else:
                 for conn in sockets:
                     await conn.close_conn(ConnectionClosedReason.POOL_CLOSED)
-            if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-                _debug_log(
-                    _CONNECTION_LOGGER,
-                    message=_ConnectionStatusMessage.POOL_CLOSED,
-                    clientId=self._client_id,
-                    serverHost=self.address[0],
-                    serverPort=self.address[1],
-                )
-            if self.enabled_for_cmap:
-                assert listeners is not None
-                listeners.publish_pool_closed(self.address)
+            self._telemetry.pool_closed()
         else:
             if not _IS_SYNC:
                 await asyncio.gather(
@@ -996,20 +905,7 @@ async def connect(self, handler: Optional[_MongoClientErrorHandler] = None) -> A
             tmp_context = _CancellationContext()
             self.active_contexts.add(tmp_context)
 
-        listeners = self.opts._event_listeners
-        # Log before publishing event to prevent potential listener preemption in tests
-        if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-            _debug_log(
-                _CONNECTION_LOGGER,
-                message=_ConnectionStatusMessage.CONN_CREATED,
-                clientId=self._client_id,
-                serverHost=self.address[0],
-                serverPort=self.address[1],
-                driverConnectionId=conn_id,
-            )
-        if self.enabled_for_cmap:
-            assert listeners is not None
-            listeners.publish_connection_created(self.address, conn_id)
+        self._telemetry.connection_created(conn_id)
 
         try:
             networking_interface = await _configured_protocol_interface(
@@ -1019,22 +915,7 @@ async def connect(self, handler: Optional[_MongoClientErrorHandler] = None) -> A
         except BaseException as error:
             async with self.lock:
                 self.active_contexts.discard(tmp_context)
-            if self.enabled_for_cmap:
-                assert listeners is not None
-                listeners.publish_connection_closed(
-                    self.address, conn_id, ConnectionClosedReason.ERROR
-                )
-            if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-                _debug_log(
-                    _CONNECTION_LOGGER,
-                    message=_ConnectionStatusMessage.CONN_CLOSED,
-                    clientId=self._client_id,
-                    serverHost=self.address[0],
-                    serverPort=self.address[1],
-                    driverConnectionId=conn_id,
-                    reason=_verbose_connection_error_reason(ConnectionClosedReason.ERROR),
-                    error=ConnectionClosedReason.ERROR,
-                )
+            self._telemetry.connection_closed(conn_id, ConnectionClosedReason.ERROR)
             if isinstance(error, (IOError, OSError, *SSLErrors)):
                 details = _get_timeout_details(self.opts)
                 # Wrap to AutoReconnect/NetworkTimeout BEFORE labeling so the
@@ -1096,36 +977,11 @@ async def checkout(
 
         :param handler: A _MongoClientErrorHandler.
         """
-        listeners = self.opts._event_listeners
-        checkout_started_time = time.monotonic()
-        if self.enabled_for_cmap:
-            assert listeners is not None
-            listeners.publish_connection_check_out_started(self.address)
-        if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-            _debug_log(
-                _CONNECTION_LOGGER,
-                message=_ConnectionStatusMessage.CHECKOUT_STARTED,
-                clientId=self._client_id,
-                serverHost=self.address[0],
-                serverPort=self.address[1],
-            )
+        checkout_started_time = self._telemetry.checkout_started()
 
         conn = await self._get_conn(checkout_started_time, handler=handler)
 
-        duration = time.monotonic() - checkout_started_time
-        if self.enabled_for_cmap:
-            assert listeners is not None
-            listeners.publish_connection_checked_out(self.address, conn.id, duration)
-        if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-            _debug_log(
-                _CONNECTION_LOGGER,
-                message=_ConnectionStatusMessage.CHECKOUT_SUCCEEDED,
-                clientId=self._client_id,
-                serverHost=self.address[0],
-                serverPort=self.address[1],
-                driverConnectionId=conn.id,
-                durationMS=duration,
-            )
+        self._telemetry.checkout_succeeded(conn.id, checkout_started_time)
         try:
             async with self.lock:
                 self.active_contexts.add(conn.cancel_context)
@@ -1159,23 +1015,11 @@ async def checkout(
     def _raise_if_not_ready(self, checkout_started_time: float, emit_event: bool) -> None:
         if self.state != PoolState.READY:
             if emit_event:
-                duration = time.monotonic() - checkout_started_time
-                if self.enabled_for_cmap:
-                    assert self.opts._event_listeners is not None
-                    self.opts._event_listeners.publish_connection_check_out_failed(
-                        self.address, ConnectionCheckOutFailedReason.CONN_ERROR, duration
-                    )
-                if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-                    _debug_log(
-                        _CONNECTION_LOGGER,
-                        message=_ConnectionStatusMessage.CHECKOUT_FAILED,
-                        clientId=self._client_id,
-                        serverHost=self.address[0],
-                        serverPort=self.address[1],
-                        reason="An error occurred while trying to establish a new connection",
-                        error=ConnectionCheckOutFailedReason.CONN_ERROR,
-                        durationMS=duration,
-                    )
+                self._telemetry.checkout_failed(
+                    "An error occurred while trying to establish a new connection",
+                    ConnectionCheckOutFailedReason.CONN_ERROR,
+                    checkout_started_time,
+                )
 
             details = _get_timeout_details(self.opts)
             _raise_connection_failure(
@@ -1193,23 +1037,11 @@ async def _get_conn(
             await self.reset_without_pause()
 
         if self.closed:
-            duration = time.monotonic() - checkout_started_time
-            if self.enabled_for_cmap:
-                assert self.opts._event_listeners is not None
-                self.opts._event_listeners.publish_connection_check_out_failed(
-                    self.address, ConnectionCheckOutFailedReason.POOL_CLOSED, duration
-                )
-            if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-                _debug_log(
-                    _CONNECTION_LOGGER,
-                    message=_ConnectionStatusMessage.CHECKOUT_FAILED,
-                    clientId=self._client_id,
-                    serverHost=self.address[0],
-                    serverPort=self.address[1],
-                    reason="Connection pool was closed",
-                    error=ConnectionCheckOutFailedReason.POOL_CLOSED,
-                    durationMS=duration,
-                )
+            self._telemetry.checkout_failed(
+                "Connection pool was closed",
+                ConnectionCheckOutFailedReason.POOL_CLOSED,
+                checkout_started_time,
+            )
             raise _PoolClosedError(
                 "Attempted to check out a connection from closed connection pool"
             )
@@ -1289,23 +1121,11 @@ async def _get_conn(
                 self.size_cond.notify()
 
             if not emitted_event:
-                duration = time.monotonic() - checkout_started_time
-                if self.enabled_for_cmap:
-                    assert self.opts._event_listeners is not None
-                    self.opts._event_listeners.publish_connection_check_out_failed(
-                        self.address, ConnectionCheckOutFailedReason.CONN_ERROR, duration
-                    )
-                if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-                    _debug_log(
-                        _CONNECTION_LOGGER,
-                        message=_ConnectionStatusMessage.CHECKOUT_FAILED,
-                        clientId=self._client_id,
-                        serverHost=self.address[0],
-                        serverPort=self.address[1],
-                        reason="An error occurred while trying to establish a new connection",
-                        error=ConnectionCheckOutFailedReason.CONN_ERROR,
-                        durationMS=duration,
-                    )
+                self._telemetry.checkout_failed(
+                    "An error occurred while trying to establish a new connection",
+                    ConnectionCheckOutFailedReason.CONN_ERROR,
+                    checkout_started_time,
+                )
             raise
 
         conn.active = True
@@ -1322,21 +1142,9 @@ async def checkin(self, conn: AsyncConnection) -> None:
         conn.pinned_txn = False
         conn.pinned_cursor = False
         self.__pinned_sockets.discard(conn)
-        listeners = self.opts._event_listeners
         async with self.lock:
             self.active_contexts.discard(conn.cancel_context)
-        if self.enabled_for_cmap:
-            assert listeners is not None
-            listeners.publish_connection_checked_in(self.address, conn.id)
-        if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-            _debug_log(
-                _CONNECTION_LOGGER,
-                message=_ConnectionStatusMessage.CHECKEDIN,
-                clientId=self._client_id,
-                serverHost=self.address[0],
-                serverPort=self.address[1],
-                driverConnectionId=conn.id,
-            )
+        self._telemetry.checked_in(conn.id)
         if self.pid != os.getpid():
             await self.reset_without_pause()
         else:
@@ -1344,22 +1152,7 @@ async def checkin(self, conn: AsyncConnection) -> None:
                 await conn.close_conn(ConnectionClosedReason.POOL_CLOSED)
             elif conn.closed:
                 # CMAP requires the closed event be emitted after the check in.
-                if self.enabled_for_cmap:
-                    assert listeners is not None
-                    listeners.publish_connection_closed(
-                        self.address, conn.id, ConnectionClosedReason.ERROR
-                    )
-                if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-                    _debug_log(
-                        _CONNECTION_LOGGER,
-                        message=_ConnectionStatusMessage.CONN_CLOSED,
-                        clientId=self._client_id,
-                        serverHost=self.address[0],
-                        serverPort=self.address[1],
-                        driverConnectionId=conn.id,
-                        reason=_verbose_connection_error_reason(ConnectionClosedReason.ERROR),
-                        error=ConnectionClosedReason.ERROR,
-                    )
+                self._telemetry.connection_closed(conn.id, ConnectionClosedReason.ERROR)
             else:
                 close_conn = False
                 async with self.lock:
@@ -1424,24 +1217,11 @@ async def _perished(self, conn: AsyncConnection) -> bool:
         return False
 
     def _raise_wait_queue_timeout(self, checkout_started_time: float) -> NoReturn:
-        listeners = self.opts._event_listeners
-        duration = time.monotonic() - checkout_started_time
-        if self.enabled_for_cmap:
-            assert listeners is not None
-            listeners.publish_connection_check_out_failed(
-                self.address, ConnectionCheckOutFailedReason.TIMEOUT, duration
-            )
-        if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-            _debug_log(
-                _CONNECTION_LOGGER,
-                message=_ConnectionStatusMessage.CHECKOUT_FAILED,
-                clientId=self._client_id,
-                serverHost=self.address[0],
-                serverPort=self.address[1],
-                reason="Wait queue timeout elapsed without a connection becoming available",
-                error=ConnectionCheckOutFailedReason.TIMEOUT,
-                durationMS=duration,
-            )
+        self._telemetry.checkout_failed(
+            "Wait queue timeout elapsed without a connection becoming available",
+            ConnectionCheckOutFailedReason.TIMEOUT,
+            checkout_started_time,
+        )
         timeout = _csot.get_timeout() or self.opts.wait_queue_timeout
         if self.opts.load_balanced:
             other_ops = self.active_sockets - self.ncursors - self.ntxns
diff --git a/pymongo/asynchronous/server.py b/pymongo/asynchronous/server.py
index 0b19fd16a3..607cb3c492 100644
--- a/pymongo/asynchronous/server.py
+++ b/pymongo/asynchronous/server.py
@@ -16,7 +16,6 @@
 
 from __future__ import annotations
 
-import logging
 from contextlib import AbstractAsyncContextManager
 from typing import (
     TYPE_CHECKING,
@@ -24,11 +23,7 @@
     Optional,
 )
 
-from pymongo.logger import (
-    _SDAM_LOGGER,
-    _debug_log,
-    _SDAMStatusMessage,
-)
+from pymongo._telemetry import _SdamTelemetry
 
 if TYPE_CHECKING:
     from queue import Queue
@@ -58,12 +53,8 @@ def __init__(
         self._description = server_description
         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]
+        _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.
@@ -82,23 +73,7 @@ async def close(self) -> None:
 
         Reconnect with open().
         """
-        if self._publish:
-            assert self._listener is not None
-            assert self._events is not None
-            self._events.put(
-                (
-                    self._listener.publish_server_closed,
-                    (self._description.address, self._topology_id),
-                )
-            )
-        if _SDAM_LOGGER.isEnabledFor(logging.DEBUG):
-            _debug_log(
-                _SDAM_LOGGER,
-                message=_SDAMStatusMessage.STOP_SERVER,
-                topologyId=self._topology_id,
-                serverHost=self._description.address[0],
-                serverPort=self._description.address[1],
-            )
+        self._sdam.server_closed(self._description.address)
 
         await self._monitor.close()
         await self._pool.close()
diff --git a/pymongo/asynchronous/topology.py b/pymongo/asynchronous/topology.py
index 6af075cdc2..ca8244c02a 100644
--- a/pymongo/asynchronous/topology.py
+++ b/pymongo/asynchronous/topology.py
@@ -17,7 +17,6 @@
 from __future__ import annotations
 
 import asyncio
-import logging
 import os
 import queue
 import random
@@ -30,6 +29,11 @@
 from typing import TYPE_CHECKING, Any, Callable, Optional, cast
 
 from pymongo import _csot, common, helpers_shared, periodic_executor
+from pymongo._telemetry import (
+    _SdamTelemetry,
+    _ServerSelectionTelemetry,
+    log_server_selection_succeeded,
+)
 from pymongo.asynchronous.client_session import _ServerSession, _ServerSessionPool
 from pymongo.asynchronous.monitor import MonitorBase, SrvMonitor
 from pymongo.asynchronous.pool import Pool
@@ -51,13 +55,6 @@
     _async_create_condition,
     _async_create_lock,
 )
-from pymongo.logger import (
-    _SDAM_LOGGER,
-    _SERVER_SELECTION_LOGGER,
-    _debug_log,
-    _SDAMStatusMessage,
-    _ServerSelectionStatusMessage,
-)
 from pymongo.pool_options import PoolOptions
 from pymongo.server_description import ServerDescription
 from pymongo.server_selectors import (
@@ -108,27 +105,19 @@ 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)
 
-        if _SDAM_LOGGER.isEnabledFor(logging.DEBUG):
-            _debug_log(
-                _SDAM_LOGGER,
-                message=_SDAMStatusMessage.START_TOPOLOGY,
-                topologyId=self._topology_id,
-            )
+        self._sdam = _SdamTelemetry(self._topology_id, self._listeners, self._events)
+        self._sdam.topology_opened()
 
-        if self._publish_tp:
-            assert self._events is not None
-            assert self._listeners is not None
-            self._events.put((self._listeners.publish_topology_opened, (self._topology_id,)))
         self._settings = topology_settings
         topology_description = TopologyDescription(
             topology_settings.get_topology_type(),
@@ -143,37 +132,10 @@ def __init__(self, topology_settings: TopologySettings):
         initial_td = TopologyDescription(
             TOPOLOGY_TYPE.Unknown, {}, None, None, None, self._settings
         )
-        if self._publish_tp:
-            assert self._events is not None
-            assert self._listeners is not None
-            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,
-                message=_SDAMStatusMessage.TOPOLOGY_CHANGE,
-                topologyId=self._topology_id,
-                previousDescription=repr(initial_td),
-                newDescription=repr(self._description),
-            )
+        self._sdam.topology_description_changed(initial_td, self._description)
 
         for seed in topology_settings.seeds:
-            if self._publish_server:
-                assert self._events is not None
-                assert self._listeners 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,
-                    message=_SDAMStatusMessage.START_SERVER,
-                    topologyId=self._topology_id,
-                    serverHost=seed[0],
-                    serverPort=seed[1],
-                )
+            self._sdam.server_opened(seed)
 
         # Store the seed list to help diagnose errors in _error_message().
         self._seed_addresses = list(topology_description.server_descriptions())
@@ -188,7 +150,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]]
 
@@ -321,17 +283,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,
@@ -345,32 +300,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), self.description)
                 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()
@@ -435,18 +371,15 @@ 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],
-            )
+        log_server_selection_succeeded(
+            self._topology_id,
+            selector,
+            operation,
+            operation_id,
+            self.description,
+            server.description.address[0],
+            server.description.address[1],
+        )
         return server
 
     async def select_server_by_address(
@@ -508,36 +441,16 @@ async def _process_change(
                 await server.pool.ready()
 
         suppress_event = sd_old == server_description
-        if self._publish_server and not suppress_event:
-            assert self._events is not None
-            assert self._listeners is not None
-            self._events.put(
-                (
-                    self._listeners.publish_server_description_changed,
-                    (sd_old, server_description, server_description.address, self._topology_id),
-                )
+        if not suppress_event:
+            self._sdam.server_description_changed(
+                sd_old, server_description, server_description.address
             )
 
         self._description = new_td
         await self._update_servers()
 
-        if self._publish_tp and not suppress_event:
-            assert self._events is not None
-            assert self._listeners is not None
-            self._events.put(
-                (
-                    self._listeners.publish_topology_description_changed,
-                    (td_old, self._description, self._topology_id),
-                )
-            )
-        if _SDAM_LOGGER.isEnabledFor(logging.DEBUG) and not suppress_event:
-            _debug_log(
-                _SDAM_LOGGER,
-                message=_SDAMStatusMessage.TOPOLOGY_CHANGE,
-                topologyId=self._topology_id,
-                previousDescription=repr(td_old),
-                newDescription=repr(self._description),
-            )
+        if not suppress_event:
+            self._sdam.topology_description_changed(td_old, self._description)
 
         # Shutdown SRV polling for unsupported cluster types.
         # This is only applicable if the old topology was Unknown, and the
@@ -588,24 +501,7 @@ async def _process_srv_update(self, seedlist: list[tuple[str, Any]]) -> None:
         self._description = _updated_topology_description_srv_polling(self._description, seedlist)
 
         await self._update_servers()
-
-        if self._publish_tp:
-            assert self._events is not None
-            assert self._listeners is not None
-            self._events.put(
-                (
-                    self._listeners.publish_topology_description_changed,
-                    (td_old, self._description, self._topology_id),
-                )
-            )
-        if _SDAM_LOGGER.isEnabledFor(logging.DEBUG):
-            _debug_log(
-                _SDAM_LOGGER,
-                message=_SDAMStatusMessage.TOPOLOGY_CHANGE,
-                topologyId=self._topology_id,
-                previousDescription=repr(td_old),
-                newDescription=repr(self._description),
-            )
+        self._sdam.topology_description_changed(td_old, self._description)
 
     async def on_srv_update(self, seedlist: list[tuple[str, Any]]) -> None:
         """Process a new list of nodes obtained from scanning SRV records."""
@@ -743,9 +639,7 @@ async def close(self) -> None:
             self._closed = True
 
         # Publish only after releasing the lock.
-        if self._publish_tp:
-            assert self._events is not None
-            assert self._listeners is not None
+        if self._sdam._publish_tp:
             self._description = TopologyDescription(
                 TOPOLOGY_TYPE.Unknown,
                 {},
@@ -754,30 +648,9 @@ async def close(self) -> None:
                 self._description.max_election_id,
                 self._description._topology_settings,
             )
-            self._events.put(
-                (
-                    self._listeners.publish_topology_description_changed,
-                    (
-                        old_td,
-                        self._description,
-                        self._topology_id,
-                    ),
-                )
-            )
-            self._events.put((self._listeners.publish_topology_closed, (self._topology_id,)))
-        if _SDAM_LOGGER.isEnabledFor(logging.DEBUG):
-            _debug_log(
-                _SDAM_LOGGER,
-                message=_SDAMStatusMessage.TOPOLOGY_CHANGE,
-                topologyId=self._topology_id,
-                previousDescription=repr(old_td),
-                newDescription=repr(self._description),
-            )
-            _debug_log(
-                _SDAM_LOGGER, message=_SDAMStatusMessage.STOP_TOPOLOGY, topologyId=self._topology_id
-            )
+        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)
@@ -818,7 +691,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.
@@ -957,7 +830,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,
diff --git a/pymongo/synchronous/mongo_client.py b/pymongo/synchronous/mongo_client.py
index 90912dd6f5..24a0401410 100644
--- a/pymongo/synchronous/mongo_client.py
+++ b/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,
 )
@@ -2971,6 +2970,15 @@ 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,
+        )
+
     def _write(self) -> T:
         """Wrapper method for write-type retryable client executions
 
@@ -2994,13 +3002,7 @@ 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 self._func(self._session, conn, self._retryable)  # type: ignore
         except PyMongoError as exc:
             if not self._retryable:
@@ -3023,13 +3025,7 @@ 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 self._func(self._session, self._server, conn, read_pref)  # type: ignore
 
 
diff --git a/pymongo/synchronous/monitor.py b/pymongo/synchronous/monitor.py
index f395588814..d56206b127 100644
--- a/pymongo/synchronous/monitor.py
+++ b/pymongo/synchronous/monitor.py
@@ -18,17 +18,16 @@
 
 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.errors import NetworkTimeout, _OperationCancelled
 from pymongo.hello import Hello
 from pymongo.lock import _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,
@@ -255,32 +246,17 @@ def _check_server(self) -> ServerDescription:
         Returns a ServerDescription.
         """
         self._conn_id = None
-        start = time.monotonic()
+        self._current_hb = None
+        self._awaited = False
         try:
             return 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)
             self._reset_connection()
             if isinstance(error, _OperationCancelled):
                 raise
@@ -298,28 +274,17 @@ 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:
             self._reset_connection()
         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
@@ -329,24 +294,7 @@ def _check_once(self) -> ServerDescription:
 
             avg_rtt, min_rtt = 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
 
     def _check_with_socket(self, conn: Connection) -> tuple[Hello, float]:  # type: ignore[type-arg]
@@ -427,7 +375,7 @@ 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))
diff --git a/pymongo/synchronous/pool.py b/pymongo/synchronous/pool.py
index 1006735444..9159ff704a 100644
--- a/pymongo/synchronous/pool.py
+++ b/pymongo/synchronous/pool.py
@@ -17,7 +17,6 @@
 import asyncio
 import collections
 import contextlib
-import logging
 import os
 import socket
 import ssl
@@ -35,6 +34,7 @@
 
 from bson import DEFAULT_CODEC_OPTIONS
 from pymongo import _csot, helpers_shared
+from pymongo._telemetry import _CmapTelemetry
 from pymongo.common import (
     MAX_BSON_SIZE,
     MAX_MESSAGE_SIZE,
@@ -62,12 +62,6 @@
     _create_condition,
     _create_lock,
 )
-from pymongo.logger import (
-    _CONNECTION_LOGGER,
-    _ConnectionStatusMessage,
-    _debug_log,
-    _verbose_connection_error_reason,
-)
 from pymongo.monitoring import (
     ConnectionCheckOutFailedReason,
     ConnectionClosedReason,
@@ -134,6 +128,7 @@ def __init__(
         self.id = id
         self.is_sdam = is_sdam
         self.closed = False
+        self.creation_time = time.monotonic()
         self.last_checkin_time = time.monotonic()
         self.performed_handshake = False
         self.is_writable: bool = False
@@ -145,8 +140,7 @@ def __init__(
         self.hello_ok: bool = False
         self.is_mongos = False
         self.listeners = pool.opts._event_listeners
-        self.enabled_for_cmap = pool.enabled_for_cmap
-        self.enabled_for_logging = pool.enabled_for_logging
+        self._telemetry = pool._telemetry
         self.compression_settings = pool.opts._compression_settings
         self.compression_context: Union[SnappyContext, ZlibContext, ZstdContext, None] = None
         self.socket_checker: SocketChecker = SocketChecker()
@@ -173,8 +167,6 @@ def __init__(
         self.active = False
         self.last_timeout = self.opts.socket_timeout
         self.connect_rtt = 0.0
-        self._client_id = pool._client_id
-        self.creation_time = time.monotonic()
         # For gossiping $clusterTime from the connection handshake to the client.
         self._cluster_time = None
 
@@ -475,21 +467,7 @@ def authenticate(self, reauthenticate: bool = False) -> None:
 
                 auth.authenticate(creds, self, reauthenticate=reauthenticate)
             self.ready = True
-            duration = time.monotonic() - self.creation_time
-            # Log before publishing event to prevent potential listener preemption in tests
-            if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-                _debug_log(
-                    _CONNECTION_LOGGER,
-                    message=_ConnectionStatusMessage.CONN_READY,
-                    clientId=self._client_id,
-                    serverHost=self.address[0],
-                    serverPort=self.address[1],
-                    driverConnectionId=self.id,
-                    durationMS=duration,
-                )
-            if self.enabled_for_cmap:
-                assert self.listeners is not None
-                self.listeners.publish_connection_ready(self.address, self.id, duration)
+            self._telemetry.connection_ready(self.id, self.creation_time)
 
     def validate_session(
         self, client: Optional[MongoClient[Any]], session: Optional[ClientSession]
@@ -508,20 +486,7 @@ def close_conn(self, reason: Optional[str]) -> None:
             return
         self._close_conn()
         if reason:
-            if self.enabled_for_cmap:
-                assert self.listeners is not None
-                self.listeners.publish_connection_closed(self.address, self.id, reason)
-            if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-                _debug_log(
-                    _CONNECTION_LOGGER,
-                    message=_ConnectionStatusMessage.CONN_CLOSED,
-                    clientId=self._client_id,
-                    serverHost=self.address[0],
-                    serverPort=self.address[1],
-                    driverConnectionId=self.id,
-                    reason=_verbose_connection_error_reason(reason),
-                    error=reason,
-                )
+            self._telemetry.connection_closed(self.id, reason)
 
     def _close_conn(self) -> None:
         """Close this connection."""
@@ -697,13 +662,6 @@ def __init__(
         self.address = address
         self.opts = options
         self.is_sdam = is_sdam
-        # Don't publish events or logs in Monitor pools.
-        self.enabled_for_cmap = (
-            not self.is_sdam
-            and self.opts._event_listeners is not None
-            and self.opts._event_listeners.enabled_for_cmap
-        )
-        self.enabled_for_logging = not self.is_sdam
 
         # The first portion of the wait queue.
         # Enforces: maxPoolSize
@@ -719,25 +677,13 @@ def __init__(
         self._max_connecting_cond = _create_condition(self.lock)
         self._pending = 0
         self._max_connecting = self.opts.max_connecting
-        self._client_id = client_id
         self._ssl_session_cache: Optional[list[Any]] = (
             [None] if self.opts._ssl_context is not None else None
         )
-        # Log before publishing event to prevent potential listener preemption in tests
-        if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-            _debug_log(
-                _CONNECTION_LOGGER,
-                message=_ConnectionStatusMessage.POOL_CREATED,
-                clientId=self._client_id,
-                serverHost=self.address[0],
-                serverPort=self.address[1],
-                **self.opts.non_default_options,
-            )
-        if self.enabled_for_cmap:
-            assert self.opts._event_listeners is not None
-            self.opts._event_listeners.publish_pool_created(
-                self.address, self.opts.non_default_options
-            )
+        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
         # Retain references to pinned connections to prevent the CPython GC
@@ -752,17 +698,7 @@ def ready(self) -> None:
         with self.lock:
             if self.state != PoolState.READY:
                 self.state = PoolState.READY
-                if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-                    _debug_log(
-                        _CONNECTION_LOGGER,
-                        message=_ConnectionStatusMessage.POOL_READY,
-                        clientId=self._client_id,
-                        serverHost=self.address[0],
-                        serverPort=self.address[1],
-                    )
-                if self.enabled_for_cmap:
-                    assert self.opts._event_listeners is not None
-                    self.opts._event_listeners.publish_pool_ready(self.address)
+                self._telemetry.pool_ready()
 
     @property
     def closed(self) -> bool:
@@ -810,23 +746,7 @@ def _reset(
             # and free-threaded Python causes ConnectionCheckOutFailedEvent to
             # arrive before PoolClearedEvent (PYTHON-3519).
             if not close and old_state != PoolState.PAUSED:
-                _listeners = self.opts._event_listeners
-                if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-                    _debug_log(
-                        _CONNECTION_LOGGER,
-                        message=_ConnectionStatusMessage.POOL_CLEARED,
-                        clientId=self._client_id,
-                        serverHost=self.address[0],
-                        serverPort=self.address[1],
-                        serviceId=service_id,
-                    )
-                if self.enabled_for_cmap:
-                    assert _listeners is not None
-                    _listeners.publish_pool_cleared(
-                        self.address,
-                        service_id=service_id,
-                        interrupt_connections=interrupt_connections,
-                    )
+                self._telemetry.pool_cleared(service_id, interrupt_connections)
 
             # Clear the wait queue
             self._max_connecting_cond.notify_all()
@@ -836,7 +756,6 @@ def _reset(
                 for context in self.active_contexts:
                     context.cancel()
 
-        listeners = self.opts._event_listeners
         # CMAP spec says that close() MUST close sockets before publishing the
         # PoolClosedEvent but that reset() SHOULD close sockets *after*
         # publishing the PoolClearedEvent.
@@ -849,17 +768,7 @@ def _reset(
             else:
                 for conn in sockets:
                     conn.close_conn(ConnectionClosedReason.POOL_CLOSED)
-            if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-                _debug_log(
-                    _CONNECTION_LOGGER,
-                    message=_ConnectionStatusMessage.POOL_CLOSED,
-                    clientId=self._client_id,
-                    serverHost=self.address[0],
-                    serverPort=self.address[1],
-                )
-            if self.enabled_for_cmap:
-                assert listeners is not None
-                listeners.publish_pool_closed(self.address)
+            self._telemetry.pool_closed()
         else:
             if not _IS_SYNC:
                 asyncio.gather(
@@ -992,20 +901,7 @@ def connect(self, handler: Optional[_MongoClientErrorHandler] = None) -> Connect
             tmp_context = _CancellationContext()
             self.active_contexts.add(tmp_context)
 
-        listeners = self.opts._event_listeners
-        # Log before publishing event to prevent potential listener preemption in tests
-        if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-            _debug_log(
-                _CONNECTION_LOGGER,
-                message=_ConnectionStatusMessage.CONN_CREATED,
-                clientId=self._client_id,
-                serverHost=self.address[0],
-                serverPort=self.address[1],
-                driverConnectionId=conn_id,
-            )
-        if self.enabled_for_cmap:
-            assert listeners is not None
-            listeners.publish_connection_created(self.address, conn_id)
+        self._telemetry.connection_created(conn_id)
 
         try:
             networking_interface = _configured_socket_interface(
@@ -1015,22 +911,7 @@ def connect(self, handler: Optional[_MongoClientErrorHandler] = None) -> Connect
         except BaseException as error:
             with self.lock:
                 self.active_contexts.discard(tmp_context)
-            if self.enabled_for_cmap:
-                assert listeners is not None
-                listeners.publish_connection_closed(
-                    self.address, conn_id, ConnectionClosedReason.ERROR
-                )
-            if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-                _debug_log(
-                    _CONNECTION_LOGGER,
-                    message=_ConnectionStatusMessage.CONN_CLOSED,
-                    clientId=self._client_id,
-                    serverHost=self.address[0],
-                    serverPort=self.address[1],
-                    driverConnectionId=conn_id,
-                    reason=_verbose_connection_error_reason(ConnectionClosedReason.ERROR),
-                    error=ConnectionClosedReason.ERROR,
-                )
+            self._telemetry.connection_closed(conn_id, ConnectionClosedReason.ERROR)
             if isinstance(error, (IOError, OSError, *SSLErrors)):
                 details = _get_timeout_details(self.opts)
                 # Wrap to AutoReconnect/NetworkTimeout BEFORE labeling so the
@@ -1092,36 +973,11 @@ def checkout(
 
         :param handler: A _MongoClientErrorHandler.
         """
-        listeners = self.opts._event_listeners
-        checkout_started_time = time.monotonic()
-        if self.enabled_for_cmap:
-            assert listeners is not None
-            listeners.publish_connection_check_out_started(self.address)
-        if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-            _debug_log(
-                _CONNECTION_LOGGER,
-                message=_ConnectionStatusMessage.CHECKOUT_STARTED,
-                clientId=self._client_id,
-                serverHost=self.address[0],
-                serverPort=self.address[1],
-            )
+        checkout_started_time = self._telemetry.checkout_started()
 
         conn = self._get_conn(checkout_started_time, handler=handler)
 
-        duration = time.monotonic() - checkout_started_time
-        if self.enabled_for_cmap:
-            assert listeners is not None
-            listeners.publish_connection_checked_out(self.address, conn.id, duration)
-        if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-            _debug_log(
-                _CONNECTION_LOGGER,
-                message=_ConnectionStatusMessage.CHECKOUT_SUCCEEDED,
-                clientId=self._client_id,
-                serverHost=self.address[0],
-                serverPort=self.address[1],
-                driverConnectionId=conn.id,
-                durationMS=duration,
-            )
+        self._telemetry.checkout_succeeded(conn.id, checkout_started_time)
         try:
             with self.lock:
                 self.active_contexts.add(conn.cancel_context)
@@ -1155,23 +1011,11 @@ def checkout(
     def _raise_if_not_ready(self, checkout_started_time: float, emit_event: bool) -> None:
         if self.state != PoolState.READY:
             if emit_event:
-                duration = time.monotonic() - checkout_started_time
-                if self.enabled_for_cmap:
-                    assert self.opts._event_listeners is not None
-                    self.opts._event_listeners.publish_connection_check_out_failed(
-                        self.address, ConnectionCheckOutFailedReason.CONN_ERROR, duration
-                    )
-                if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-                    _debug_log(
-                        _CONNECTION_LOGGER,
-                        message=_ConnectionStatusMessage.CHECKOUT_FAILED,
-                        clientId=self._client_id,
-                        serverHost=self.address[0],
-                        serverPort=self.address[1],
-                        reason="An error occurred while trying to establish a new connection",
-                        error=ConnectionCheckOutFailedReason.CONN_ERROR,
-                        durationMS=duration,
-                    )
+                self._telemetry.checkout_failed(
+                    "An error occurred while trying to establish a new connection",
+                    ConnectionCheckOutFailedReason.CONN_ERROR,
+                    checkout_started_time,
+                )
 
             details = _get_timeout_details(self.opts)
             _raise_connection_failure(
@@ -1189,23 +1033,11 @@ def _get_conn(
             self.reset_without_pause()
 
         if self.closed:
-            duration = time.monotonic() - checkout_started_time
-            if self.enabled_for_cmap:
-                assert self.opts._event_listeners is not None
-                self.opts._event_listeners.publish_connection_check_out_failed(
-                    self.address, ConnectionCheckOutFailedReason.POOL_CLOSED, duration
-                )
-            if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-                _debug_log(
-                    _CONNECTION_LOGGER,
-                    message=_ConnectionStatusMessage.CHECKOUT_FAILED,
-                    clientId=self._client_id,
-                    serverHost=self.address[0],
-                    serverPort=self.address[1],
-                    reason="Connection pool was closed",
-                    error=ConnectionCheckOutFailedReason.POOL_CLOSED,
-                    durationMS=duration,
-                )
+            self._telemetry.checkout_failed(
+                "Connection pool was closed",
+                ConnectionCheckOutFailedReason.POOL_CLOSED,
+                checkout_started_time,
+            )
             raise _PoolClosedError(
                 "Attempted to check out a connection from closed connection pool"
             )
@@ -1285,23 +1117,11 @@ def _get_conn(
                 self.size_cond.notify()
 
             if not emitted_event:
-                duration = time.monotonic() - checkout_started_time
-                if self.enabled_for_cmap:
-                    assert self.opts._event_listeners is not None
-                    self.opts._event_listeners.publish_connection_check_out_failed(
-                        self.address, ConnectionCheckOutFailedReason.CONN_ERROR, duration
-                    )
-                if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-                    _debug_log(
-                        _CONNECTION_LOGGER,
-                        message=_ConnectionStatusMessage.CHECKOUT_FAILED,
-                        clientId=self._client_id,
-                        serverHost=self.address[0],
-                        serverPort=self.address[1],
-                        reason="An error occurred while trying to establish a new connection",
-                        error=ConnectionCheckOutFailedReason.CONN_ERROR,
-                        durationMS=duration,
-                    )
+                self._telemetry.checkout_failed(
+                    "An error occurred while trying to establish a new connection",
+                    ConnectionCheckOutFailedReason.CONN_ERROR,
+                    checkout_started_time,
+                )
             raise
 
         conn.active = True
@@ -1318,21 +1138,9 @@ def checkin(self, conn: Connection) -> None:
         conn.pinned_txn = False
         conn.pinned_cursor = False
         self.__pinned_sockets.discard(conn)
-        listeners = self.opts._event_listeners
         with self.lock:
             self.active_contexts.discard(conn.cancel_context)
-        if self.enabled_for_cmap:
-            assert listeners is not None
-            listeners.publish_connection_checked_in(self.address, conn.id)
-        if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-            _debug_log(
-                _CONNECTION_LOGGER,
-                message=_ConnectionStatusMessage.CHECKEDIN,
-                clientId=self._client_id,
-                serverHost=self.address[0],
-                serverPort=self.address[1],
-                driverConnectionId=conn.id,
-            )
+        self._telemetry.checked_in(conn.id)
         if self.pid != os.getpid():
             self.reset_without_pause()
         else:
@@ -1340,22 +1148,7 @@ def checkin(self, conn: Connection) -> None:
                 conn.close_conn(ConnectionClosedReason.POOL_CLOSED)
             elif conn.closed:
                 # CMAP requires the closed event be emitted after the check in.
-                if self.enabled_for_cmap:
-                    assert listeners is not None
-                    listeners.publish_connection_closed(
-                        self.address, conn.id, ConnectionClosedReason.ERROR
-                    )
-                if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-                    _debug_log(
-                        _CONNECTION_LOGGER,
-                        message=_ConnectionStatusMessage.CONN_CLOSED,
-                        clientId=self._client_id,
-                        serverHost=self.address[0],
-                        serverPort=self.address[1],
-                        driverConnectionId=conn.id,
-                        reason=_verbose_connection_error_reason(ConnectionClosedReason.ERROR),
-                        error=ConnectionClosedReason.ERROR,
-                    )
+                self._telemetry.connection_closed(conn.id, ConnectionClosedReason.ERROR)
             else:
                 close_conn = False
                 with self.lock:
@@ -1420,24 +1213,11 @@ def _perished(self, conn: Connection) -> bool:
         return False
 
     def _raise_wait_queue_timeout(self, checkout_started_time: float) -> NoReturn:
-        listeners = self.opts._event_listeners
-        duration = time.monotonic() - checkout_started_time
-        if self.enabled_for_cmap:
-            assert listeners is not None
-            listeners.publish_connection_check_out_failed(
-                self.address, ConnectionCheckOutFailedReason.TIMEOUT, duration
-            )
-        if self.enabled_for_logging and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG):
-            _debug_log(
-                _CONNECTION_LOGGER,
-                message=_ConnectionStatusMessage.CHECKOUT_FAILED,
-                clientId=self._client_id,
-                serverHost=self.address[0],
-                serverPort=self.address[1],
-                reason="Wait queue timeout elapsed without a connection becoming available",
-                error=ConnectionCheckOutFailedReason.TIMEOUT,
-                durationMS=duration,
-            )
+        self._telemetry.checkout_failed(
+            "Wait queue timeout elapsed without a connection becoming available",
+            ConnectionCheckOutFailedReason.TIMEOUT,
+            checkout_started_time,
+        )
         timeout = _csot.get_timeout() or self.opts.wait_queue_timeout
         if self.opts.load_balanced:
             other_ops = self.active_sockets - self.ncursors - self.ntxns
diff --git a/pymongo/synchronous/server.py b/pymongo/synchronous/server.py
index 8b0ad4bcb0..ca5eee7b49 100644
--- a/pymongo/synchronous/server.py
+++ b/pymongo/synchronous/server.py
@@ -16,7 +16,6 @@
 
 from __future__ import annotations
 
-import logging
 from contextlib import AbstractContextManager
 from typing import (
     TYPE_CHECKING,
@@ -24,11 +23,7 @@
     Optional,
 )
 
-from pymongo.logger import (
-    _SDAM_LOGGER,
-    _debug_log,
-    _SDAMStatusMessage,
-)
+from pymongo._telemetry import _SdamTelemetry
 
 if TYPE_CHECKING:
     from queue import Queue
@@ -58,12 +53,8 @@ def __init__(
         self._description = server_description
         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]
+        _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]
 
     def open(self) -> None:
         """Start monitoring, or restart after a fork.
@@ -82,23 +73,7 @@ def close(self) -> None:
 
         Reconnect with open().
         """
-        if self._publish:
-            assert self._listener is not None
-            assert self._events is not None
-            self._events.put(
-                (
-                    self._listener.publish_server_closed,
-                    (self._description.address, self._topology_id),
-                )
-            )
-        if _SDAM_LOGGER.isEnabledFor(logging.DEBUG):
-            _debug_log(
-                _SDAM_LOGGER,
-                message=_SDAMStatusMessage.STOP_SERVER,
-                topologyId=self._topology_id,
-                serverHost=self._description.address[0],
-                serverPort=self._description.address[1],
-            )
+        self._sdam.server_closed(self._description.address)
 
         self._monitor.close()
         self._pool.close()
diff --git a/pymongo/synchronous/topology.py b/pymongo/synchronous/topology.py
index b419833256..ff442905fe 100644
--- a/pymongo/synchronous/topology.py
+++ b/pymongo/synchronous/topology.py
@@ -17,7 +17,6 @@
 from __future__ import annotations
 
 import asyncio
-import logging
 import os
 import queue
 import random
@@ -30,6 +29,11 @@
 from typing import TYPE_CHECKING, Any, Callable, Optional, cast
 
 from pymongo import _csot, common, helpers_shared, periodic_executor
+from pymongo._telemetry import (
+    _SdamTelemetry,
+    _ServerSelectionTelemetry,
+    log_server_selection_succeeded,
+)
 from pymongo.errors import (
     ConnectionFailure,
     InvalidOperation,
@@ -47,13 +51,6 @@
     _create_condition,
     _create_lock,
 )
-from pymongo.logger import (
-    _SDAM_LOGGER,
-    _SERVER_SELECTION_LOGGER,
-    _debug_log,
-    _SDAMStatusMessage,
-    _ServerSelectionStatusMessage,
-)
 from pymongo.pool_options import PoolOptions
 from pymongo.server_description import ServerDescription
 from pymongo.server_selectors import (
@@ -108,27 +105,19 @@ 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)
 
-        if _SDAM_LOGGER.isEnabledFor(logging.DEBUG):
-            _debug_log(
-                _SDAM_LOGGER,
-                message=_SDAMStatusMessage.START_TOPOLOGY,
-                topologyId=self._topology_id,
-            )
+        self._sdam = _SdamTelemetry(self._topology_id, self._listeners, self._events)
+        self._sdam.topology_opened()
 
-        if self._publish_tp:
-            assert self._events is not None
-            assert self._listeners is not None
-            self._events.put((self._listeners.publish_topology_opened, (self._topology_id,)))
         self._settings = topology_settings
         topology_description = TopologyDescription(
             topology_settings.get_topology_type(),
@@ -143,37 +132,10 @@ def __init__(self, topology_settings: TopologySettings):
         initial_td = TopologyDescription(
             TOPOLOGY_TYPE.Unknown, {}, None, None, None, self._settings
         )
-        if self._publish_tp:
-            assert self._events is not None
-            assert self._listeners is not None
-            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,
-                message=_SDAMStatusMessage.TOPOLOGY_CHANGE,
-                topologyId=self._topology_id,
-                previousDescription=repr(initial_td),
-                newDescription=repr(self._description),
-            )
+        self._sdam.topology_description_changed(initial_td, self._description)
 
         for seed in topology_settings.seeds:
-            if self._publish_server:
-                assert self._events is not None
-                assert self._listeners 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,
-                    message=_SDAMStatusMessage.START_SERVER,
-                    topologyId=self._topology_id,
-                    serverHost=seed[0],
-                    serverPort=seed[1],
-                )
+            self._sdam.server_opened(seed)
 
         # Store the seed list to help diagnose errors in _error_message().
         self._seed_addresses = list(topology_description.server_descriptions())
@@ -188,7 +150,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]]
 
@@ -321,17 +283,10 @@ 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,
@@ -345,32 +300,13 @@ 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), self.description)
                 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
 
             self._ensure_opened()
@@ -435,18 +371,15 @@ 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],
-            )
+        log_server_selection_succeeded(
+            self._topology_id,
+            selector,
+            operation,
+            operation_id,
+            self.description,
+            server.description.address[0],
+            server.description.address[1],
+        )
         return server
 
     def select_server_by_address(
@@ -508,36 +441,16 @@ def _process_change(
                 server.pool.ready()
 
         suppress_event = sd_old == server_description
-        if self._publish_server and not suppress_event:
-            assert self._events is not None
-            assert self._listeners is not None
-            self._events.put(
-                (
-                    self._listeners.publish_server_description_changed,
-                    (sd_old, server_description, server_description.address, self._topology_id),
-                )
+        if not suppress_event:
+            self._sdam.server_description_changed(
+                sd_old, server_description, server_description.address
             )
 
         self._description = new_td
         self._update_servers()
 
-        if self._publish_tp and not suppress_event:
-            assert self._events is not None
-            assert self._listeners is not None
-            self._events.put(
-                (
-                    self._listeners.publish_topology_description_changed,
-                    (td_old, self._description, self._topology_id),
-                )
-            )
-        if _SDAM_LOGGER.isEnabledFor(logging.DEBUG) and not suppress_event:
-            _debug_log(
-                _SDAM_LOGGER,
-                message=_SDAMStatusMessage.TOPOLOGY_CHANGE,
-                topologyId=self._topology_id,
-                previousDescription=repr(td_old),
-                newDescription=repr(self._description),
-            )
+        if not suppress_event:
+            self._sdam.topology_description_changed(td_old, self._description)
 
         # Shutdown SRV polling for unsupported cluster types.
         # This is only applicable if the old topology was Unknown, and the
@@ -588,24 +501,7 @@ def _process_srv_update(self, seedlist: list[tuple[str, Any]]) -> None:
         self._description = _updated_topology_description_srv_polling(self._description, seedlist)
 
         self._update_servers()
-
-        if self._publish_tp:
-            assert self._events is not None
-            assert self._listeners is not None
-            self._events.put(
-                (
-                    self._listeners.publish_topology_description_changed,
-                    (td_old, self._description, self._topology_id),
-                )
-            )
-        if _SDAM_LOGGER.isEnabledFor(logging.DEBUG):
-            _debug_log(
-                _SDAM_LOGGER,
-                message=_SDAMStatusMessage.TOPOLOGY_CHANGE,
-                topologyId=self._topology_id,
-                previousDescription=repr(td_old),
-                newDescription=repr(self._description),
-            )
+        self._sdam.topology_description_changed(td_old, self._description)
 
     def on_srv_update(self, seedlist: list[tuple[str, Any]]) -> None:
         """Process a new list of nodes obtained from scanning SRV records."""
@@ -741,9 +637,7 @@ def close(self) -> None:
             self._closed = True
 
         # Publish only after releasing the lock.
-        if self._publish_tp:
-            assert self._events is not None
-            assert self._listeners is not None
+        if self._sdam._publish_tp:
             self._description = TopologyDescription(
                 TOPOLOGY_TYPE.Unknown,
                 {},
@@ -752,30 +646,9 @@ def close(self) -> None:
                 self._description.max_election_id,
                 self._description._topology_settings,
             )
-            self._events.put(
-                (
-                    self._listeners.publish_topology_description_changed,
-                    (
-                        old_td,
-                        self._description,
-                        self._topology_id,
-                    ),
-                )
-            )
-            self._events.put((self._listeners.publish_topology_closed, (self._topology_id,)))
-        if _SDAM_LOGGER.isEnabledFor(logging.DEBUG):
-            _debug_log(
-                _SDAM_LOGGER,
-                message=_SDAMStatusMessage.TOPOLOGY_CHANGE,
-                topologyId=self._topology_id,
-                previousDescription=repr(old_td),
-                newDescription=repr(self._description),
-            )
-            _debug_log(
-                _SDAM_LOGGER, message=_SDAMStatusMessage.STOP_TOPOLOGY, topologyId=self._topology_id
-            )
+        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()
             self.__events_executor.join(1)
@@ -816,7 +689,7 @@ def _ensure_opened(self) -> None:
             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.
@@ -955,7 +828,7 @@ 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,
diff --git a/test/asynchronous/utils.py b/test/asynchronous/utils.py
index 5842224220..32e8c89502 100644
--- a/test/asynchronous/utils.py
+++ b/test/asynchronous/utils.py
@@ -49,12 +49,8 @@ async def async_get_pool(client: AsyncMongoClient) -> Pool:
 
 async def async_get_pools(client: AsyncMongoClient) -> list[Pool]:
     """Get all pools."""
-    return [
-        server.pool
-        for server in await (await client._get_topology()).select_servers(
-            any_server_selector, _Op.TEST
-        )
-    ]
+    servers = await (await client._get_topology()).select_servers(any_server_selector, _Op.TEST)
+    return [server.pool for server in servers]
 
 
 async def async_wait_until(predicate, success_description, timeout=10):
diff --git a/test/utils.py b/test/utils.py
index 54b7e351e3..2e67008431 100644
--- a/test/utils.py
+++ b/test/utils.py
@@ -49,10 +49,8 @@ def get_pool(client: MongoClient) -> Pool:
 
 def get_pools(client: MongoClient) -> list[Pool]:
     """Get all pools."""
-    return [
-        server.pool
-        for server in (client._get_topology()).select_servers(any_server_selector, _Op.TEST)
-    ]
+    servers = (client._get_topology()).select_servers(any_server_selector, _Op.TEST)
+    return [server.pool for server in servers]
 
 
 def wait_until(predicate, success_description, timeout=10):