PYTHON-5977 APM/logging hot-path optimizations

pymongo/_telemetry.py

@@ -23,6 +23,7 @@
 from collections.abc import MutableMapping
 from typing import TYPE_CHECKING, Any, Optional
 
+from pymongo import _op_id
 from pymongo.logger import (
     _COMMAND_LOGGER,
     _CONNECTION_LOGGER,
@@ -55,20 +56,36 @@ def _monotonic_duration(start: float) -> float:
     return max(0.0, time.monotonic() - start)
 
 
+def _should_generate_op_id(listeners: Optional[_EventListeners]) -> bool:
+    """Return True if an operation id would be consumed by APM command events
+    or command/server-selection log entries; generating one is otherwise wasted work.
+    """
+    return (
+        (listeners is not None and listeners.enabled_for_commands)
+        or _COMMAND_LOGGER.isEnabledFor(logging.DEBUG)
+        or _SERVER_SELECTION_LOGGER.isEnabledFor(logging.DEBUG)
+    )
+
+
 class _CommandTelemetry:
     """Combines structured logging and APM event publishing for a single command.
 
     Construct once per command, call :meth:`started` before the network send,
     then call :meth:`succeeded` or :meth:`failed` when the outcome is known.
     Duration is measured from the :meth:`started` call.
+
+    This sits on the hot path of every command: when neither APM nor command
+    logging is enabled, only the gate flags and the monotonic duration clock
+    are maintained — the identifying fields are not stored and the ``OP_ID``
+    contextvar is not read.
     """
 
     __slots__ = (
         "_active",
         "_cmd",
         "_conn",
         "_dbname",
-        "_duration",
+        "_duration_s",
         "_listeners",
         "_name",
         "_op_id",
@@ -88,20 +105,23 @@ def __init__(
         dbname: str,
         request_id: int,
         op_id: Optional[int],
+        name: Optional[str] = None,
     ) -> None:
-        self._topology_id = topology_id
         self._should_log = topology_id is not None and _COMMAND_LOGGER.isEnabledFor(logging.DEBUG)
         self._publish = listeners is not None and listeners.enabled_for_commands
         self._active = self._should_log or self._publish
+        self._start = 0.0
+        self._duration_s = 0.0
+        if not self._active:
+            return
+        self._topology_id = topology_id
         self._listeners = listeners
         self._conn = conn
         self._cmd = cmd
-        self._name = next(iter(cmd))
+        self._name = name if name is not None else next(iter(cmd))
         self._dbname = dbname
         self._request_id = request_id
-        self._op_id = op_id
-        self._start: datetime.datetime
-        self._duration: datetime.timedelta
+        self._op_id = op_id if op_id is not None else _op_id.OP_ID.get()
 
     def _emit_log(self, message: _CommandStatusMessage, **extra: Any) -> None:
         _debug_log(
@@ -122,7 +142,7 @@ def _emit_log(self, message: _CommandStatusMessage, **extra: Any) -> None:
 
     def started(self, orig: MutableMapping[str, Any], ensure_db: bool) -> None:
         """Emit the STARTED log entry and APM event, and start the duration clock."""
-        self._start = datetime.datetime.now()
+        self._start = time.monotonic()
         if not self._active:
             return
         if self._should_log:
@@ -142,9 +162,9 @@ def started(self, orig: MutableMapping[str, Any], ensure_db: bool) -> None:
             )
 
     @property
-    def duration(self) -> datetime.timedelta:
-        """Duration from :meth:`started` to :meth:`succeeded` or :meth:`failed`."""
-        return self._duration
+    def duration_s(self) -> float:
+        """Duration in seconds from :meth:`started` to :meth:`succeeded` or :meth:`failed`."""
+        return self._duration_s
 
     def succeeded(
         self,
@@ -153,20 +173,21 @@ def succeeded(
         speculative_hello: bool,
     ) -> None:
         """Emit the SUCCEEDED log entry and APM event."""
-        self._duration = datetime.datetime.now() - self._start
+        self._duration_s = _monotonic_duration(self._start)
         if not self._active:
             return
+        duration = datetime.timedelta(seconds=self._duration_s)
         if self._should_log:
             self._emit_log(
                 _CommandStatusMessage.SUCCEEDED,
-                durationMS=self._duration,
+                durationMS=duration,
                 reply=reply,
                 speculative_authenticate=speculative_hello,
             )
         if self._publish:
             assert self._listeners is not None
             self._listeners.publish_command_success(
-                self._duration,
+                duration,
                 reply,
                 command_name,
                 self._request_id,
@@ -185,20 +206,21 @@ def failed(
         is_server_side_error: bool,
     ) -> None:
         """Emit the FAILED log entry and APM event."""
-        self._duration = datetime.datetime.now() - self._start
+        self._duration_s = _monotonic_duration(self._start)
         if not self._active:
             return
+        duration = datetime.timedelta(seconds=self._duration_s)
         if self._should_log:
             self._emit_log(
                 _CommandStatusMessage.FAILED,
-                durationMS=self._duration,
+                durationMS=duration,
                 failure=failure,
                 isServerSideError=is_server_side_error,
             )
         if self._publish:
             assert self._listeners is not None
             self._listeners.publish_command_failure(
-                self._duration,
+                duration,
                 failure,
                 command_name,
                 self._request_id,
@@ -232,13 +254,18 @@ def __init__(
         self._client_id = client_id
         self._address = address
         self._listeners = listeners
-        self._publish = publish
+        # The CMAP listener set is fixed once the client is constructed
+        # (_EventListeners copies the global listeners at __init__), so this
+        # gate is static for the life of the pool.
+        # NOTE: the checkout/checkin fast paths in pool.py read _publish and
+        # _log directly and inline the "_should_publish or _should_log" gate;
+        # keep them in sync with any change to this gating logic.
+        self._publish = publish and listeners is not None and listeners.enabled_for_cmap
         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
+        return self._publish
 
     @property
     def _should_log(self) -> bool:

pymongo/asynchronous/bulk.py

@@ -32,6 +32,7 @@
 from bson.objectid import ObjectId
 from bson.raw_bson import RawBSONDocument
 from pymongo import _csot, common
+from pymongo._telemetry import _should_generate_op_id
 from pymongo.asynchronous.client_session import AsyncClientSession, _validate_session_write_concern
 from pymongo.asynchronous.command_runner import (
     run_bulk_write_command,
@@ -339,7 +340,7 @@ async def _execute_command(
         write_concern: WriteConcern,
         session: Optional[AsyncClientSession],
         conn: AsyncConnection,
-        op_id: int,
+        op_id: Optional[int],
         retryable: bool,
         full_result: MutableMapping[str, Any],
         final_write_concern: Optional[WriteConcern] = None,
@@ -455,7 +456,8 @@ async def execute_command(
             "nRemoved": 0,
             "upserted": [],
         }
-        op_id = _randint()
+        client = self.collection.database.client
+        op_id = _randint() if _should_generate_op_id(client._event_listeners) else None
 
         async def retryable_bulk(
             session: Optional[AsyncClientSession], conn: AsyncConnection, retryable: bool
@@ -470,7 +472,6 @@ async def retryable_bulk(
                 full_result,
             )
 
-        client = self.collection.database.client
         _ = await client._retryable_write(
             self.is_retryable,
             retryable_bulk,
@@ -491,7 +492,7 @@ async def execute_op_msg_no_results(
         db_name = self.collection.database.name
         client = self.collection.database.client
         listeners = client._event_listeners
-        op_id = _randint()
+        op_id = _randint() if _should_generate_op_id(listeners) else None
 
         if not self.current_run:
             self.current_run = next(generator)
@@ -544,7 +545,11 @@ async def execute_command_no_results(
         # processing at the first error, even when the application
         # specified unacknowledged writeConcern.
         initial_write_concern = WriteConcern()
-        op_id = _randint()
+        op_id = (
+            _randint()
+            if _should_generate_op_id(self.collection.database.client._event_listeners)
+            else None
+        )
         try:
             await self._execute_command(
                 generator,

pymongo/asynchronous/client_bulk.py

@@ -32,6 +32,7 @@
 from bson.objectid import ObjectId
 from bson.raw_bson import RawBSONDocument
 from pymongo import _csot, common
+from pymongo._telemetry import _should_generate_op_id
 from pymongo.asynchronous.client_session import (
     AsyncClientSession,
     _validate_session_write_concern,
@@ -370,7 +371,7 @@ async def _execute_command(
         write_concern: WriteConcern,
         session: Optional[AsyncClientSession],
         conn: AsyncConnection,
-        op_id: int,
+        op_id: Optional[int],
         retryable: bool,
         full_result: MutableMapping[str, Any],
         final_write_concern: Optional[WriteConcern] = None,
@@ -524,7 +525,7 @@ async def execute_command(
             "updateResults": {},
             "deleteResults": {},
         }
-        op_id = _randint()
+        op_id = _randint() if _should_generate_op_id(self.client._event_listeners) else None
 
         async def retryable_bulk(
             session: Optional[AsyncClientSession],
@@ -565,7 +566,7 @@ async def execute_command_unack(
         db_name = "admin"
         cmd_name = "bulkWrite"
         listeners = self.client._event_listeners
-        op_id = _randint()
+        op_id = _randint() if _should_generate_op_id(listeners) else None
 
         bwc = self.bulk_ctx_class(
             db_name,

pymongo/asynchronous/command_runner.py

@@ -36,6 +36,8 @@
 from __future__ import annotations
 
 import datetime
+import logging
+import time
 from collections.abc import Mapping, MutableMapping, Sequence
 from typing import (
     TYPE_CHECKING,
@@ -47,10 +49,11 @@
 )
 
 from bson import _decode_all_selective
-from pymongo import _csot, _op_id, helpers_shared, message
+from pymongo import _csot, helpers_shared, message
 from pymongo._telemetry import _CommandTelemetry
 from pymongo.compression_support import _NO_COMPRESSION
 from pymongo.errors import NotPrimaryError, OperationFailure
+from pymongo.logger import _COMMAND_LOGGER
 from pymongo.message import _BulkWriteContextBase, _convert_exception, _OpMsg
 from pymongo.monitoring import _is_speculative_authenticate
 
@@ -100,8 +103,9 @@ async def _run_command(
     set_conn_more_to_come: bool = False,
     unpack_res: Optional[Callable[..., Any]] = None,
     cursor_id: Optional[int] = None,
-) -> tuple[list[dict[str, Any]], Optional[_OpMsg], datetime.timedelta]:
-    """Send ``msg`` over ``conn`` and return ``(docs, reply, duration)``.
+) -> tuple[list[dict[str, Any]], Optional[_OpMsg], float]:
+    """Send ``msg`` over ``conn`` and return ``(docs, reply, duration_s)``,
+    where ``duration_s`` is the round-trip duration in seconds.
 
     Private shared implementation. Should not be called directly outside this module. Use :func:`run_command`, :func:`run_bulk_write_command`, or :func:`run_cursor_command` instead.
 
@@ -128,8 +132,8 @@ async def _run_command(
     :param orig: The command document published in the ``STARTED`` APM event;
         defaults to ``cmd`` (differs only when the wire command was mutated,
         e.g. with a read preference or after encryption).
-    :param op_id: The APM operation id; defaults to the ``OP_ID`` contextvar,
-        then ``request_id``.
+    :param op_id: The APM operation id; when ``None`` it is resolved from the
+        ``OP_ID`` contextvar (then ``request_id``) only if APM/logging is enabled.
     :param command_name: The command name for the ``SUCCEEDED``/``FAILED`` APM
         events; defaults to the first key of ``cmd``.
     :param check: Raise OperationFailure on a command error.
@@ -159,11 +163,21 @@ async def _run_command(
         command_name = name
     if orig is None:
         orig = cmd
-    if op_id is None:
-        op_id = _op_id.OP_ID.get()
 
-    telemetry = _CommandTelemetry(topology_id, conn, listeners, cmd, dbname, request_id, op_id)
-    telemetry.started(orig, ensure_db)
+    # Fast path: when neither command logging nor APM command listeners are
+    # active, skip constructing the telemetry object entirely and track the
+    # round-trip duration inline.
+    telemetry: Optional[_CommandTelemetry] = None
+    if (topology_id is not None and _COMMAND_LOGGER.isEnabledFor(logging.DEBUG)) or (
+        listeners is not None and listeners.enabled_for_commands
+    ):
+        telemetry = _CommandTelemetry(
+            topology_id, conn, listeners, cmd, dbname, request_id, op_id, name=name
+        )
+        telemetry.started(orig, ensure_db)
+        start = 0.0
+    else:
+        start = time.monotonic()
 
     reply: Optional[_OpMsg] = None
     docs: list[dict[str, Any]] = [{"ok": 1}]
@@ -211,18 +225,23 @@ async def _run_command(
             failure: _DocumentOut = exc.details  # type: ignore[assignment]
         else:
             failure = _convert_exception(exc)
-        telemetry.failed(failure, command_name, isinstance(exc, OperationFailure))
+        if telemetry is not None:
+            telemetry.failed(failure, command_name, isinstance(exc, OperationFailure))
         raise
 
-    telemetry.succeeded(docs[0], command_name, speculative_hello)
+    if telemetry is not None:
+        telemetry.succeeded(docs[0], command_name, speculative_hello)
+        duration_s = telemetry.duration_s
+    else:
+        duration_s = max(0.0, time.monotonic() - start)
 
     if client and client._encrypter and reply and decrypt_reply:
         decrypted = await client._encrypter.decrypt(reply.raw_command_response())
         docs = cast(
             "list[dict[str, Any]]", _decode_all_selective(decrypted, codec_options, user_fields)
         )
 
-    return docs, reply, telemetry.duration
+    return docs, reply, duration_s
 
 
 async def run_bulk_write_command(
@@ -235,8 +254,8 @@ async def run_bulk_write_command(
     orig: Optional[MutableMapping[str, Any]] = None,
     max_doc_size: int = 0,
     unacknowledged: bool = False,
-) -> tuple[list[dict[str, Any]], Optional[_OpMsg], datetime.timedelta]:
-    """Send a bulk write batch and return ``(docs, reply, duration)``.
+) -> tuple[list[dict[str, Any]], Optional[_OpMsg], float]:
+    """Send a bulk write batch and return ``(docs, reply, duration_s)``.
 
     :param bwc: Bulk write context supplying the connection, session, listeners, etc.
     :param cmd: The encoded command document.
@@ -309,7 +328,7 @@ async def run_cursor_command(
     :param cursor_id: The cursor id passed to ``unpack_res``.
     """
     topology_id = client._topology_id if client is not None else None
-    return await _run_command(
+    docs, reply, duration_s = await _run_command(
         conn,
         cmd,
         dbname,
@@ -329,6 +348,8 @@ async def run_cursor_command(
         unpack_res=unpack_res,
         cursor_id=cursor_id,
     )
+    # The cursor path stores the duration on Response, which expects a timedelta.
+    return docs, reply, datetime.timedelta(seconds=duration_s)
 
 
 async def run_command(