From e46bac00f71105caf502d7642458e6a912ac4f92 Mon Sep 17 00:00:00 2001 From: Noah Stapp Date: Tue, 28 Jul 2026 11:27:54 -0400 Subject: [PATCH 01/12] APM-disabled optimizations --- pymongo/_telemetry.py | 54 ++++++++++++++------ pymongo/asynchronous/bulk.py | 15 ++++-- pymongo/asynchronous/client_bulk.py | 7 +-- pymongo/asynchronous/command_runner.py | 27 +++++----- pymongo/asynchronous/mongo_client.py | 24 ++++----- pymongo/message.py | 6 +-- pymongo/monitoring.py | 5 +- pymongo/synchronous/bulk.py | 15 ++++-- pymongo/synchronous/client_bulk.py | 7 +-- pymongo/synchronous/command_runner.py | 27 +++++----- pymongo/synchronous/mongo_client.py | 24 ++++----- test/asynchronous/test_operation_id_retry.py | 8 ++- test/test_operation_id_retry.py | 8 ++- 13 files changed, 134 insertions(+), 93 deletions(-) diff --git a/pymongo/_telemetry.py b/pymongo/_telemetry.py index f972a402df..2b7ba00fbe 100644 --- a/pymongo/_telemetry.py +++ b/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,12 +56,28 @@ 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__ = ( @@ -68,7 +85,7 @@ class _CommandTelemetry: "_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, diff --git a/pymongo/asynchronous/bulk.py b/pymongo/asynchronous/bulk.py index 3075afa2b3..082da1fabe 100644 --- a/pymongo/asynchronous/bulk.py +++ b/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, diff --git a/pymongo/asynchronous/client_bulk.py b/pymongo/asynchronous/client_bulk.py index cfa2ea9853..f56ab062cb 100644 --- a/pymongo/asynchronous/client_bulk.py +++ b/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, diff --git a/pymongo/asynchronous/command_runner.py b/pymongo/asynchronous/command_runner.py index 683dc0d8e8..540b88304b 100644 --- a/pymongo/asynchronous/command_runner.py +++ b/pymongo/asynchronous/command_runner.py @@ -47,7 +47,7 @@ ) 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 @@ -100,8 +100,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 +129,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,10 +160,10 @@ 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 = _CommandTelemetry( + topology_id, conn, listeners, cmd, dbname, request_id, op_id, name=name + ) telemetry.started(orig, ensure_db) reply: Optional[_OpMsg] = None @@ -222,7 +223,7 @@ async def _run_command( "list[dict[str, Any]]", _decode_all_selective(decrypted, codec_options, user_fields) ) - return docs, reply, telemetry.duration + return docs, reply, telemetry.duration_s async def run_bulk_write_command( @@ -235,8 +236,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 +310,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 +330,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( diff --git a/pymongo/asynchronous/mongo_client.py b/pymongo/asynchronous/mongo_client.py index e93602e4ea..bcd8bb6793 100644 --- a/pymongo/asynchronous/mongo_client.py +++ b/pymongo/asynchronous/mongo_client.py @@ -35,7 +35,6 @@ import asyncio import contextlib -import logging import os import time as time # noqa: PLC0414 # needed in sync version import warnings @@ -57,7 +56,7 @@ from bson.codec_options import DEFAULT_CODEC_OPTIONS, CodecOptions, TypeRegistry from bson.timestamp import Timestamp from pymongo import _csot, _op_id, common, helpers_shared, periodic_executor -from pymongo._telemetry import log_command_retry +from pymongo._telemetry import _should_generate_op_id, 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 @@ -91,8 +90,6 @@ ) from pymongo.logger import ( _CLIENT_LOGGER, - _COMMAND_LOGGER, - _SERVER_SELECTION_LOGGER, _log_client_error, _log_or_warn, ) @@ -2891,14 +2888,7 @@ def __init__( self._deprioritized_servers: Optional[list[Server]] = None self._operation = operation # Only generate an operation id when APM/logging is enabled - if operation_id is None and ( - ( - self._client._event_listeners is not None - and self._client._event_listeners.enabled_for_commands - ) - or _COMMAND_LOGGER.isEnabledFor(logging.DEBUG) - or _SERVER_SELECTION_LOGGER.isEnabledFor(logging.DEBUG) - ): + if operation_id is None and _should_generate_op_id(self._client._event_listeners): operation_id = _randint() self._operation_id = operation_id self._attempt_number = 0 @@ -3142,8 +3132,11 @@ async def _write(self) -> T: # One operation id across all attempts of this operation if APM/logging is enabled if self._operation_id is None: return await self._func(self._session, conn, self._retryable) # type: ignore - with _op_id._OpIdContext(self._operation_id): + token = _op_id.OP_ID.set(self._operation_id) + try: return await self._func(self._session, conn, self._retryable) # type: ignore + finally: + _op_id.OP_ID.reset(token) except PyMongoError as exc: if not self._retryable: raise @@ -3169,8 +3162,11 @@ async def _read(self) -> T: # One operation id across all attempts of this operation if APM/logging is enabled if self._operation_id is None: return await self._func(self._session, self._server, conn, read_pref) # type: ignore - with _op_id._OpIdContext(self._operation_id): + token = _op_id.OP_ID.set(self._operation_id) + try: return await self._func(self._session, self._server, conn, read_pref) # type: ignore + finally: + _op_id.OP_ID.reset(token) def _after_fork_child() -> None: diff --git a/pymongo/message.py b/pymongo/message.py index 2e3aa1dcbd..a41dab2506 100644 --- a/pymongo/message.py +++ b/pymongo/message.py @@ -457,7 +457,7 @@ def __init__( database_name: str, cmd_name: str, conn: _AgnosticConnection, - operation_id: int, + operation_id: Optional[int], listeners: _EventListeners, session: Optional[_AgnosticClientSession], op_type: int, @@ -508,7 +508,7 @@ def __init__( database_name: str, cmd_name: str, conn: _AgnosticConnection, - operation_id: int, + operation_id: Optional[int], listeners: _EventListeners, session: Optional[_AgnosticClientSession], op_type: int, @@ -751,7 +751,7 @@ def __init__( database_name: str, cmd_name: str, conn: _AgnosticConnection, - operation_id: int, + operation_id: Optional[int], listeners: _EventListeners, session: Optional[_AgnosticClientSession], codec: CodecOptions[Any], diff --git a/pymongo/monitoring.py b/pymongo/monitoring.py index a6fcbaffa7..e40b4d71ab 100644 --- a/pymongo/monitoring.py +++ b/pymongo/monitoring.py @@ -516,9 +516,10 @@ def register(listener: _EventListener) -> None: # The "hello" command is also deemed sensitive when attempting speculative # authentication. def _is_speculative_authenticate(command_name: str, doc: Mapping[str, Any]) -> bool: + # Called on every command; probe the dict first, it is cheaper than str.lower(). return bool( - command_name.lower() in ("hello", HelloCompat.LEGACY_CMD) - and "speculativeAuthenticate" in doc + "speculativeAuthenticate" in doc + and command_name.lower() in ("hello", HelloCompat.LEGACY_CMD) ) diff --git a/pymongo/synchronous/bulk.py b/pymongo/synchronous/bulk.py index 36081fe222..c5b7ed8c1d 100644 --- a/pymongo/synchronous/bulk.py +++ b/pymongo/synchronous/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.bulk_shared import ( _COMMANDS, _DELETE_ALL, @@ -339,7 +340,7 @@ def _execute_command( write_concern: WriteConcern, session: Optional[ClientSession], conn: Connection, - op_id: int, + op_id: Optional[int], retryable: bool, full_result: MutableMapping[str, Any], final_write_concern: Optional[WriteConcern] = None, @@ -455,7 +456,8 @@ 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 def retryable_bulk( session: Optional[ClientSession], conn: Connection, retryable: bool @@ -470,7 +472,6 @@ def retryable_bulk( full_result, ) - client = self.collection.database.client _ = client._retryable_write( self.is_retryable, retryable_bulk, @@ -489,7 +490,7 @@ def execute_op_msg_no_results(self, conn: Connection, generator: Iterator[Any]) 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) @@ -542,7 +543,11 @@ 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: self._execute_command( generator, diff --git a/pymongo/synchronous/client_bulk.py b/pymongo/synchronous/client_bulk.py index 4cf1d9dbb0..586b6911f9 100644 --- a/pymongo/synchronous/client_bulk.py +++ b/pymongo/synchronous/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.synchronous.client_session import ( ClientSession, _validate_session_write_concern, @@ -368,7 +369,7 @@ def _execute_command( write_concern: WriteConcern, session: Optional[ClientSession], conn: Connection, - op_id: int, + op_id: Optional[int], retryable: bool, full_result: MutableMapping[str, Any], final_write_concern: Optional[WriteConcern] = None, @@ -522,7 +523,7 @@ def execute_command( "updateResults": {}, "deleteResults": {}, } - op_id = _randint() + op_id = _randint() if _should_generate_op_id(self.client._event_listeners) else None def retryable_bulk( session: Optional[ClientSession], @@ -563,7 +564,7 @@ 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, diff --git a/pymongo/synchronous/command_runner.py b/pymongo/synchronous/command_runner.py index e81f514003..1278220f97 100644 --- a/pymongo/synchronous/command_runner.py +++ b/pymongo/synchronous/command_runner.py @@ -47,7 +47,7 @@ ) 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 @@ -100,8 +100,9 @@ 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 +129,8 @@ 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,10 +160,10 @@ 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 = _CommandTelemetry( + topology_id, conn, listeners, cmd, dbname, request_id, op_id, name=name + ) telemetry.started(orig, ensure_db) reply: Optional[_OpMsg] = None @@ -222,7 +223,7 @@ def _run_command( "list[dict[str, Any]]", _decode_all_selective(decrypted, codec_options, user_fields) ) - return docs, reply, telemetry.duration + return docs, reply, telemetry.duration_s def run_bulk_write_command( @@ -235,8 +236,8 @@ 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 +310,7 @@ 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 _run_command( + docs, reply, duration_s = _run_command( conn, cmd, dbname, @@ -329,6 +330,8 @@ 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) def run_command( diff --git a/pymongo/synchronous/mongo_client.py b/pymongo/synchronous/mongo_client.py index bc77b2bb3c..728ec5a35d 100644 --- a/pymongo/synchronous/mongo_client.py +++ b/pymongo/synchronous/mongo_client.py @@ -35,7 +35,6 @@ import asyncio import contextlib -import logging import os import time as time # noqa: PLC0414 # needed in sync version import warnings @@ -57,7 +56,7 @@ from bson.codec_options import DEFAULT_CODEC_OPTIONS, CodecOptions, TypeRegistry from bson.timestamp import Timestamp from pymongo import _csot, _op_id, common, helpers_shared, periodic_executor -from pymongo._telemetry import log_command_retry +from pymongo._telemetry import _should_generate_op_id, log_command_retry from pymongo.client_options import ClientOptions from pymongo.driver_info import DriverInfo from pymongo.errors import ( @@ -81,8 +80,6 @@ ) from pymongo.logger import ( _CLIENT_LOGGER, - _COMMAND_LOGGER, - _SERVER_SELECTION_LOGGER, _log_client_error, _log_or_warn, ) @@ -2880,14 +2877,7 @@ def __init__( self._deprioritized_servers: Optional[list[Server]] = None self._operation = operation # Only generate an operation id when APM/logging is enabled - if operation_id is None and ( - ( - self._client._event_listeners is not None - and self._client._event_listeners.enabled_for_commands - ) - or _COMMAND_LOGGER.isEnabledFor(logging.DEBUG) - or _SERVER_SELECTION_LOGGER.isEnabledFor(logging.DEBUG) - ): + if operation_id is None and _should_generate_op_id(self._client._event_listeners): operation_id = _randint() self._operation_id = operation_id self._attempt_number = 0 @@ -3131,8 +3121,11 @@ def _write(self) -> T: # One operation id across all attempts of this operation if APM/logging is enabled if self._operation_id is None: return self._func(self._session, conn, self._retryable) # type: ignore - with _op_id._OpIdContext(self._operation_id): + token = _op_id.OP_ID.set(self._operation_id) + try: return self._func(self._session, conn, self._retryable) # type: ignore + finally: + _op_id.OP_ID.reset(token) except PyMongoError as exc: if not self._retryable: raise @@ -3158,8 +3151,11 @@ def _read(self) -> T: # One operation id across all attempts of this operation if APM/logging is enabled if self._operation_id is None: return self._func(self._session, self._server, conn, read_pref) # type: ignore - with _op_id._OpIdContext(self._operation_id): + token = _op_id.OP_ID.set(self._operation_id) + try: return self._func(self._session, self._server, conn, read_pref) # type: ignore + finally: + _op_id.OP_ID.reset(token) def _after_fork_child() -> None: diff --git a/test/asynchronous/test_operation_id_retry.py b/test/asynchronous/test_operation_id_retry.py index e5a1d16c90..43944ec3c5 100644 --- a/test/asynchronous/test_operation_id_retry.py +++ b/test/asynchronous/test_operation_id_retry.py @@ -151,10 +151,14 @@ async def test_retry_without_listeners_or_logging_creates_no_operation_id(self): find_op_ids = [] original_init = _CommandTelemetry.__init__ - def recording_init(self, topology_id, conn, listeners, cmd, dbname, request_id, op_id): + def recording_init( + self, topology_id, conn, listeners, cmd, dbname, request_id, op_id, name=None + ): if next(iter(cmd)) == "find": find_op_ids.append(op_id) - original_init(self, topology_id, conn, listeners, cmd, dbname, request_id, op_id) + original_init( + self, topology_id, conn, listeners, cmd, dbname, request_id, op_id, name=name + ) fail_point = { "mode": {"times": 1}, diff --git a/test/test_operation_id_retry.py b/test/test_operation_id_retry.py index cae8927a37..b298a20e0a 100644 --- a/test/test_operation_id_retry.py +++ b/test/test_operation_id_retry.py @@ -149,10 +149,14 @@ def test_retry_without_listeners_or_logging_creates_no_operation_id(self): find_op_ids = [] original_init = _CommandTelemetry.__init__ - def recording_init(self, topology_id, conn, listeners, cmd, dbname, request_id, op_id): + def recording_init( + self, topology_id, conn, listeners, cmd, dbname, request_id, op_id, name=None + ): if next(iter(cmd)) == "find": find_op_ids.append(op_id) - original_init(self, topology_id, conn, listeners, cmd, dbname, request_id, op_id) + original_init( + self, topology_id, conn, listeners, cmd, dbname, request_id, op_id, name=name + ) fail_point = { "mode": {"times": 1}, From a15d983dc7d41841f5ef764619349c63ba49983e Mon Sep 17 00:00:00 2001 From: Noah Stapp Date: Wed, 29 Jul 2026 09:57:29 -0400 Subject: [PATCH 02/12] Use attempt -1 instead of attempt for backoff --- pymongo/asynchronous/helpers.py | 4 ++-- pymongo/synchronous/helpers.py | 4 ++-- 2 files changed, 4 insertions(+), 4 deletions(-) diff --git a/pymongo/asynchronous/helpers.py b/pymongo/asynchronous/helpers.py index a6c35b2f05..b7ca91101c 100644 --- a/pymongo/asynchronous/helpers.py +++ b/pymongo/asynchronous/helpers.py @@ -90,7 +90,7 @@ def _backoff( max_delay: float = _BACKOFF_MAX, ) -> float: jitter = random.random() # noqa: S311 - return jitter * min(base_backoff * (2**attempt), max_delay) + return jitter * min(base_backoff * (2 ** (attempt)), max_delay) class _RetryPolicy: @@ -109,7 +109,7 @@ def __init__( def backoff(self, attempt: int, base_backoff: Optional[float] = None) -> float: """Return the actual backoff duration for the given attempt and base backoff.""" return _backoff( - max(0, attempt), + max(0, attempt - 1), self.backoff_initial if base_backoff is None or base_backoff < 0 else base_backoff, self.backoff_max, ) diff --git a/pymongo/synchronous/helpers.py b/pymongo/synchronous/helpers.py index 3f0d540c57..94a1371afb 100644 --- a/pymongo/synchronous/helpers.py +++ b/pymongo/synchronous/helpers.py @@ -90,7 +90,7 @@ def _backoff( max_delay: float = _BACKOFF_MAX, ) -> float: jitter = random.random() # noqa: S311 - return jitter * min(base_backoff * (2**attempt), max_delay) + return jitter * min(base_backoff * (2 ** (attempt)), max_delay) class _RetryPolicy: @@ -109,7 +109,7 @@ def __init__( def backoff(self, attempt: int, base_backoff: Optional[float] = None) -> float: """Return the actual backoff duration for the given attempt and base backoff.""" return _backoff( - max(0, attempt), + max(0, attempt - 1), self.backoff_initial if base_backoff is None or base_backoff < 0 else base_backoff, self.backoff_max, ) From 2e67f4822e7601bda99ed4b725ac42bd7fd0a020 Mon Sep 17 00:00:00 2001 From: Noah Stapp Date: Thu, 30 Jul 2026 13:49:17 -0400 Subject: [PATCH 03/12] Final APM optimizations --- pymongo/_telemetry.py | 8 ++- pymongo/asynchronous/command_runner.py | 74 ++++++++++++-------- pymongo/asynchronous/mongo_client.py | 13 +++- pymongo/asynchronous/pool.py | 21 +++++- pymongo/asynchronous/topology.py | 38 ++++++---- pymongo/synchronous/command_runner.py | 74 ++++++++++++-------- pymongo/synchronous/mongo_client.py | 13 +++- pymongo/synchronous/pool.py | 21 +++++- pymongo/synchronous/topology.py | 38 ++++++---- test/asynchronous/test_operation_id_retry.py | 4 +- test/test_operation_id_retry.py | 4 +- 11 files changed, 205 insertions(+), 103 deletions(-) diff --git a/pymongo/_telemetry.py b/pymongo/_telemetry.py index 2b7ba00fbe..176b8e3b89 100644 --- a/pymongo/_telemetry.py +++ b/pymongo/_telemetry.py @@ -254,13 +254,15 @@ 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. + 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: diff --git a/pymongo/asynchronous/command_runner.py b/pymongo/asynchronous/command_runner.py index 540b88304b..78d2ad3525 100644 --- a/pymongo/asynchronous/command_runner.py +++ b/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, @@ -51,6 +53,7 @@ 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 @@ -77,7 +80,6 @@ async def _run_command( dbname: str, request_id: int, msg: bytes, - *, client: Optional[AsyncMongoClient[Any]], session: Optional[AsyncClientSession], listeners: Optional[_EventListeners], @@ -85,19 +87,20 @@ async def _run_command( codec_options: CodecOptions[_DocumentType], user_fields: Optional[Mapping[str, Any]] = None, orig: Optional[MutableMapping[str, Any]] = None, - op_id: Optional[int] = None, - command_name: Optional[str] = None, check: bool = True, allowable_errors: Optional[Sequence[Union[str, int]]] = None, parse_write_concern_error: bool = False, - pool_opts: Optional[PoolOptions] = None, - unacknowledged: bool = False, speculative_hello: bool = False, + unacknowledged: bool = False, + set_conn_more_to_come: bool = False, + *, + op_id: Optional[int] = None, + command_name: Optional[str] = None, + pool_opts: Optional[PoolOptions] = None, ensure_db: bool = False, decrypt_reply: bool = True, max_doc_size: int = 0, more_to_come: bool = False, - 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], float]: @@ -161,10 +164,20 @@ async def _run_command( if orig is None: orig = cmd - telemetry = _CommandTelemetry( - topology_id, conn, listeners, cmd, dbname, request_id, op_id, name=name - ) - 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}] @@ -212,10 +225,15 @@ 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()) @@ -223,7 +241,7 @@ async def _run_command( "list[dict[str, Any]]", _decode_all_selective(decrypted, codec_options, user_fields) ) - return docs, reply, telemetry.duration_s + return docs, reply, duration_s async def run_bulk_write_command( @@ -256,11 +274,11 @@ async def run_bulk_write_command( bwc.db_name, request_id, msg, - client=client, - session=bwc.session, # type: ignore[arg-type] - listeners=bwc.listeners, - topology_id=topology_id, - codec_options=bwc.codec, + client, + bwc.session, # type: ignore[arg-type] + bwc.listeners, + topology_id, + bwc.codec, op_id=bwc.op_id, command_name=bwc.name, orig=orig, @@ -316,11 +334,11 @@ async def run_cursor_command( dbname, request_id, msg, - client=client, - session=session, - listeners=listeners, - topology_id=topology_id, - codec_options=codec_options, + client, + session, + listeners, + topology_id, + codec_options, user_fields=user_fields, command_name=command_name, pool_opts=pool_opts, @@ -427,11 +445,11 @@ async def run_command( dbname, request_id, msg, - client=client, - session=session, - listeners=listeners, - topology_id=topology_id, - codec_options=codec_options, + client, + session, + listeners, + topology_id, + codec_options, user_fields=user_fields, orig=orig, check=check, diff --git a/pymongo/asynchronous/mongo_client.py b/pymongo/asynchronous/mongo_client.py index bcd8bb6793..a38e14190f 100644 --- a/pymongo/asynchronous/mongo_client.py +++ b/pymongo/asynchronous/mongo_client.py @@ -35,6 +35,7 @@ import asyncio import contextlib +import logging import os import time as time # noqa: PLC0414 # needed in sync version import warnings @@ -56,7 +57,7 @@ from bson.codec_options import DEFAULT_CODEC_OPTIONS, CodecOptions, TypeRegistry from bson.timestamp import Timestamp from pymongo import _csot, _op_id, common, helpers_shared, periodic_executor -from pymongo._telemetry import _should_generate_op_id, log_command_retry +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,6 +91,8 @@ ) from pymongo.logger import ( _CLIENT_LOGGER, + _COMMAND_LOGGER, + _SERVER_SELECTION_LOGGER, _log_client_error, _log_or_warn, ) @@ -2888,7 +2891,13 @@ def __init__( self._deprioritized_servers: Optional[list[Server]] = None self._operation = operation # Only generate an operation id when APM/logging is enabled - if operation_id is None and _should_generate_op_id(self._client._event_listeners): + # (_should_generate_op_id inlined: this runs once per operation). + listeners = self._client._event_listeners + if operation_id is None and ( + (listeners is not None and listeners.enabled_for_commands) + or _COMMAND_LOGGER.isEnabledFor(logging.DEBUG) + or _SERVER_SELECTION_LOGGER.isEnabledFor(logging.DEBUG) + ): operation_id = _randint() self._operation_id = operation_id self._attempt_number = 0 diff --git a/pymongo/asynchronous/pool.py b/pymongo/asynchronous/pool.py index fdf3b1d816..37fd6edf1d 100644 --- a/pymongo/asynchronous/pool.py +++ b/pymongo/asynchronous/pool.py @@ -16,6 +16,7 @@ import asyncio import collections +import logging import os import socket import time @@ -62,6 +63,7 @@ _async_create_condition, _async_create_lock, ) +from pymongo.logger import _CONNECTION_LOGGER from pymongo.monitoring import ( ConnectionCheckOutFailedReason, ConnectionClosedReason, @@ -1113,7 +1115,11 @@ async def checkin(self, conn: AsyncConnection) -> None: self._pinned_sockets.discard(conn) async with self.lock: self.active_contexts.discard(conn.cancel_context) - self._telemetry.checked_in(conn.id) + telemetry = self._telemetry + if telemetry._publish or ( + telemetry._log and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG) + ): + telemetry.checked_in(conn.id) if self.pid != os.getpid(): await self.reset_without_pause() else: @@ -1230,12 +1236,21 @@ def __init__( async def __aenter__(self) -> AsyncConnection: pool = self._pool - checkout_started_time = pool._telemetry.checkout_started() + telemetry = pool._telemetry + # Fast path: when neither CMAP listeners nor connection logging are + # active, skip the checkout started/succeeded telemetry calls. + if not telemetry._publish and not ( + telemetry._log and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG) + ): + conn = await pool._get_conn(time.monotonic(), handler=self._handler) + self._conn = conn + return conn + checkout_started_time = telemetry.checkout_started() conn = await pool._get_conn(checkout_started_time, handler=self._handler) self._conn = conn try: - pool._telemetry.checkout_succeeded(conn.id, checkout_started_time) + telemetry.checkout_succeeded(conn.id, checkout_started_time) except BaseException: await pool.checkin(conn) self._conn = None diff --git a/pymongo/asynchronous/topology.py b/pymongo/asynchronous/topology.py index ca8244c02a..d00f2ea4ed 100644 --- a/pymongo/asynchronous/topology.py +++ b/pymongo/asynchronous/topology.py @@ -17,6 +17,7 @@ from __future__ import annotations import asyncio +import logging import os import queue import random @@ -55,6 +56,7 @@ _async_create_condition, _async_create_lock, ) +from pymongo.logger import _SERVER_SELECTION_LOGGER from pymongo.pool_options import PoolOptions from pymongo.server_description import ServerDescription from pymongo.server_selectors import ( @@ -283,10 +285,14 @@ async def _select_servers_loop( now = time.monotonic() end_time = now + timeout logged_waiting = False - ss = _ServerSelectionTelemetry( - self._topology_id, selector, operation, operation_id, self.description - ) - ss.started() + # Server selection defines only log entries (no APM events); skip the + # telemetry object entirely when the logger is not enabled. + ss: Optional[_ServerSelectionTelemetry] = None + if _SERVER_SELECTION_LOGGER.isEnabledFor(logging.DEBUG): + ss = _ServerSelectionTelemetry( + self._topology_id, selector, operation, operation_id, self.description + ) + ss.started() server_descriptions = self._description.apply_selector( selector, @@ -300,12 +306,13 @@ async def _select_servers_loop( while not server_descriptions: # No suitable servers. if timeout == 0 or now > end_time: - ss.failed(self._error_message(selector), self.description) + if ss is not None: + 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: + if ss is not None and not logged_waiting: ss.waiting(int(1000 * (end_time - time.monotonic()))) logged_waiting = True @@ -371,15 +378,16 @@ async def select_server( ) if _csot.get_timeout(): _csot.set_rtt(server.description.min_round_trip_time) - log_server_selection_succeeded( - self._topology_id, - selector, - operation, - operation_id, - self.description, - server.description.address[0], - server.description.address[1], - ) + if _SERVER_SELECTION_LOGGER.isEnabledFor(logging.DEBUG): + 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( diff --git a/pymongo/synchronous/command_runner.py b/pymongo/synchronous/command_runner.py index 1278220f97..08f8e5afe8 100644 --- a/pymongo/synchronous/command_runner.py +++ b/pymongo/synchronous/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, @@ -51,6 +53,7 @@ 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 @@ -77,7 +80,6 @@ def _run_command( dbname: str, request_id: int, msg: bytes, - *, client: Optional[MongoClient[Any]], session: Optional[ClientSession], listeners: Optional[_EventListeners], @@ -85,19 +87,20 @@ def _run_command( codec_options: CodecOptions[_DocumentType], user_fields: Optional[Mapping[str, Any]] = None, orig: Optional[MutableMapping[str, Any]] = None, - op_id: Optional[int] = None, - command_name: Optional[str] = None, check: bool = True, allowable_errors: Optional[Sequence[Union[str, int]]] = None, parse_write_concern_error: bool = False, - pool_opts: Optional[PoolOptions] = None, - unacknowledged: bool = False, speculative_hello: bool = False, + unacknowledged: bool = False, + set_conn_more_to_come: bool = False, + *, + op_id: Optional[int] = None, + command_name: Optional[str] = None, + pool_opts: Optional[PoolOptions] = None, ensure_db: bool = False, decrypt_reply: bool = True, max_doc_size: int = 0, more_to_come: bool = False, - 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], float]: @@ -161,10 +164,20 @@ def _run_command( if orig is None: orig = cmd - telemetry = _CommandTelemetry( - topology_id, conn, listeners, cmd, dbname, request_id, op_id, name=name - ) - 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}] @@ -212,10 +225,15 @@ 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 = client._encrypter.decrypt(reply.raw_command_response()) @@ -223,7 +241,7 @@ def _run_command( "list[dict[str, Any]]", _decode_all_selective(decrypted, codec_options, user_fields) ) - return docs, reply, telemetry.duration_s + return docs, reply, duration_s def run_bulk_write_command( @@ -256,11 +274,11 @@ def run_bulk_write_command( bwc.db_name, request_id, msg, - client=client, - session=bwc.session, # type: ignore[arg-type] - listeners=bwc.listeners, - topology_id=topology_id, - codec_options=bwc.codec, + client, + bwc.session, # type: ignore[arg-type] + bwc.listeners, + topology_id, + bwc.codec, op_id=bwc.op_id, command_name=bwc.name, orig=orig, @@ -316,11 +334,11 @@ def run_cursor_command( dbname, request_id, msg, - client=client, - session=session, - listeners=listeners, - topology_id=topology_id, - codec_options=codec_options, + client, + session, + listeners, + topology_id, + codec_options, user_fields=user_fields, command_name=command_name, pool_opts=pool_opts, @@ -427,11 +445,11 @@ def run_command( dbname, request_id, msg, - client=client, - session=session, - listeners=listeners, - topology_id=topology_id, - codec_options=codec_options, + client, + session, + listeners, + topology_id, + codec_options, user_fields=user_fields, orig=orig, check=check, diff --git a/pymongo/synchronous/mongo_client.py b/pymongo/synchronous/mongo_client.py index 728ec5a35d..e14c4514ec 100644 --- a/pymongo/synchronous/mongo_client.py +++ b/pymongo/synchronous/mongo_client.py @@ -35,6 +35,7 @@ import asyncio import contextlib +import logging import os import time as time # noqa: PLC0414 # needed in sync version import warnings @@ -56,7 +57,7 @@ from bson.codec_options import DEFAULT_CODEC_OPTIONS, CodecOptions, TypeRegistry from bson.timestamp import Timestamp from pymongo import _csot, _op_id, common, helpers_shared, periodic_executor -from pymongo._telemetry import _should_generate_op_id, log_command_retry +from pymongo._telemetry import log_command_retry from pymongo.client_options import ClientOptions from pymongo.driver_info import DriverInfo from pymongo.errors import ( @@ -80,6 +81,8 @@ ) from pymongo.logger import ( _CLIENT_LOGGER, + _COMMAND_LOGGER, + _SERVER_SELECTION_LOGGER, _log_client_error, _log_or_warn, ) @@ -2877,7 +2880,13 @@ def __init__( self._deprioritized_servers: Optional[list[Server]] = None self._operation = operation # Only generate an operation id when APM/logging is enabled - if operation_id is None and _should_generate_op_id(self._client._event_listeners): + # (_should_generate_op_id inlined: this runs once per operation). + listeners = self._client._event_listeners + if operation_id is None and ( + (listeners is not None and listeners.enabled_for_commands) + or _COMMAND_LOGGER.isEnabledFor(logging.DEBUG) + or _SERVER_SELECTION_LOGGER.isEnabledFor(logging.DEBUG) + ): operation_id = _randint() self._operation_id = operation_id self._attempt_number = 0 diff --git a/pymongo/synchronous/pool.py b/pymongo/synchronous/pool.py index 1304921781..54a31e43d2 100644 --- a/pymongo/synchronous/pool.py +++ b/pymongo/synchronous/pool.py @@ -16,6 +16,7 @@ import asyncio import collections +import logging import os import socket import time @@ -59,6 +60,7 @@ _create_condition, _create_lock, ) +from pymongo.logger import _CONNECTION_LOGGER from pymongo.monitoring import ( ConnectionCheckOutFailedReason, ConnectionClosedReason, @@ -1109,7 +1111,11 @@ def checkin(self, conn: Connection) -> None: self._pinned_sockets.discard(conn) with self.lock: self.active_contexts.discard(conn.cancel_context) - self._telemetry.checked_in(conn.id) + telemetry = self._telemetry + if telemetry._publish or ( + telemetry._log and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG) + ): + telemetry.checked_in(conn.id) if self.pid != os.getpid(): self.reset_without_pause() else: @@ -1226,12 +1232,21 @@ def __init__( def __enter__(self) -> Connection: pool = self._pool - checkout_started_time = pool._telemetry.checkout_started() + telemetry = pool._telemetry + # Fast path: when neither CMAP listeners nor connection logging are + # active, skip the checkout started/succeeded telemetry calls. + if not telemetry._publish and not ( + telemetry._log and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG) + ): + conn = pool._get_conn(time.monotonic(), handler=self._handler) + self._conn = conn + return conn + checkout_started_time = telemetry.checkout_started() conn = pool._get_conn(checkout_started_time, handler=self._handler) self._conn = conn try: - pool._telemetry.checkout_succeeded(conn.id, checkout_started_time) + telemetry.checkout_succeeded(conn.id, checkout_started_time) except BaseException: pool.checkin(conn) self._conn = None diff --git a/pymongo/synchronous/topology.py b/pymongo/synchronous/topology.py index ff442905fe..ab6bb05cb6 100644 --- a/pymongo/synchronous/topology.py +++ b/pymongo/synchronous/topology.py @@ -17,6 +17,7 @@ from __future__ import annotations import asyncio +import logging import os import queue import random @@ -51,6 +52,7 @@ _create_condition, _create_lock, ) +from pymongo.logger import _SERVER_SELECTION_LOGGER from pymongo.pool_options import PoolOptions from pymongo.server_description import ServerDescription from pymongo.server_selectors import ( @@ -283,10 +285,14 @@ def _select_servers_loop( now = time.monotonic() end_time = now + timeout logged_waiting = False - ss = _ServerSelectionTelemetry( - self._topology_id, selector, operation, operation_id, self.description - ) - ss.started() + # Server selection defines only log entries (no APM events); skip the + # telemetry object entirely when the logger is not enabled. + ss: Optional[_ServerSelectionTelemetry] = None + if _SERVER_SELECTION_LOGGER.isEnabledFor(logging.DEBUG): + ss = _ServerSelectionTelemetry( + self._topology_id, selector, operation, operation_id, self.description + ) + ss.started() server_descriptions = self._description.apply_selector( selector, @@ -300,12 +306,13 @@ def _select_servers_loop( while not server_descriptions: # No suitable servers. if timeout == 0 or now > end_time: - ss.failed(self._error_message(selector), self.description) + if ss is not None: + 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: + if ss is not None and not logged_waiting: ss.waiting(int(1000 * (end_time - time.monotonic()))) logged_waiting = True @@ -371,15 +378,16 @@ def select_server( ) if _csot.get_timeout(): _csot.set_rtt(server.description.min_round_trip_time) - log_server_selection_succeeded( - self._topology_id, - selector, - operation, - operation_id, - self.description, - server.description.address[0], - server.description.address[1], - ) + if _SERVER_SELECTION_LOGGER.isEnabledFor(logging.DEBUG): + 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( diff --git a/test/asynchronous/test_operation_id_retry.py b/test/asynchronous/test_operation_id_retry.py index 43944ec3c5..ce1c82207b 100644 --- a/test/asynchronous/test_operation_id_retry.py +++ b/test/asynchronous/test_operation_id_retry.py @@ -182,8 +182,8 @@ def recording_init( ) self.assertEqual( find_op_ids, - [None, None], - "expected two attempts, neither carrying a shared operation id", + [], + "expected no _CommandTelemetry construction without APM/logging enabled", ) async def test_reauth_does_not_reuse_operation_id(self): diff --git a/test/test_operation_id_retry.py b/test/test_operation_id_retry.py index b298a20e0a..1253aaee7f 100644 --- a/test/test_operation_id_retry.py +++ b/test/test_operation_id_retry.py @@ -180,8 +180,8 @@ def recording_init( ) self.assertEqual( find_op_ids, - [None, None], - "expected two attempts, neither carrying a shared operation id", + [], + "expected no _CommandTelemetry construction without APM/logging enabled", ) def test_reauth_does_not_reuse_operation_id(self): From 5cbf32778bd9a69c48a6d55801bd5e0a33a6e98b Mon Sep 17 00:00:00 2001 From: Noah Stapp Date: Thu, 30 Jul 2026 16:21:06 -0400 Subject: [PATCH 04/12] Cleanup pass #1 --- pymongo/_telemetry.py | 3 +++ pymongo/asynchronous/helpers.py | 2 +- pymongo/asynchronous/mongo_client.py | 15 +++------------ pymongo/monitoring.py | 6 +++--- pymongo/synchronous/helpers.py | 2 +- pymongo/synchronous/mongo_client.py | 15 +++------------ test/asynchronous/test_monitoring.py | 19 +++++++++++++++++++ test/test_monitoring.py | 19 +++++++++++++++++++ 8 files changed, 52 insertions(+), 29 deletions(-) diff --git a/pymongo/_telemetry.py b/pymongo/_telemetry.py index 176b8e3b89..bbd578a6c0 100644 --- a/pymongo/_telemetry.py +++ b/pymongo/_telemetry.py @@ -257,6 +257,9 @@ def __init__( # 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 diff --git a/pymongo/asynchronous/helpers.py b/pymongo/asynchronous/helpers.py index b7ca91101c..65a6b0230e 100644 --- a/pymongo/asynchronous/helpers.py +++ b/pymongo/asynchronous/helpers.py @@ -109,7 +109,7 @@ def __init__( def backoff(self, attempt: int, base_backoff: Optional[float] = None) -> float: """Return the actual backoff duration for the given attempt and base backoff.""" return _backoff( - max(0, attempt - 1), + max(0, attempt), self.backoff_initial if base_backoff is None or base_backoff < 0 else base_backoff, self.backoff_max, ) diff --git a/pymongo/asynchronous/mongo_client.py b/pymongo/asynchronous/mongo_client.py index a38e14190f..0fbce9c5a2 100644 --- a/pymongo/asynchronous/mongo_client.py +++ b/pymongo/asynchronous/mongo_client.py @@ -35,7 +35,6 @@ import asyncio import contextlib -import logging import os import time as time # noqa: PLC0414 # needed in sync version import warnings @@ -57,7 +56,7 @@ from bson.codec_options import DEFAULT_CODEC_OPTIONS, CodecOptions, TypeRegistry from bson.timestamp import Timestamp from pymongo import _csot, _op_id, common, helpers_shared, periodic_executor -from pymongo._telemetry import log_command_retry +from pymongo._telemetry import _should_generate_op_id, 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 @@ -91,8 +90,6 @@ ) from pymongo.logger import ( _CLIENT_LOGGER, - _COMMAND_LOGGER, - _SERVER_SELECTION_LOGGER, _log_client_error, _log_or_warn, ) @@ -2890,14 +2887,8 @@ def __init__( self._server: Server = None # type: ignore self._deprioritized_servers: Optional[list[Server]] = None self._operation = operation - # Only generate an operation id when APM/logging is enabled - # (_should_generate_op_id inlined: this runs once per operation). - listeners = self._client._event_listeners - if operation_id is None and ( - (listeners is not None and listeners.enabled_for_commands) - or _COMMAND_LOGGER.isEnabledFor(logging.DEBUG) - or _SERVER_SELECTION_LOGGER.isEnabledFor(logging.DEBUG) - ): + # Only generate an operation id when APM/logging is enabled. + if operation_id is None and _should_generate_op_id(self._client._event_listeners): operation_id = _randint() self._operation_id = operation_id self._attempt_number = 0 diff --git a/pymongo/monitoring.py b/pymongo/monitoring.py index e40b4d71ab..252b0daef0 100644 --- a/pymongo/monitoring.py +++ b/pymongo/monitoring.py @@ -516,10 +516,10 @@ def register(listener: _EventListener) -> None: # The "hello" command is also deemed sensitive when attempting speculative # authentication. def _is_speculative_authenticate(command_name: str, doc: Mapping[str, Any]) -> bool: - # Called on every command; probe the dict first, it is cheaper than str.lower(). + # Check the name first, doc may be a RawBSONDocument where `in` decodes the whole document. return bool( - "speculativeAuthenticate" in doc - and command_name.lower() in ("hello", HelloCompat.LEGACY_CMD) + command_name.lower() in ("hello", HelloCompat.LEGACY_CMD) + and "speculativeAuthenticate" in doc ) diff --git a/pymongo/synchronous/helpers.py b/pymongo/synchronous/helpers.py index 94a1371afb..113423f3d6 100644 --- a/pymongo/synchronous/helpers.py +++ b/pymongo/synchronous/helpers.py @@ -109,7 +109,7 @@ def __init__( def backoff(self, attempt: int, base_backoff: Optional[float] = None) -> float: """Return the actual backoff duration for the given attempt and base backoff.""" return _backoff( - max(0, attempt - 1), + max(0, attempt), self.backoff_initial if base_backoff is None or base_backoff < 0 else base_backoff, self.backoff_max, ) diff --git a/pymongo/synchronous/mongo_client.py b/pymongo/synchronous/mongo_client.py index e14c4514ec..ced0cf7b1c 100644 --- a/pymongo/synchronous/mongo_client.py +++ b/pymongo/synchronous/mongo_client.py @@ -35,7 +35,6 @@ import asyncio import contextlib -import logging import os import time as time # noqa: PLC0414 # needed in sync version import warnings @@ -57,7 +56,7 @@ from bson.codec_options import DEFAULT_CODEC_OPTIONS, CodecOptions, TypeRegistry from bson.timestamp import Timestamp from pymongo import _csot, _op_id, common, helpers_shared, periodic_executor -from pymongo._telemetry import log_command_retry +from pymongo._telemetry import _should_generate_op_id, log_command_retry from pymongo.client_options import ClientOptions from pymongo.driver_info import DriverInfo from pymongo.errors import ( @@ -81,8 +80,6 @@ ) from pymongo.logger import ( _CLIENT_LOGGER, - _COMMAND_LOGGER, - _SERVER_SELECTION_LOGGER, _log_client_error, _log_or_warn, ) @@ -2879,14 +2876,8 @@ def __init__( self._server: Server = None # type: ignore self._deprioritized_servers: Optional[list[Server]] = None self._operation = operation - # Only generate an operation id when APM/logging is enabled - # (_should_generate_op_id inlined: this runs once per operation). - listeners = self._client._event_listeners - if operation_id is None and ( - (listeners is not None and listeners.enabled_for_commands) - or _COMMAND_LOGGER.isEnabledFor(logging.DEBUG) - or _SERVER_SELECTION_LOGGER.isEnabledFor(logging.DEBUG) - ): + # Only generate an operation id when APM/logging is enabled. + if operation_id is None and _should_generate_op_id(self._client._event_listeners): operation_id = _randint() self._operation_id = operation_id self._attempt_number = 0 diff --git a/test/asynchronous/test_monitoring.py b/test/asynchronous/test_monitoring.py index 929bd35104..19dc1abcc7 100644 --- a/test/asynchronous/test_monitoring.py +++ b/test/asynchronous/test_monitoring.py @@ -22,8 +22,10 @@ sys.path[0:0] = [""] +from bson import encode from bson.int64 import Int64 from bson.objectid import ObjectId +from bson.raw_bson import RawBSONDocument from bson.son import SON from pymongo import CursorType, DeleteOne, InsertOne, UpdateOne, monitoring from pymongo.asynchronous.command_cursor import AsyncCommandCursor @@ -1195,6 +1197,23 @@ def test_command_event_repr(self): "failure: {'ok': 0}, service_id: None, server_connection_id: None>", ) + def test_succeeded_event_does_not_inflate_raw_reply(self): + # The speculativeAuthenticate redaction check must not decode a lazy + # reply document for non-hello commands. + delta = datetime.timedelta(milliseconds=100) + reply = RawBSONDocument(encode({"ok": 1, "cursor": {"id": Int64(0), "firstBatch": []}})) + event = monitoring.CommandSucceededEvent( + delta, reply, "find", 1, ("localhost", 27017), 2, database_name="test" + ) + self.assertIsNone(reply._RawBSONDocument__inflated_doc) + self.assertIs(event.reply, reply) + # Speculative authentication replies are still redacted. + speculative = RawBSONDocument(encode({"ok": 1, "speculativeAuthenticate": {}})) + event = monitoring.CommandSucceededEvent( + delta, speculative, "hello", 1, ("localhost", 27017), 2, database_name="admin" + ) + self.assertEqual(event.reply, {}) + def test_server_heartbeat_event_repr(self): connection_id = ("localhost", 27017) event = monitoring.ServerHeartbeatStartedEvent(connection_id) diff --git a/test/test_monitoring.py b/test/test_monitoring.py index e8047a6848..cb9417787b 100644 --- a/test/test_monitoring.py +++ b/test/test_monitoring.py @@ -22,8 +22,10 @@ sys.path[0:0] = [""] +from bson import encode from bson.int64 import Int64 from bson.objectid import ObjectId +from bson.raw_bson import RawBSONDocument from bson.son import SON from pymongo import CursorType, DeleteOne, InsertOne, UpdateOne, monitoring from pymongo.errors import AutoReconnect, NotPrimaryError, OperationFailure @@ -1193,6 +1195,23 @@ def test_command_event_repr(self): "failure: {'ok': 0}, service_id: None, server_connection_id: None>", ) + def test_succeeded_event_does_not_inflate_raw_reply(self): + # The speculativeAuthenticate redaction check must not decode a lazy + # reply document for non-hello commands. + delta = datetime.timedelta(milliseconds=100) + reply = RawBSONDocument(encode({"ok": 1, "cursor": {"id": Int64(0), "firstBatch": []}})) + event = monitoring.CommandSucceededEvent( + delta, reply, "find", 1, ("localhost", 27017), 2, database_name="test" + ) + self.assertIsNone(reply._RawBSONDocument__inflated_doc) + self.assertIs(event.reply, reply) + # Speculative authentication replies are still redacted. + speculative = RawBSONDocument(encode({"ok": 1, "speculativeAuthenticate": {}})) + event = monitoring.CommandSucceededEvent( + delta, speculative, "hello", 1, ("localhost", 27017), 2, database_name="admin" + ) + self.assertEqual(event.reply, {}) + def test_server_heartbeat_event_repr(self): connection_id = ("localhost", 27017) event = monitoring.ServerHeartbeatStartedEvent(connection_id) From f608c1a5f044d8c713c93fe9ca8da6d7ca10edee Mon Sep 17 00:00:00 2001 From: Noah Stapp Date: Fri, 31 Jul 2026 10:02:44 -0400 Subject: [PATCH 05/12] Revert changes without significant performance bumps --- pymongo/asynchronous/command_runner.py | 42 +++++++++++++------------- pymongo/asynchronous/helpers.py | 2 +- pymongo/asynchronous/mongo_client.py | 10 ++---- pymongo/synchronous/command_runner.py | 42 +++++++++++++------------- pymongo/synchronous/helpers.py | 2 +- pymongo/synchronous/mongo_client.py | 10 ++---- 6 files changed, 48 insertions(+), 60 deletions(-) diff --git a/pymongo/asynchronous/command_runner.py b/pymongo/asynchronous/command_runner.py index 78d2ad3525..2f0d2a1e76 100644 --- a/pymongo/asynchronous/command_runner.py +++ b/pymongo/asynchronous/command_runner.py @@ -80,6 +80,7 @@ async def _run_command( dbname: str, request_id: int, msg: bytes, + *, client: Optional[AsyncMongoClient[Any]], session: Optional[AsyncClientSession], listeners: Optional[_EventListeners], @@ -87,20 +88,19 @@ async def _run_command( codec_options: CodecOptions[_DocumentType], user_fields: Optional[Mapping[str, Any]] = None, orig: Optional[MutableMapping[str, Any]] = None, + op_id: Optional[int] = None, + command_name: Optional[str] = None, check: bool = True, allowable_errors: Optional[Sequence[Union[str, int]]] = None, parse_write_concern_error: bool = False, - speculative_hello: bool = False, - unacknowledged: bool = False, - set_conn_more_to_come: bool = False, - *, - op_id: Optional[int] = None, - command_name: Optional[str] = None, pool_opts: Optional[PoolOptions] = None, + unacknowledged: bool = False, + speculative_hello: bool = False, ensure_db: bool = False, decrypt_reply: bool = True, max_doc_size: int = 0, more_to_come: bool = False, + 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], float]: @@ -274,11 +274,11 @@ async def run_bulk_write_command( bwc.db_name, request_id, msg, - client, - bwc.session, # type: ignore[arg-type] - bwc.listeners, - topology_id, - bwc.codec, + client=client, + session=bwc.session, # type: ignore[arg-type] + listeners=bwc.listeners, + topology_id=topology_id, + codec_options=bwc.codec, op_id=bwc.op_id, command_name=bwc.name, orig=orig, @@ -334,11 +334,11 @@ async def run_cursor_command( dbname, request_id, msg, - client, - session, - listeners, - topology_id, - codec_options, + client=client, + session=session, + listeners=listeners, + topology_id=topology_id, + codec_options=codec_options, user_fields=user_fields, command_name=command_name, pool_opts=pool_opts, @@ -445,11 +445,11 @@ async def run_command( dbname, request_id, msg, - client, - session, - listeners, - topology_id, - codec_options, + client=client, + session=session, + listeners=listeners, + topology_id=topology_id, + codec_options=codec_options, user_fields=user_fields, orig=orig, check=check, diff --git a/pymongo/asynchronous/helpers.py b/pymongo/asynchronous/helpers.py index 65a6b0230e..a6c35b2f05 100644 --- a/pymongo/asynchronous/helpers.py +++ b/pymongo/asynchronous/helpers.py @@ -90,7 +90,7 @@ def _backoff( max_delay: float = _BACKOFF_MAX, ) -> float: jitter = random.random() # noqa: S311 - return jitter * min(base_backoff * (2 ** (attempt)), max_delay) + return jitter * min(base_backoff * (2**attempt), max_delay) class _RetryPolicy: diff --git a/pymongo/asynchronous/mongo_client.py b/pymongo/asynchronous/mongo_client.py index 0fbce9c5a2..b5b7513676 100644 --- a/pymongo/asynchronous/mongo_client.py +++ b/pymongo/asynchronous/mongo_client.py @@ -3132,11 +3132,8 @@ async def _write(self) -> T: # One operation id across all attempts of this operation if APM/logging is enabled if self._operation_id is None: return await self._func(self._session, conn, self._retryable) # type: ignore - token = _op_id.OP_ID.set(self._operation_id) - try: + with _op_id._OpIdContext(self._operation_id): return await self._func(self._session, conn, self._retryable) # type: ignore - finally: - _op_id.OP_ID.reset(token) except PyMongoError as exc: if not self._retryable: raise @@ -3162,11 +3159,8 @@ async def _read(self) -> T: # One operation id across all attempts of this operation if APM/logging is enabled if self._operation_id is None: return await self._func(self._session, self._server, conn, read_pref) # type: ignore - token = _op_id.OP_ID.set(self._operation_id) - try: + with _op_id._OpIdContext(self._operation_id): return await self._func(self._session, self._server, conn, read_pref) # type: ignore - finally: - _op_id.OP_ID.reset(token) def _after_fork_child() -> None: diff --git a/pymongo/synchronous/command_runner.py b/pymongo/synchronous/command_runner.py index 08f8e5afe8..3a44ab2183 100644 --- a/pymongo/synchronous/command_runner.py +++ b/pymongo/synchronous/command_runner.py @@ -80,6 +80,7 @@ def _run_command( dbname: str, request_id: int, msg: bytes, + *, client: Optional[MongoClient[Any]], session: Optional[ClientSession], listeners: Optional[_EventListeners], @@ -87,20 +88,19 @@ def _run_command( codec_options: CodecOptions[_DocumentType], user_fields: Optional[Mapping[str, Any]] = None, orig: Optional[MutableMapping[str, Any]] = None, + op_id: Optional[int] = None, + command_name: Optional[str] = None, check: bool = True, allowable_errors: Optional[Sequence[Union[str, int]]] = None, parse_write_concern_error: bool = False, - speculative_hello: bool = False, - unacknowledged: bool = False, - set_conn_more_to_come: bool = False, - *, - op_id: Optional[int] = None, - command_name: Optional[str] = None, pool_opts: Optional[PoolOptions] = None, + unacknowledged: bool = False, + speculative_hello: bool = False, ensure_db: bool = False, decrypt_reply: bool = True, max_doc_size: int = 0, more_to_come: bool = False, + 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], float]: @@ -274,11 +274,11 @@ def run_bulk_write_command( bwc.db_name, request_id, msg, - client, - bwc.session, # type: ignore[arg-type] - bwc.listeners, - topology_id, - bwc.codec, + client=client, + session=bwc.session, # type: ignore[arg-type] + listeners=bwc.listeners, + topology_id=topology_id, + codec_options=bwc.codec, op_id=bwc.op_id, command_name=bwc.name, orig=orig, @@ -334,11 +334,11 @@ def run_cursor_command( dbname, request_id, msg, - client, - session, - listeners, - topology_id, - codec_options, + client=client, + session=session, + listeners=listeners, + topology_id=topology_id, + codec_options=codec_options, user_fields=user_fields, command_name=command_name, pool_opts=pool_opts, @@ -445,11 +445,11 @@ def run_command( dbname, request_id, msg, - client, - session, - listeners, - topology_id, - codec_options, + client=client, + session=session, + listeners=listeners, + topology_id=topology_id, + codec_options=codec_options, user_fields=user_fields, orig=orig, check=check, diff --git a/pymongo/synchronous/helpers.py b/pymongo/synchronous/helpers.py index 113423f3d6..3f0d540c57 100644 --- a/pymongo/synchronous/helpers.py +++ b/pymongo/synchronous/helpers.py @@ -90,7 +90,7 @@ def _backoff( max_delay: float = _BACKOFF_MAX, ) -> float: jitter = random.random() # noqa: S311 - return jitter * min(base_backoff * (2 ** (attempt)), max_delay) + return jitter * min(base_backoff * (2**attempt), max_delay) class _RetryPolicy: diff --git a/pymongo/synchronous/mongo_client.py b/pymongo/synchronous/mongo_client.py index ced0cf7b1c..c3058ec80e 100644 --- a/pymongo/synchronous/mongo_client.py +++ b/pymongo/synchronous/mongo_client.py @@ -3121,11 +3121,8 @@ def _write(self) -> T: # One operation id across all attempts of this operation if APM/logging is enabled if self._operation_id is None: return self._func(self._session, conn, self._retryable) # type: ignore - token = _op_id.OP_ID.set(self._operation_id) - try: + with _op_id._OpIdContext(self._operation_id): return self._func(self._session, conn, self._retryable) # type: ignore - finally: - _op_id.OP_ID.reset(token) except PyMongoError as exc: if not self._retryable: raise @@ -3151,11 +3148,8 @@ def _read(self) -> T: # One operation id across all attempts of this operation if APM/logging is enabled if self._operation_id is None: return self._func(self._session, self._server, conn, read_pref) # type: ignore - token = _op_id.OP_ID.set(self._operation_id) - try: + with _op_id._OpIdContext(self._operation_id): return self._func(self._session, self._server, conn, read_pref) # type: ignore - finally: - _op_id.OP_ID.reset(token) def _after_fork_child() -> None: From 6d58a7d65126a0886807032ebf01c196b412fae8 Mon Sep 17 00:00:00 2001 From: Noah Stapp Date: Tue, 4 Aug 2026 09:38:04 -0400 Subject: [PATCH 06/12] Comment cleanup --- pymongo/_telemetry.py | 30 ++++++++++---------------- pymongo/asynchronous/command_runner.py | 4 +--- pymongo/asynchronous/pool.py | 3 +-- pymongo/asynchronous/topology.py | 3 +-- pymongo/synchronous/command_runner.py | 4 +--- pymongo/synchronous/pool.py | 3 +-- pymongo/synchronous/topology.py | 3 +-- 7 files changed, 17 insertions(+), 33 deletions(-) diff --git a/pymongo/_telemetry.py b/pymongo/_telemetry.py index bbd578a6c0..057a0446eb 100644 --- a/pymongo/_telemetry.py +++ b/pymongo/_telemetry.py @@ -57,9 +57,7 @@ def _monotonic_duration(start: float) -> float: 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 True if an operation id would be consumed by APM events or logging.""" return ( (listeners is not None and listeners.enabled_for_commands) or _COMMAND_LOGGER.isEnabledFor(logging.DEBUG) @@ -70,14 +68,13 @@ def _should_generate_op_id(listeners: Optional[_EventListeners]) -> bool: class _CommandTelemetry: """Combines structured logging and APM event publishing for a single command. - Construct once per command, call :meth:`started` before the network send, + Construct up to 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. + This sits on the hot path of every command: when both APM events and command + logging are disabled, only the gate flags and the monotonic duration clock + are maintained. """ __slots__ = ( @@ -529,7 +526,7 @@ class _SdamTelemetry: Topology events are queued for asynchronous delivery; log entries are emitted inline. """ - __slots__ = ("_events", "_listeners", "_topology_id") + __slots__ = ("_events", "_listeners", "_publish_server", "_publish_tp", "_topology_id") def __init__( self, @@ -540,16 +537,11 @@ def __init__( 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 + # The SDAM listener set is fixed once the client is constructed + # (_EventListeners copies the global listeners at __init__), so these + # gates are static for the life of the client. + 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 @property def _should_log(self) -> bool: diff --git a/pymongo/asynchronous/command_runner.py b/pymongo/asynchronous/command_runner.py index 2f0d2a1e76..6ead91e10c 100644 --- a/pymongo/asynchronous/command_runner.py +++ b/pymongo/asynchronous/command_runner.py @@ -164,9 +164,7 @@ async def _run_command( if orig is None: orig = cmd - # 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. + # Fast path: skip telemetry construction when logging and APM are disabled 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 diff --git a/pymongo/asynchronous/pool.py b/pymongo/asynchronous/pool.py index 37fd6edf1d..38c34922bf 100644 --- a/pymongo/asynchronous/pool.py +++ b/pymongo/asynchronous/pool.py @@ -1237,8 +1237,7 @@ def __init__( async def __aenter__(self) -> AsyncConnection: pool = self._pool telemetry = pool._telemetry - # Fast path: when neither CMAP listeners nor connection logging are - # active, skip the checkout started/succeeded telemetry calls. + # Fast path: skip telemetry calls when CMAP events/logging are disabled if not telemetry._publish and not ( telemetry._log and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG) ): diff --git a/pymongo/asynchronous/topology.py b/pymongo/asynchronous/topology.py index d00f2ea4ed..e6aae66c3c 100644 --- a/pymongo/asynchronous/topology.py +++ b/pymongo/asynchronous/topology.py @@ -285,8 +285,7 @@ async def _select_servers_loop( now = time.monotonic() end_time = now + timeout logged_waiting = False - # Server selection defines only log entries (no APM events); skip the - # telemetry object entirely when the logger is not enabled. + # Server selection does not have APM events, gate only on logging ss: Optional[_ServerSelectionTelemetry] = None if _SERVER_SELECTION_LOGGER.isEnabledFor(logging.DEBUG): ss = _ServerSelectionTelemetry( diff --git a/pymongo/synchronous/command_runner.py b/pymongo/synchronous/command_runner.py index 3a44ab2183..3c6588e19f 100644 --- a/pymongo/synchronous/command_runner.py +++ b/pymongo/synchronous/command_runner.py @@ -164,9 +164,7 @@ def _run_command( if orig is None: orig = cmd - # 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. + # Fast path: skip telemetry construction when logging and APM are disabled 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 diff --git a/pymongo/synchronous/pool.py b/pymongo/synchronous/pool.py index 54a31e43d2..a75eb6ac90 100644 --- a/pymongo/synchronous/pool.py +++ b/pymongo/synchronous/pool.py @@ -1233,8 +1233,7 @@ def __init__( def __enter__(self) -> Connection: pool = self._pool telemetry = pool._telemetry - # Fast path: when neither CMAP listeners nor connection logging are - # active, skip the checkout started/succeeded telemetry calls. + # Fast path: skip telemetry calls when CMAP events/logging are disabled if not telemetry._publish and not ( telemetry._log and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG) ): diff --git a/pymongo/synchronous/topology.py b/pymongo/synchronous/topology.py index ab6bb05cb6..7e1e6c09d2 100644 --- a/pymongo/synchronous/topology.py +++ b/pymongo/synchronous/topology.py @@ -285,8 +285,7 @@ def _select_servers_loop( now = time.monotonic() end_time = now + timeout logged_waiting = False - # Server selection defines only log entries (no APM events); skip the - # telemetry object entirely when the logger is not enabled. + # Server selection does not have APM events, gate only on logging ss: Optional[_ServerSelectionTelemetry] = None if _SERVER_SELECTION_LOGGER.isEnabledFor(logging.DEBUG): ss = _ServerSelectionTelemetry( From 345fd715afa69c1dc97e4025d30484d9f4f214f7 Mon Sep 17 00:00:00 2001 From: Noah Stapp Date: Tue, 4 Aug 2026 12:03:55 -0400 Subject: [PATCH 07/12] More clarity cleanup --- pymongo/_telemetry.py | 7 +- pymongo/asynchronous/command_runner.py | 1 + pymongo/synchronous/command_runner.py | 1 + test/asynchronous/test_operation_id_retry.py | 71 +++++++++++++++++++- test/test_operation_id_retry.py | 69 ++++++++++++++++++- 5 files changed, 142 insertions(+), 7 deletions(-) diff --git a/pymongo/_telemetry.py b/pymongo/_telemetry.py index 057a0446eb..b77fd20928 100644 --- a/pymongo/_telemetry.py +++ b/pymongo/_telemetry.py @@ -104,6 +104,8 @@ def __init__( op_id: Optional[int], name: Optional[str] = None, ) -> None: + # NOTE: the _run_command fast path in command_runner.py inline this gate for performance + # They must be kept in sync with any gating changes 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 @@ -254,9 +256,8 @@ def __init__( # 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. + # NOTE: the checkout/checkin fast paths in pool.py inline this gate for performance + # They must be kept in sync with any gating changes self._publish = publish and listeners is not None and listeners.enabled_for_cmap self._log = log diff --git a/pymongo/asynchronous/command_runner.py b/pymongo/asynchronous/command_runner.py index 6ead91e10c..e8defd50f1 100644 --- a/pymongo/asynchronous/command_runner.py +++ b/pymongo/asynchronous/command_runner.py @@ -165,6 +165,7 @@ async def _run_command( orig = cmd # Fast path: skip telemetry construction when logging and APM are disabled + # Inline enabled check here for performance 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 diff --git a/pymongo/synchronous/command_runner.py b/pymongo/synchronous/command_runner.py index 3c6588e19f..0f37cf9440 100644 --- a/pymongo/synchronous/command_runner.py +++ b/pymongo/synchronous/command_runner.py @@ -165,6 +165,7 @@ def _run_command( orig = cmd # Fast path: skip telemetry construction when logging and APM are disabled + # Inline enabled check here for performance 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 diff --git a/test/asynchronous/test_operation_id_retry.py b/test/asynchronous/test_operation_id_retry.py index ce1c82207b..581ecb4009 100644 --- a/test/asynchronous/test_operation_id_retry.py +++ b/test/asynchronous/test_operation_id_retry.py @@ -26,14 +26,16 @@ from bson.codec_options import DEFAULT_CODEC_OPTIONS from pymongo import _op_id from pymongo._telemetry import _CommandTelemetry -from pymongo.asynchronous import mongo_client +from pymongo.asynchronous import bulk, client_bulk, mongo_client from pymongo.asynchronous.encryption import _Encrypter from pymongo.asynchronous.helpers import _handle_reauth from pymongo.asynchronous.pool import AsyncConnection from pymongo.errors import OperationFailure from pymongo.helpers_shared import _REAUTHENTICATION_REQUIRED_CODE from pymongo.logger import _COMMAND_LOGGER, _SERVER_SELECTION_LOGGER +from pymongo.message import _randint from pymongo.operations import InsertOne +from pymongo.write_concern import WriteConcern from test.asynchronous import AsyncIntegrationTest, async_client_context, unittest from test.utils_shared import AllowListEventListener @@ -139,7 +141,7 @@ async def test_retryable_reads_reuse_operation_id(self): with self.subTest(command=name, index=i): await self._check_stable_operation_id(name, f, self.RETRIES) - async def test_retry_without_listeners_or_logging_creates_no_operation_id(self): + async def test_retry_without_telemetry_creates_no_operation_id(self): appname = _APP_NAME + "noapm" client = await self.async_rs_or_single_client(appname=appname) @@ -186,6 +188,71 @@ def recording_init( "expected no _CommandTelemetry construction without APM/logging enabled", ) + async def test_bulk_write_without_telemetry_creates_no_operation_id(self): + client = await self.async_rs_or_single_client() + + # Make sure APM and logging are disabled + for logger in (_COMMAND_LOGGER, _SERVER_SELECTION_LOGGER): + self.assertFalse(logger.isEnabledFor(logging.DEBUG)) + self.assertFalse(client._event_listeners.enabled_for_commands) + + coll = client.pymongo_test.test_operation_id_retry + coll_w0 = coll.with_options(write_concern=WriteConcern(w=0)) + with ( + patch.object(bulk, "_randint") as bulk_randint, + patch.object(mongo_client, "_randint") as client_randint, + ): + # Acknowledged + result = await coll.bulk_write([InsertOne({})]) + self.assertEqual(result.inserted_count, 1) + # Unacknowledged ordered + self.assertFalse((await coll_w0.bulk_write([InsertOne({})])).acknowledged) + # Unacknowledged unordered + self.assertFalse( + (await coll_w0.bulk_write([InsertOne({})], ordered=False)).acknowledged + ) + self.assertEqual( + bulk_randint.call_count, 0, "generated an operation id without APM/logging enabled" + ) + self.assertEqual( + client_randint.call_count, 0, "generated an operation id without APM/logging enabled" + ) + + # Ensure we see randint() calls with APM enabled + with patch.object(bulk, "_randint", wraps=_randint) as bulk_randint: + await self.coll.bulk_write([InsertOne({})]) + self.assertEqual(bulk_randint.call_count, 1) + + @async_client_context.require_version_min(8, 0, 0, -24) + async def test_client_bulk_write_without_telemetry_creates_no_operation_id(self): + client = await self.async_rs_or_single_client() + + # Make sure APM and logging are disabled + for logger in (_COMMAND_LOGGER, _SERVER_SELECTION_LOGGER): + self.assertFalse(logger.isEnabledFor(logging.DEBUG)) + self.assertFalse(client._event_listeners.enabled_for_commands) + + ns = "pymongo_test.test_operation_id_retry" + with patch.object(client_bulk, "_randint") as bulk_randint: + # Acknowledged + result = await client.bulk_write([InsertOne(namespace=ns, document={})]) + self.assertEqual(result.inserted_count, 1) + # Unacknowledged + result = await client.bulk_write( + [InsertOne(namespace=ns, document={})], + write_concern=WriteConcern(w=0), + ordered=False, + ) + self.assertFalse(result.acknowledged) + self.assertEqual( + bulk_randint.call_count, 0, "generated an operation id without APM/logging enabled" + ) + + # Ensure we see randint() calls with APM enabled + with patch.object(client_bulk, "_randint", wraps=_randint) as bulk_randint: + await self.client.bulk_write([InsertOne(namespace=ns, document={})]) + self.assertEqual(bulk_randint.call_count, 1) + async def test_reauth_does_not_reuse_operation_id(self): class FakeConnection(AsyncConnection): def __init__(self): diff --git a/test/test_operation_id_retry.py b/test/test_operation_id_retry.py index 1253aaee7f..e1cb699e54 100644 --- a/test/test_operation_id_retry.py +++ b/test/test_operation_id_retry.py @@ -29,11 +29,13 @@ from pymongo.errors import OperationFailure from pymongo.helpers_shared import _REAUTHENTICATION_REQUIRED_CODE from pymongo.logger import _COMMAND_LOGGER, _SERVER_SELECTION_LOGGER +from pymongo.message import _randint from pymongo.operations import InsertOne -from pymongo.synchronous import mongo_client +from pymongo.synchronous import bulk, client_bulk, mongo_client from pymongo.synchronous.encryption import _Encrypter from pymongo.synchronous.helpers import _handle_reauth from pymongo.synchronous.pool import Connection +from pymongo.write_concern import WriteConcern from test import IntegrationTest, client_context, unittest from test.utils_shared import AllowListEventListener @@ -137,7 +139,7 @@ def test_retryable_reads_reuse_operation_id(self): with self.subTest(command=name, index=i): self._check_stable_operation_id(name, f, self.RETRIES) - def test_retry_without_listeners_or_logging_creates_no_operation_id(self): + def test_retry_without_telemetry_creates_no_operation_id(self): appname = _APP_NAME + "noapm" client = self.rs_or_single_client(appname=appname) @@ -184,6 +186,69 @@ def recording_init( "expected no _CommandTelemetry construction without APM/logging enabled", ) + def test_bulk_write_without_telemetry_creates_no_operation_id(self): + client = self.rs_or_single_client() + + # Make sure APM and logging are disabled + for logger in (_COMMAND_LOGGER, _SERVER_SELECTION_LOGGER): + self.assertFalse(logger.isEnabledFor(logging.DEBUG)) + self.assertFalse(client._event_listeners.enabled_for_commands) + + coll = client.pymongo_test.test_operation_id_retry + coll_w0 = coll.with_options(write_concern=WriteConcern(w=0)) + with ( + patch.object(bulk, "_randint") as bulk_randint, + patch.object(mongo_client, "_randint") as client_randint, + ): + # Acknowledged + result = coll.bulk_write([InsertOne({})]) + self.assertEqual(result.inserted_count, 1) + # Unacknowledged ordered + self.assertFalse((coll_w0.bulk_write([InsertOne({})])).acknowledged) + # Unacknowledged unordered + self.assertFalse((coll_w0.bulk_write([InsertOne({})], ordered=False)).acknowledged) + self.assertEqual( + bulk_randint.call_count, 0, "generated an operation id without APM/logging enabled" + ) + self.assertEqual( + client_randint.call_count, 0, "generated an operation id without APM/logging enabled" + ) + + # Ensure we see randint() calls with APM enabled + with patch.object(bulk, "_randint", wraps=_randint) as bulk_randint: + self.coll.bulk_write([InsertOne({})]) + self.assertEqual(bulk_randint.call_count, 1) + + @client_context.require_version_min(8, 0, 0, -24) + def test_client_bulk_write_without_telemetry_creates_no_operation_id(self): + client = self.rs_or_single_client() + + # Make sure APM and logging are disabled + for logger in (_COMMAND_LOGGER, _SERVER_SELECTION_LOGGER): + self.assertFalse(logger.isEnabledFor(logging.DEBUG)) + self.assertFalse(client._event_listeners.enabled_for_commands) + + ns = "pymongo_test.test_operation_id_retry" + with patch.object(client_bulk, "_randint") as bulk_randint: + # Acknowledged + result = client.bulk_write([InsertOne(namespace=ns, document={})]) + self.assertEqual(result.inserted_count, 1) + # Unacknowledged + result = client.bulk_write( + [InsertOne(namespace=ns, document={})], + write_concern=WriteConcern(w=0), + ordered=False, + ) + self.assertFalse(result.acknowledged) + self.assertEqual( + bulk_randint.call_count, 0, "generated an operation id without APM/logging enabled" + ) + + # Ensure we see randint() calls with APM enabled + with patch.object(client_bulk, "_randint", wraps=_randint) as bulk_randint: + self.client.bulk_write([InsertOne(namespace=ns, document={})]) + self.assertEqual(bulk_randint.call_count, 1) + def test_reauth_does_not_reuse_operation_id(self): class FakeConnection(Connection): def __init__(self): From d61df7a03ef9aac145e6ba0888109883a40fb628 Mon Sep 17 00:00:00 2001 From: Noah Stapp Date: Tue, 4 Aug 2026 12:31:30 -0400 Subject: [PATCH 08/12] Remove unneeded properties --- pymongo/_telemetry.py | 8 ++------ pymongo/asynchronous/pool.py | 4 ++-- pymongo/synchronous/pool.py | 4 ++-- 3 files changed, 6 insertions(+), 10 deletions(-) diff --git a/pymongo/_telemetry.py b/pymongo/_telemetry.py index b77fd20928..9917b06469 100644 --- a/pymongo/_telemetry.py +++ b/pymongo/_telemetry.py @@ -239,7 +239,7 @@ class _CmapTelemetry: "_client_id", "_listeners", "_log", - "_publish", + "_should_publish", ) def __init__( @@ -258,13 +258,9 @@ def __init__( # gate is static for the life of the pool. # NOTE: the checkout/checkin fast paths in pool.py inline this gate for performance # They must be kept in sync with any gating changes - self._publish = publish and listeners is not None and listeners.enabled_for_cmap + self._should_publish = publish and listeners is not None and listeners.enabled_for_cmap self._log = log - @property - def _should_publish(self) -> bool: - return self._publish - @property def _should_log(self) -> bool: """Computed per-call because logging level can be reconfigured at runtime.""" diff --git a/pymongo/asynchronous/pool.py b/pymongo/asynchronous/pool.py index 38c34922bf..a635010b05 100644 --- a/pymongo/asynchronous/pool.py +++ b/pymongo/asynchronous/pool.py @@ -1116,7 +1116,7 @@ async def checkin(self, conn: AsyncConnection) -> None: async with self.lock: self.active_contexts.discard(conn.cancel_context) telemetry = self._telemetry - if telemetry._publish or ( + if telemetry._should_publish or ( telemetry._log and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG) ): telemetry.checked_in(conn.id) @@ -1238,7 +1238,7 @@ async def __aenter__(self) -> AsyncConnection: pool = self._pool telemetry = pool._telemetry # Fast path: skip telemetry calls when CMAP events/logging are disabled - if not telemetry._publish and not ( + if not telemetry._should_publish and not ( telemetry._log and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG) ): conn = await pool._get_conn(time.monotonic(), handler=self._handler) diff --git a/pymongo/synchronous/pool.py b/pymongo/synchronous/pool.py index a75eb6ac90..7d1b8443ee 100644 --- a/pymongo/synchronous/pool.py +++ b/pymongo/synchronous/pool.py @@ -1112,7 +1112,7 @@ def checkin(self, conn: Connection) -> None: with self.lock: self.active_contexts.discard(conn.cancel_context) telemetry = self._telemetry - if telemetry._publish or ( + if telemetry._should_publish or ( telemetry._log and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG) ): telemetry.checked_in(conn.id) @@ -1234,7 +1234,7 @@ def __enter__(self) -> Connection: pool = self._pool telemetry = pool._telemetry # Fast path: skip telemetry calls when CMAP events/logging are disabled - if not telemetry._publish and not ( + if not telemetry._should_publish and not ( telemetry._log and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG) ): conn = pool._get_conn(time.monotonic(), handler=self._handler) From c47ddfaefde735adefb49b000e50c78b09aed490 Mon Sep 17 00:00:00 2001 From: Noah Stapp Date: Wed, 5 Aug 2026 10:50:42 -0400 Subject: [PATCH 09/12] Use _is_debug_enabled instead of isEnabledFor --- pymongo/_telemetry.py | 22 +++++++++++----------- pymongo/asynchronous/pool.py | 9 +++------ pymongo/asynchronous/topology.py | 4 ++-- pymongo/logger.py | 4 ++++ pymongo/synchronous/pool.py | 9 +++------ pymongo/synchronous/topology.py | 4 ++-- 6 files changed, 25 insertions(+), 27 deletions(-) diff --git a/pymongo/_telemetry.py b/pymongo/_telemetry.py index 9917b06469..efccf162ac 100644 --- a/pymongo/_telemetry.py +++ b/pymongo/_telemetry.py @@ -17,7 +17,6 @@ from __future__ import annotations import datetime -import logging import queue import time from collections.abc import MutableMapping @@ -32,6 +31,7 @@ _CommandStatusMessage, _ConnectionStatusMessage, _debug_log, + _is_debug_enabled, _SDAMStatusMessage, _ServerSelectionStatusMessage, _verbose_connection_error_reason, @@ -60,8 +60,8 @@ def _should_generate_op_id(listeners: Optional[_EventListeners]) -> bool: """Return True if an operation id would be consumed by APM events or logging.""" return ( (listeners is not None and listeners.enabled_for_commands) - or _COMMAND_LOGGER.isEnabledFor(logging.DEBUG) - or _SERVER_SELECTION_LOGGER.isEnabledFor(logging.DEBUG) + or _is_debug_enabled(_COMMAND_LOGGER) + or _is_debug_enabled(_SERVER_SELECTION_LOGGER) ) @@ -106,7 +106,7 @@ def __init__( ) -> None: # NOTE: the _run_command fast path in command_runner.py inline this gate for performance # They must be kept in sync with any gating changes - self._should_log = topology_id is not None and _COMMAND_LOGGER.isEnabledFor(logging.DEBUG) + self._should_log = topology_id is not None and _is_debug_enabled(_COMMAND_LOGGER) self._publish = listeners is not None and listeners.enabled_for_commands self._active = self._should_log or self._publish self._start = 0.0 @@ -264,7 +264,7 @@ def __init__( @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) + return self._log and _is_debug_enabled(_CONNECTION_LOGGER) def _emit_log(self, message: _ConnectionStatusMessage, **extra: Any) -> None: _debug_log( @@ -442,7 +442,7 @@ def __init__( # 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._should_log = _is_debug_enabled(_SDAM_LOGGER) self._start: float = 0.0 def _emit_log(self, message: _SDAMStatusMessage, awaited: bool, **extra: Any) -> None: @@ -543,7 +543,7 @@ def __init__( @property def _should_log(self) -> bool: """Computed per-call because logging level can be reconfigured at runtime.""" - return _SDAM_LOGGER.isEnabledFor(logging.DEBUG) + return _is_debug_enabled(_SDAM_LOGGER) def _enqueue(self, fn: Any, args: tuple[Any, ...]) -> None: if self._events is not None: @@ -668,7 +668,7 @@ def __init__( 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) + self._should_log = _is_debug_enabled(_SERVER_SELECTION_LOGGER) def _emit_log( self, @@ -721,7 +721,7 @@ def log_server_selection_succeeded( server_port: Optional[int], ) -> None: """Emit the server selection SUCCEEDED log entry.""" - if _SERVER_SELECTION_LOGGER.isEnabledFor(logging.DEBUG): + if _is_debug_enabled(_SERVER_SELECTION_LOGGER): _debug_log( _SERVER_SELECTION_LOGGER, message=_ServerSelectionStatusMessage.SUCCEEDED, @@ -737,7 +737,7 @@ def log_server_selection_succeeded( 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): + if _is_debug_enabled(_SDAM_LOGGER): _debug_log(_SDAM_LOGGER, message="SRV monitor check failed", failure=repr(failure)) @@ -749,7 +749,7 @@ def log_command_retry( is_write: bool, ) -> None: """Emit a command-retry log entry.""" - if _COMMAND_LOGGER.isEnabledFor(logging.DEBUG): + if _is_debug_enabled(_COMMAND_LOGGER): op = "write" if is_write else "read" _debug_log( _COMMAND_LOGGER, diff --git a/pymongo/asynchronous/pool.py b/pymongo/asynchronous/pool.py index a635010b05..a126962e4a 100644 --- a/pymongo/asynchronous/pool.py +++ b/pymongo/asynchronous/pool.py @@ -16,7 +16,6 @@ import asyncio import collections -import logging import os import socket import time @@ -63,7 +62,7 @@ _async_create_condition, _async_create_lock, ) -from pymongo.logger import _CONNECTION_LOGGER +from pymongo.logger import _CONNECTION_LOGGER, _is_debug_enabled from pymongo.monitoring import ( ConnectionCheckOutFailedReason, ConnectionClosedReason, @@ -1116,9 +1115,7 @@ async def checkin(self, conn: AsyncConnection) -> None: async with self.lock: self.active_contexts.discard(conn.cancel_context) telemetry = self._telemetry - if telemetry._should_publish or ( - telemetry._log and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG) - ): + if telemetry._should_publish or (telemetry._log and _is_debug_enabled(_CONNECTION_LOGGER)): telemetry.checked_in(conn.id) if self.pid != os.getpid(): await self.reset_without_pause() @@ -1239,7 +1236,7 @@ async def __aenter__(self) -> AsyncConnection: telemetry = pool._telemetry # Fast path: skip telemetry calls when CMAP events/logging are disabled if not telemetry._should_publish and not ( - telemetry._log and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG) + telemetry._log and _is_debug_enabled(_CONNECTION_LOGGER) ): conn = await pool._get_conn(time.monotonic(), handler=self._handler) self._conn = conn diff --git a/pymongo/asynchronous/topology.py b/pymongo/asynchronous/topology.py index e6aae66c3c..cb1d348df0 100644 --- a/pymongo/asynchronous/topology.py +++ b/pymongo/asynchronous/topology.py @@ -56,7 +56,7 @@ _async_create_condition, _async_create_lock, ) -from pymongo.logger import _SERVER_SELECTION_LOGGER +from pymongo.logger import _SERVER_SELECTION_LOGGER, _is_debug_enabled from pymongo.pool_options import PoolOptions from pymongo.server_description import ServerDescription from pymongo.server_selectors import ( @@ -287,7 +287,7 @@ async def _select_servers_loop( logged_waiting = False # Server selection does not have APM events, gate only on logging ss: Optional[_ServerSelectionTelemetry] = None - if _SERVER_SELECTION_LOGGER.isEnabledFor(logging.DEBUG): + if _is_debug_enabled(_SERVER_SELECTION_LOGGER): ss = _ServerSelectionTelemetry( self._topology_id, selector, operation, operation_id, self.description ) diff --git a/pymongo/logger.py b/pymongo/logger.py index 0441f05fc3..43256d18a5 100644 --- a/pymongo/logger.py +++ b/pymongo/logger.py @@ -96,6 +96,10 @@ class _SDAMStatusMessage(str, enum.Enum): } +def _is_debug_enabled(logger: logging.Logger) -> bool: + return logger.isEnabledFor(logging.DEBUG) + + def _log_client_error() -> None: # This is called from a daemon thread so check for None to account for interpreter shutdown. logger = _CLIENT_LOGGER diff --git a/pymongo/synchronous/pool.py b/pymongo/synchronous/pool.py index 7d1b8443ee..b75415eeeb 100644 --- a/pymongo/synchronous/pool.py +++ b/pymongo/synchronous/pool.py @@ -16,7 +16,6 @@ import asyncio import collections -import logging import os import socket import time @@ -60,7 +59,7 @@ _create_condition, _create_lock, ) -from pymongo.logger import _CONNECTION_LOGGER +from pymongo.logger import _CONNECTION_LOGGER, _is_debug_enabled from pymongo.monitoring import ( ConnectionCheckOutFailedReason, ConnectionClosedReason, @@ -1112,9 +1111,7 @@ def checkin(self, conn: Connection) -> None: with self.lock: self.active_contexts.discard(conn.cancel_context) telemetry = self._telemetry - if telemetry._should_publish or ( - telemetry._log and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG) - ): + if telemetry._should_publish or (telemetry._log and _is_debug_enabled(_CONNECTION_LOGGER)): telemetry.checked_in(conn.id) if self.pid != os.getpid(): self.reset_without_pause() @@ -1235,7 +1232,7 @@ def __enter__(self) -> Connection: telemetry = pool._telemetry # Fast path: skip telemetry calls when CMAP events/logging are disabled if not telemetry._should_publish and not ( - telemetry._log and _CONNECTION_LOGGER.isEnabledFor(logging.DEBUG) + telemetry._log and _is_debug_enabled(_CONNECTION_LOGGER) ): conn = pool._get_conn(time.monotonic(), handler=self._handler) self._conn = conn diff --git a/pymongo/synchronous/topology.py b/pymongo/synchronous/topology.py index 7e1e6c09d2..75278fa942 100644 --- a/pymongo/synchronous/topology.py +++ b/pymongo/synchronous/topology.py @@ -52,7 +52,7 @@ _create_condition, _create_lock, ) -from pymongo.logger import _SERVER_SELECTION_LOGGER +from pymongo.logger import _SERVER_SELECTION_LOGGER, _is_debug_enabled from pymongo.pool_options import PoolOptions from pymongo.server_description import ServerDescription from pymongo.server_selectors import ( @@ -287,7 +287,7 @@ def _select_servers_loop( logged_waiting = False # Server selection does not have APM events, gate only on logging ss: Optional[_ServerSelectionTelemetry] = None - if _SERVER_SELECTION_LOGGER.isEnabledFor(logging.DEBUG): + if _is_debug_enabled(_SERVER_SELECTION_LOGGER): ss = _ServerSelectionTelemetry( self._topology_id, selector, operation, operation_id, self.description ) From 62928d61523fe2a98d77332b5acdae293936b4be Mon Sep 17 00:00:00 2001 From: Noah Stapp Date: Wed, 5 Aug 2026 11:17:44 -0400 Subject: [PATCH 10/12] _should_generate_op_id now returns an op_id or none --- pymongo/_telemetry.py | 15 ++++++++++----- pymongo/asynchronous/bulk.py | 13 ++++--------- pymongo/asynchronous/client_bulk.py | 7 +++---- pymongo/asynchronous/mongo_client.py | 8 ++++---- pymongo/synchronous/bulk.py | 13 ++++--------- pymongo/synchronous/client_bulk.py | 7 +++---- pymongo/synchronous/mongo_client.py | 8 ++++---- 7 files changed, 32 insertions(+), 39 deletions(-) diff --git a/pymongo/_telemetry.py b/pymongo/_telemetry.py index efccf162ac..499309b336 100644 --- a/pymongo/_telemetry.py +++ b/pymongo/_telemetry.py @@ -36,6 +36,7 @@ _ServerSelectionStatusMessage, _verbose_connection_error_reason, ) +from pymongo.message import _randint from pymongo.pool_shared import _ConnectionTelemetryInfo if TYPE_CHECKING: @@ -56,12 +57,16 @@ 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 events or logging.""" +def _generate_op_id_or_none(listeners: Optional[_EventListeners]) -> Optional[int]: + """Return a random operation id if it would be consumed by APM events or logging, else None.""" return ( - (listeners is not None and listeners.enabled_for_commands) - or _is_debug_enabled(_COMMAND_LOGGER) - or _is_debug_enabled(_SERVER_SELECTION_LOGGER) + _randint() + if ( + (listeners is not None and listeners.enabled_for_commands) + or _is_debug_enabled(_COMMAND_LOGGER) + or _is_debug_enabled(_SERVER_SELECTION_LOGGER) + ) + else None ) diff --git a/pymongo/asynchronous/bulk.py b/pymongo/asynchronous/bulk.py index 082da1fabe..ef28d77c1b 100644 --- a/pymongo/asynchronous/bulk.py +++ b/pymongo/asynchronous/bulk.py @@ -32,7 +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._telemetry import _generate_op_id_or_none from pymongo.asynchronous.client_session import AsyncClientSession, _validate_session_write_concern from pymongo.asynchronous.command_runner import ( run_bulk_write_command, @@ -62,7 +62,6 @@ _UPDATE, _BulkWriteContext, _EncryptedBulkWriteContext, - _randint, ) from pymongo.read_preferences import ReadPreference from pymongo.write_concern import WriteConcern @@ -457,7 +456,7 @@ async def execute_command( "upserted": [], } client = self.collection.database.client - op_id = _randint() if _should_generate_op_id(client._event_listeners) else None + op_id = _generate_op_id_or_none(client._event_listeners) async def retryable_bulk( session: Optional[AsyncClientSession], conn: AsyncConnection, retryable: bool @@ -492,7 +491,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() if _should_generate_op_id(listeners) else None + op_id = _generate_op_id_or_none(listeners) if not self.current_run: self.current_run = next(generator) @@ -545,11 +544,7 @@ 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() - if _should_generate_op_id(self.collection.database.client._event_listeners) - else None - ) + op_id = _generate_op_id_or_none(self.collection.database.client._event_listeners) try: await self._execute_command( generator, diff --git a/pymongo/asynchronous/client_bulk.py b/pymongo/asynchronous/client_bulk.py index f56ab062cb..367fdd492f 100644 --- a/pymongo/asynchronous/client_bulk.py +++ b/pymongo/asynchronous/client_bulk.py @@ -32,7 +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._telemetry import _generate_op_id_or_none from pymongo.asynchronous.client_session import ( AsyncClientSession, _validate_session_write_concern, @@ -70,7 +70,6 @@ from pymongo.message import ( _ClientBulkWriteContext, _convert_client_bulk_exception, - _randint, ) from pymongo.read_preferences import ReadPreference from pymongo.results import ( @@ -525,7 +524,7 @@ async def execute_command( "updateResults": {}, "deleteResults": {}, } - op_id = _randint() if _should_generate_op_id(self.client._event_listeners) else None + op_id = _generate_op_id_or_none(self.client._event_listeners) async def retryable_bulk( session: Optional[AsyncClientSession], @@ -566,7 +565,7 @@ async def execute_command_unack( db_name = "admin" cmd_name = "bulkWrite" listeners = self.client._event_listeners - op_id = _randint() if _should_generate_op_id(listeners) else None + op_id = _generate_op_id_or_none(listeners) bwc = self.bulk_ctx_class( db_name, diff --git a/pymongo/asynchronous/mongo_client.py b/pymongo/asynchronous/mongo_client.py index b5b7513676..db5c790d64 100644 --- a/pymongo/asynchronous/mongo_client.py +++ b/pymongo/asynchronous/mongo_client.py @@ -56,7 +56,7 @@ from bson.codec_options import DEFAULT_CODEC_OPTIONS, CodecOptions, TypeRegistry from bson.timestamp import Timestamp from pymongo import _csot, _op_id, common, helpers_shared, periodic_executor -from pymongo._telemetry import _should_generate_op_id, log_command_retry +from pymongo._telemetry import _generate_op_id_or_none, 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 @@ -93,7 +93,7 @@ _log_client_error, _log_or_warn, ) -from pymongo.message import _CursorAddress, _GetMore, _Query, _randint +from pymongo.message import _CursorAddress, _GetMore, _Query from pymongo.monitoring import ConnectionClosedReason, _EventListeners from pymongo.operations import ( DeleteMany, @@ -2888,8 +2888,8 @@ def __init__( self._deprioritized_servers: Optional[list[Server]] = None self._operation = operation # Only generate an operation id when APM/logging is enabled. - if operation_id is None and _should_generate_op_id(self._client._event_listeners): - operation_id = _randint() + if operation_id is None: + operation_id = _generate_op_id_or_none(self._client._event_listeners) self._operation_id = operation_id self._attempt_number = 0 self._is_run_command = is_run_command diff --git a/pymongo/synchronous/bulk.py b/pymongo/synchronous/bulk.py index c5b7ed8c1d..9338a2e6a1 100644 --- a/pymongo/synchronous/bulk.py +++ b/pymongo/synchronous/bulk.py @@ -32,7 +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._telemetry import _generate_op_id_or_none from pymongo.bulk_shared import ( _COMMANDS, _DELETE_ALL, @@ -57,7 +57,6 @@ _UPDATE, _BulkWriteContext, _EncryptedBulkWriteContext, - _randint, ) from pymongo.read_preferences import ReadPreference from pymongo.synchronous.client_session import ClientSession, _validate_session_write_concern @@ -457,7 +456,7 @@ def execute_command( "upserted": [], } client = self.collection.database.client - op_id = _randint() if _should_generate_op_id(client._event_listeners) else None + op_id = _generate_op_id_or_none(client._event_listeners) def retryable_bulk( session: Optional[ClientSession], conn: Connection, retryable: bool @@ -490,7 +489,7 @@ def execute_op_msg_no_results(self, conn: Connection, generator: Iterator[Any]) db_name = self.collection.database.name client = self.collection.database.client listeners = client._event_listeners - op_id = _randint() if _should_generate_op_id(listeners) else None + op_id = _generate_op_id_or_none(listeners) if not self.current_run: self.current_run = next(generator) @@ -543,11 +542,7 @@ def execute_command_no_results( # processing at the first error, even when the application # specified unacknowledged writeConcern. initial_write_concern = WriteConcern() - op_id = ( - _randint() - if _should_generate_op_id(self.collection.database.client._event_listeners) - else None - ) + op_id = _generate_op_id_or_none(self.collection.database.client._event_listeners) try: self._execute_command( generator, diff --git a/pymongo/synchronous/client_bulk.py b/pymongo/synchronous/client_bulk.py index 586b6911f9..3dca2f7234 100644 --- a/pymongo/synchronous/client_bulk.py +++ b/pymongo/synchronous/client_bulk.py @@ -32,7 +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._telemetry import _generate_op_id_or_none from pymongo.synchronous.client_session import ( ClientSession, _validate_session_write_concern, @@ -70,7 +70,6 @@ from pymongo.message import ( _ClientBulkWriteContext, _convert_client_bulk_exception, - _randint, ) from pymongo.read_preferences import ReadPreference from pymongo.results import ( @@ -523,7 +522,7 @@ def execute_command( "updateResults": {}, "deleteResults": {}, } - op_id = _randint() if _should_generate_op_id(self.client._event_listeners) else None + op_id = _generate_op_id_or_none(self.client._event_listeners) def retryable_bulk( session: Optional[ClientSession], @@ -564,7 +563,7 @@ def execute_command_unack( db_name = "admin" cmd_name = "bulkWrite" listeners = self.client._event_listeners - op_id = _randint() if _should_generate_op_id(listeners) else None + op_id = _generate_op_id_or_none(listeners) bwc = self.bulk_ctx_class( db_name, diff --git a/pymongo/synchronous/mongo_client.py b/pymongo/synchronous/mongo_client.py index c3058ec80e..3959455d3f 100644 --- a/pymongo/synchronous/mongo_client.py +++ b/pymongo/synchronous/mongo_client.py @@ -56,7 +56,7 @@ from bson.codec_options import DEFAULT_CODEC_OPTIONS, CodecOptions, TypeRegistry from bson.timestamp import Timestamp from pymongo import _csot, _op_id, common, helpers_shared, periodic_executor -from pymongo._telemetry import _should_generate_op_id, log_command_retry +from pymongo._telemetry import _generate_op_id_or_none, log_command_retry from pymongo.client_options import ClientOptions from pymongo.driver_info import DriverInfo from pymongo.errors import ( @@ -83,7 +83,7 @@ _log_client_error, _log_or_warn, ) -from pymongo.message import _CursorAddress, _GetMore, _Query, _randint +from pymongo.message import _CursorAddress, _GetMore, _Query from pymongo.monitoring import ConnectionClosedReason, _EventListeners from pymongo.operations import ( DeleteMany, @@ -2877,8 +2877,8 @@ def __init__( self._deprioritized_servers: Optional[list[Server]] = None self._operation = operation # Only generate an operation id when APM/logging is enabled. - if operation_id is None and _should_generate_op_id(self._client._event_listeners): - operation_id = _randint() + if operation_id is None: + operation_id = _generate_op_id_or_none(self._client._event_listeners) self._operation_id = operation_id self._attempt_number = 0 self._is_run_command = is_run_command From 5e68d3d8fc8d3ed7e1ff43a6b6c085bf4f8f0364 Mon Sep 17 00:00:00 2001 From: Noah Stapp Date: Wed, 5 Aug 2026 11:41:04 -0400 Subject: [PATCH 11/12] Missed a few uses of _is_debug_enabled --- pymongo/asynchronous/command_runner.py | 5 ++--- pymongo/asynchronous/topology.py | 3 +-- pymongo/synchronous/command_runner.py | 5 ++--- pymongo/synchronous/topology.py | 3 +-- 4 files changed, 6 insertions(+), 10 deletions(-) diff --git a/pymongo/asynchronous/command_runner.py b/pymongo/asynchronous/command_runner.py index e8defd50f1..aa0839fa4a 100644 --- a/pymongo/asynchronous/command_runner.py +++ b/pymongo/asynchronous/command_runner.py @@ -36,7 +36,6 @@ from __future__ import annotations import datetime -import logging import time from collections.abc import Mapping, MutableMapping, Sequence from typing import ( @@ -53,7 +52,7 @@ 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.logger import _COMMAND_LOGGER, _is_debug_enabled from pymongo.message import _BulkWriteContextBase, _convert_exception, _OpMsg from pymongo.monitoring import _is_speculative_authenticate @@ -167,7 +166,7 @@ async def _run_command( # Fast path: skip telemetry construction when logging and APM are disabled # Inline enabled check here for performance telemetry: Optional[_CommandTelemetry] = None - if (topology_id is not None and _COMMAND_LOGGER.isEnabledFor(logging.DEBUG)) or ( + if (topology_id is not None and _is_debug_enabled(_COMMAND_LOGGER)) or ( listeners is not None and listeners.enabled_for_commands ): telemetry = _CommandTelemetry( diff --git a/pymongo/asynchronous/topology.py b/pymongo/asynchronous/topology.py index cb1d348df0..1ed36151d2 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 @@ -377,7 +376,7 @@ async def select_server( ) if _csot.get_timeout(): _csot.set_rtt(server.description.min_round_trip_time) - if _SERVER_SELECTION_LOGGER.isEnabledFor(logging.DEBUG): + if _is_debug_enabled(_SERVER_SELECTION_LOGGER): log_server_selection_succeeded( self._topology_id, selector, diff --git a/pymongo/synchronous/command_runner.py b/pymongo/synchronous/command_runner.py index 0f37cf9440..21da51e2fc 100644 --- a/pymongo/synchronous/command_runner.py +++ b/pymongo/synchronous/command_runner.py @@ -36,7 +36,6 @@ from __future__ import annotations import datetime -import logging import time from collections.abc import Mapping, MutableMapping, Sequence from typing import ( @@ -53,7 +52,7 @@ 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.logger import _COMMAND_LOGGER, _is_debug_enabled from pymongo.message import _BulkWriteContextBase, _convert_exception, _OpMsg from pymongo.monitoring import _is_speculative_authenticate @@ -167,7 +166,7 @@ def _run_command( # Fast path: skip telemetry construction when logging and APM are disabled # Inline enabled check here for performance telemetry: Optional[_CommandTelemetry] = None - if (topology_id is not None and _COMMAND_LOGGER.isEnabledFor(logging.DEBUG)) or ( + if (topology_id is not None and _is_debug_enabled(_COMMAND_LOGGER)) or ( listeners is not None and listeners.enabled_for_commands ): telemetry = _CommandTelemetry( diff --git a/pymongo/synchronous/topology.py b/pymongo/synchronous/topology.py index 75278fa942..4d53af1711 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 @@ -377,7 +376,7 @@ def select_server( ) if _csot.get_timeout(): _csot.set_rtt(server.description.min_round_trip_time) - if _SERVER_SELECTION_LOGGER.isEnabledFor(logging.DEBUG): + if _is_debug_enabled(_SERVER_SELECTION_LOGGER): log_server_selection_succeeded( self._topology_id, selector, From d6d4b07f136f22d6d9400fb7ce0b61c812d47a81 Mon Sep 17 00:00:00 2001 From: Noah Stapp Date: Wed, 5 Aug 2026 12:07:00 -0400 Subject: [PATCH 12/12] Fix tests --- test/asynchronous/test_operation_id_retry.py | 27 ++++++++------------ test/test_operation_id_retry.py | 27 ++++++++------------ 2 files changed, 20 insertions(+), 34 deletions(-) diff --git a/test/asynchronous/test_operation_id_retry.py b/test/asynchronous/test_operation_id_retry.py index 581ecb4009..eedd25547a 100644 --- a/test/asynchronous/test_operation_id_retry.py +++ b/test/asynchronous/test_operation_id_retry.py @@ -24,9 +24,8 @@ import pymongo from bson.codec_options import DEFAULT_CODEC_OPTIONS -from pymongo import _op_id +from pymongo import _op_id, _telemetry from pymongo._telemetry import _CommandTelemetry -from pymongo.asynchronous import bulk, client_bulk, mongo_client from pymongo.asynchronous.encryption import _Encrypter from pymongo.asynchronous.helpers import _handle_reauth from pymongo.asynchronous.pool import AsyncConnection @@ -172,7 +171,7 @@ def recording_init( } async with self.fail_point(fail_point): with ( - patch.object(mongo_client, "_randint") as randint, + patch.object(_telemetry, "_randint") as randint, patch.object(_CommandTelemetry, "__init__", recording_init), ): self.assertIsNotNone( @@ -198,10 +197,7 @@ async def test_bulk_write_without_telemetry_creates_no_operation_id(self): coll = client.pymongo_test.test_operation_id_retry coll_w0 = coll.with_options(write_concern=WriteConcern(w=0)) - with ( - patch.object(bulk, "_randint") as bulk_randint, - patch.object(mongo_client, "_randint") as client_randint, - ): + with patch.object(_telemetry, "_randint") as randint: # Acknowledged result = await coll.bulk_write([InsertOne({})]) self.assertEqual(result.inserted_count, 1) @@ -212,16 +208,13 @@ async def test_bulk_write_without_telemetry_creates_no_operation_id(self): (await coll_w0.bulk_write([InsertOne({})], ordered=False)).acknowledged ) self.assertEqual( - bulk_randint.call_count, 0, "generated an operation id without APM/logging enabled" - ) - self.assertEqual( - client_randint.call_count, 0, "generated an operation id without APM/logging enabled" + randint.call_count, 0, "generated an operation id without APM/logging enabled" ) # Ensure we see randint() calls with APM enabled - with patch.object(bulk, "_randint", wraps=_randint) as bulk_randint: + with patch.object(_telemetry, "_randint", wraps=_randint) as wrapped_randint: await self.coll.bulk_write([InsertOne({})]) - self.assertEqual(bulk_randint.call_count, 1) + self.assertEqual(wrapped_randint.call_count, 1) @async_client_context.require_version_min(8, 0, 0, -24) async def test_client_bulk_write_without_telemetry_creates_no_operation_id(self): @@ -233,7 +226,7 @@ async def test_client_bulk_write_without_telemetry_creates_no_operation_id(self) self.assertFalse(client._event_listeners.enabled_for_commands) ns = "pymongo_test.test_operation_id_retry" - with patch.object(client_bulk, "_randint") as bulk_randint: + with patch.object(_telemetry, "_randint") as randint: # Acknowledged result = await client.bulk_write([InsertOne(namespace=ns, document={})]) self.assertEqual(result.inserted_count, 1) @@ -245,13 +238,13 @@ async def test_client_bulk_write_without_telemetry_creates_no_operation_id(self) ) self.assertFalse(result.acknowledged) self.assertEqual( - bulk_randint.call_count, 0, "generated an operation id without APM/logging enabled" + randint.call_count, 0, "generated an operation id without APM/logging enabled" ) # Ensure we see randint() calls with APM enabled - with patch.object(client_bulk, "_randint", wraps=_randint) as bulk_randint: + with patch.object(_telemetry, "_randint", wraps=_randint) as wrapped_randint: await self.client.bulk_write([InsertOne(namespace=ns, document={})]) - self.assertEqual(bulk_randint.call_count, 1) + self.assertEqual(wrapped_randint.call_count, 1) async def test_reauth_does_not_reuse_operation_id(self): class FakeConnection(AsyncConnection): diff --git a/test/test_operation_id_retry.py b/test/test_operation_id_retry.py index e1cb699e54..237447f994 100644 --- a/test/test_operation_id_retry.py +++ b/test/test_operation_id_retry.py @@ -24,14 +24,13 @@ import pymongo from bson.codec_options import DEFAULT_CODEC_OPTIONS -from pymongo import _op_id +from pymongo import _op_id, _telemetry from pymongo._telemetry import _CommandTelemetry from pymongo.errors import OperationFailure from pymongo.helpers_shared import _REAUTHENTICATION_REQUIRED_CODE from pymongo.logger import _COMMAND_LOGGER, _SERVER_SELECTION_LOGGER from pymongo.message import _randint from pymongo.operations import InsertOne -from pymongo.synchronous import bulk, client_bulk, mongo_client from pymongo.synchronous.encryption import _Encrypter from pymongo.synchronous.helpers import _handle_reauth from pymongo.synchronous.pool import Connection @@ -170,7 +169,7 @@ def recording_init( } with self.fail_point(fail_point): with ( - patch.object(mongo_client, "_randint") as randint, + patch.object(_telemetry, "_randint") as randint, patch.object(_CommandTelemetry, "__init__", recording_init), ): self.assertIsNotNone( @@ -196,10 +195,7 @@ def test_bulk_write_without_telemetry_creates_no_operation_id(self): coll = client.pymongo_test.test_operation_id_retry coll_w0 = coll.with_options(write_concern=WriteConcern(w=0)) - with ( - patch.object(bulk, "_randint") as bulk_randint, - patch.object(mongo_client, "_randint") as client_randint, - ): + with patch.object(_telemetry, "_randint") as randint: # Acknowledged result = coll.bulk_write([InsertOne({})]) self.assertEqual(result.inserted_count, 1) @@ -208,16 +204,13 @@ def test_bulk_write_without_telemetry_creates_no_operation_id(self): # Unacknowledged unordered self.assertFalse((coll_w0.bulk_write([InsertOne({})], ordered=False)).acknowledged) self.assertEqual( - bulk_randint.call_count, 0, "generated an operation id without APM/logging enabled" - ) - self.assertEqual( - client_randint.call_count, 0, "generated an operation id without APM/logging enabled" + randint.call_count, 0, "generated an operation id without APM/logging enabled" ) # Ensure we see randint() calls with APM enabled - with patch.object(bulk, "_randint", wraps=_randint) as bulk_randint: + with patch.object(_telemetry, "_randint", wraps=_randint) as wrapped_randint: self.coll.bulk_write([InsertOne({})]) - self.assertEqual(bulk_randint.call_count, 1) + self.assertEqual(wrapped_randint.call_count, 1) @client_context.require_version_min(8, 0, 0, -24) def test_client_bulk_write_without_telemetry_creates_no_operation_id(self): @@ -229,7 +222,7 @@ def test_client_bulk_write_without_telemetry_creates_no_operation_id(self): self.assertFalse(client._event_listeners.enabled_for_commands) ns = "pymongo_test.test_operation_id_retry" - with patch.object(client_bulk, "_randint") as bulk_randint: + with patch.object(_telemetry, "_randint") as randint: # Acknowledged result = client.bulk_write([InsertOne(namespace=ns, document={})]) self.assertEqual(result.inserted_count, 1) @@ -241,13 +234,13 @@ def test_client_bulk_write_without_telemetry_creates_no_operation_id(self): ) self.assertFalse(result.acknowledged) self.assertEqual( - bulk_randint.call_count, 0, "generated an operation id without APM/logging enabled" + randint.call_count, 0, "generated an operation id without APM/logging enabled" ) # Ensure we see randint() calls with APM enabled - with patch.object(client_bulk, "_randint", wraps=_randint) as bulk_randint: + with patch.object(_telemetry, "_randint", wraps=_randint) as wrapped_randint: self.client.bulk_write([InsertOne(namespace=ns, document={})]) - self.assertEqual(bulk_randint.call_count, 1) + self.assertEqual(wrapped_randint.call_count, 1) def test_reauth_does_not_reuse_operation_id(self): class FakeConnection(Connection):