diff --git a/pymongo/_telemetry.py b/pymongo/_telemetry.py index f972a402df..499309b336 100644 --- a/pymongo/_telemetry.py +++ b/pymongo/_telemetry.py @@ -17,12 +17,12 @@ from __future__ import annotations import datetime -import logging import queue import time 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, @@ -31,10 +31,12 @@ _CommandStatusMessage, _ConnectionStatusMessage, _debug_log, + _is_debug_enabled, _SDAMStatusMessage, _ServerSelectionStatusMessage, _verbose_connection_error_reason, ) +from pymongo.message import _randint from pymongo.pool_shared import _ConnectionTelemetryInfo if TYPE_CHECKING: @@ -55,12 +57,29 @@ def _monotonic_duration(start: float) -> float: return max(0.0, time.monotonic() - start) +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 ( + _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 + ) + + 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 both APM events and command + logging are disabled, only the gate flags and the monotonic duration clock + are maintained. """ __slots__ = ( @@ -68,7 +87,7 @@ class _CommandTelemetry: "_cmd", "_conn", "_dbname", - "_duration", + "_duration_s", "_listeners", "_name", "_op_id", @@ -88,20 +107,25 @@ 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) + # 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 _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 + 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 +146,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 +166,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 +177,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 +210,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, @@ -218,7 +244,7 @@ class _CmapTelemetry: "_client_id", "_listeners", "_log", - "_publish", + "_should_publish", ) def __init__( @@ -232,18 +258,18 @@ def __init__( self._client_id = client_id self._address = address self._listeners = listeners - self._publish = publish + # The CMAP listener set is fixed once the client is constructed + # (_EventListeners copies the global listeners at __init__), so this + # gate is static for the life of the pool. + # NOTE: the checkout/checkin fast paths in pool.py inline this gate for performance + # They must be kept in sync with any gating changes + self._should_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 - @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( @@ -421,7 +447,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: @@ -502,7 +528,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, @@ -513,21 +539,16 @@ 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: """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: @@ -652,7 +673,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, @@ -705,7 +726,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, @@ -721,7 +742,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)) @@ -733,7 +754,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/bulk.py b/pymongo/asynchronous/bulk.py index 3075afa2b3..ef28d77c1b 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 _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, @@ -61,7 +62,6 @@ _UPDATE, _BulkWriteContext, _EncryptedBulkWriteContext, - _randint, ) from pymongo.read_preferences import ReadPreference from pymongo.write_concern import WriteConcern @@ -339,7 +339,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 +455,8 @@ async def execute_command( "nRemoved": 0, "upserted": [], } - op_id = _randint() + client = self.collection.database.client + op_id = _generate_op_id_or_none(client._event_listeners) async def retryable_bulk( session: Optional[AsyncClientSession], conn: AsyncConnection, retryable: bool @@ -470,7 +471,6 @@ async def retryable_bulk( full_result, ) - client = self.collection.database.client _ = await client._retryable_write( self.is_retryable, retryable_bulk, @@ -491,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() + op_id = _generate_op_id_or_none(listeners) if not self.current_run: self.current_run = next(generator) @@ -544,7 +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() + 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 cfa2ea9853..367fdd492f 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 _generate_op_id_or_none from pymongo.asynchronous.client_session import ( AsyncClientSession, _validate_session_write_concern, @@ -69,7 +70,6 @@ from pymongo.message import ( _ClientBulkWriteContext, _convert_client_bulk_exception, - _randint, ) from pymongo.read_preferences import ReadPreference from pymongo.results import ( @@ -370,7 +370,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 +524,7 @@ async def execute_command( "updateResults": {}, "deleteResults": {}, } - op_id = _randint() + op_id = _generate_op_id_or_none(self.client._event_listeners) async def retryable_bulk( session: Optional[AsyncClientSession], @@ -565,7 +565,7 @@ async def execute_command_unack( db_name = "admin" cmd_name = "bulkWrite" listeners = self.client._event_listeners - op_id = _randint() + op_id = _generate_op_id_or_none(listeners) bwc = self.bulk_ctx_class( db_name, diff --git a/pymongo/asynchronous/command_runner.py b/pymongo/asynchronous/command_runner.py index 683dc0d8e8..aa0839fa4a 100644 --- a/pymongo/asynchronous/command_runner.py +++ b/pymongo/asynchronous/command_runner.py @@ -36,6 +36,7 @@ from __future__ import annotations import datetime +import time from collections.abc import Mapping, MutableMapping, Sequence from typing import ( TYPE_CHECKING, @@ -47,10 +48,11 @@ ) from bson import _decode_all_selective -from pymongo import _csot, _op_id, helpers_shared, message +from pymongo import _csot, helpers_shared, message from pymongo._telemetry import _CommandTelemetry from pymongo.compression_support import _NO_COMPRESSION from pymongo.errors import NotPrimaryError, OperationFailure +from pymongo.logger import _COMMAND_LOGGER, _is_debug_enabled from pymongo.message import _BulkWriteContextBase, _convert_exception, _OpMsg from pymongo.monitoring import _is_speculative_authenticate @@ -100,8 +102,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 +131,8 @@ async def _run_command( :param orig: The command document published in the ``STARTED`` APM event; defaults to ``cmd`` (differs only when the wire command was mutated, e.g. with a read preference or after encryption). - :param op_id: The APM operation id; defaults to the ``OP_ID`` contextvar, - then ``request_id``. + :param op_id: The APM operation id; when ``None`` it is resolved from the + ``OP_ID`` contextvar (then ``request_id``) only if APM/logging is enabled. :param command_name: The command name for the ``SUCCEEDED``/``FAILED`` APM events; defaults to the first key of ``cmd``. :param check: Raise OperationFailure on a command error. @@ -159,11 +162,20 @@ async def _run_command( command_name = name if orig is None: orig = cmd - if op_id is None: - op_id = _op_id.OP_ID.get() - telemetry = _CommandTelemetry(topology_id, conn, listeners, cmd, dbname, request_id, op_id) - telemetry.started(orig, ensure_db) + # Fast path: 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 _is_debug_enabled(_COMMAND_LOGGER)) or ( + listeners is not None and listeners.enabled_for_commands + ): + telemetry = _CommandTelemetry( + topology_id, conn, listeners, cmd, dbname, request_id, op_id, name=name + ) + telemetry.started(orig, ensure_db) + start = 0.0 + else: + start = time.monotonic() reply: Optional[_OpMsg] = None docs: list[dict[str, Any]] = [{"ok": 1}] @@ -211,10 +223,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()) @@ -222,7 +239,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, duration_s async def run_bulk_write_command( @@ -235,8 +252,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 +326,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 +346,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..db5c790d64 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 _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 @@ -91,12 +90,10 @@ ) from pymongo.logger import ( _CLIENT_LOGGER, - _COMMAND_LOGGER, - _SERVER_SELECTION_LOGGER, _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, @@ -2890,16 +2887,9 @@ 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 - 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) - ): - operation_id = _randint() + # Only generate an operation id when APM/logging is enabled. + 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/asynchronous/pool.py b/pymongo/asynchronous/pool.py index fdf3b1d816..a126962e4a 100644 --- a/pymongo/asynchronous/pool.py +++ b/pymongo/asynchronous/pool.py @@ -62,6 +62,7 @@ _async_create_condition, _async_create_lock, ) +from pymongo.logger import _CONNECTION_LOGGER, _is_debug_enabled from pymongo.monitoring import ( ConnectionCheckOutFailedReason, ConnectionClosedReason, @@ -1113,7 +1114,9 @@ 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._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() else: @@ -1230,12 +1233,20 @@ def __init__( async def __aenter__(self) -> AsyncConnection: pool = self._pool - checkout_started_time = pool._telemetry.checkout_started() + telemetry = pool._telemetry + # Fast path: skip telemetry calls when CMAP events/logging are disabled + if not telemetry._should_publish and not ( + telemetry._log and _is_debug_enabled(_CONNECTION_LOGGER) + ): + 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..1ed36151d2 100644 --- a/pymongo/asynchronous/topology.py +++ b/pymongo/asynchronous/topology.py @@ -55,6 +55,7 @@ _async_create_condition, _async_create_lock, ) +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 ( @@ -283,10 +284,13 @@ 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 does not have APM events, gate only on logging + ss: Optional[_ServerSelectionTelemetry] = None + if _is_debug_enabled(_SERVER_SELECTION_LOGGER): + ss = _ServerSelectionTelemetry( + self._topology_id, selector, operation, operation_id, self.description + ) + ss.started() server_descriptions = self._description.apply_selector( selector, @@ -300,12 +304,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 +376,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 _is_debug_enabled(_SERVER_SELECTION_LOGGER): + 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/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/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..252b0daef0 100644 --- a/pymongo/monitoring.py +++ b/pymongo/monitoring.py @@ -516,6 +516,7 @@ 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: + # Check the name first, doc may be a RawBSONDocument where `in` decodes the whole document. return bool( command_name.lower() in ("hello", HelloCompat.LEGACY_CMD) and "speculativeAuthenticate" in doc diff --git a/pymongo/synchronous/bulk.py b/pymongo/synchronous/bulk.py index 36081fe222..9338a2e6a1 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 _generate_op_id_or_none from pymongo.bulk_shared import ( _COMMANDS, _DELETE_ALL, @@ -56,7 +57,6 @@ _UPDATE, _BulkWriteContext, _EncryptedBulkWriteContext, - _randint, ) from pymongo.read_preferences import ReadPreference from pymongo.synchronous.client_session import ClientSession, _validate_session_write_concern @@ -339,7 +339,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 +455,8 @@ def execute_command( "nRemoved": 0, "upserted": [], } - op_id = _randint() + client = self.collection.database.client + op_id = _generate_op_id_or_none(client._event_listeners) def retryable_bulk( session: Optional[ClientSession], conn: Connection, retryable: bool @@ -470,7 +471,6 @@ def retryable_bulk( full_result, ) - client = self.collection.database.client _ = client._retryable_write( self.is_retryable, retryable_bulk, @@ -489,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() + op_id = _generate_op_id_or_none(listeners) if not self.current_run: self.current_run = next(generator) @@ -542,7 +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() + 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 4cf1d9dbb0..3dca2f7234 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 _generate_op_id_or_none from pymongo.synchronous.client_session import ( ClientSession, _validate_session_write_concern, @@ -69,7 +70,6 @@ from pymongo.message import ( _ClientBulkWriteContext, _convert_client_bulk_exception, - _randint, ) from pymongo.read_preferences import ReadPreference from pymongo.results import ( @@ -368,7 +368,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 +522,7 @@ def execute_command( "updateResults": {}, "deleteResults": {}, } - op_id = _randint() + op_id = _generate_op_id_or_none(self.client._event_listeners) def retryable_bulk( session: Optional[ClientSession], @@ -563,7 +563,7 @@ def execute_command_unack( db_name = "admin" cmd_name = "bulkWrite" listeners = self.client._event_listeners - op_id = _randint() + op_id = _generate_op_id_or_none(listeners) bwc = self.bulk_ctx_class( db_name, diff --git a/pymongo/synchronous/command_runner.py b/pymongo/synchronous/command_runner.py index e81f514003..21da51e2fc 100644 --- a/pymongo/synchronous/command_runner.py +++ b/pymongo/synchronous/command_runner.py @@ -36,6 +36,7 @@ from __future__ import annotations import datetime +import time from collections.abc import Mapping, MutableMapping, Sequence from typing import ( TYPE_CHECKING, @@ -47,10 +48,11 @@ ) from bson import _decode_all_selective -from pymongo import _csot, _op_id, helpers_shared, message +from pymongo import _csot, helpers_shared, message from pymongo._telemetry import _CommandTelemetry from pymongo.compression_support import _NO_COMPRESSION from pymongo.errors import NotPrimaryError, OperationFailure +from pymongo.logger import _COMMAND_LOGGER, _is_debug_enabled from pymongo.message import _BulkWriteContextBase, _convert_exception, _OpMsg from pymongo.monitoring import _is_speculative_authenticate @@ -100,8 +102,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 +131,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,11 +162,20 @@ def _run_command( command_name = name if orig is None: orig = cmd - if op_id is None: - op_id = _op_id.OP_ID.get() - telemetry = _CommandTelemetry(topology_id, conn, listeners, cmd, dbname, request_id, op_id) - telemetry.started(orig, ensure_db) + # Fast path: 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 _is_debug_enabled(_COMMAND_LOGGER)) or ( + listeners is not None and listeners.enabled_for_commands + ): + telemetry = _CommandTelemetry( + topology_id, conn, listeners, cmd, dbname, request_id, op_id, name=name + ) + telemetry.started(orig, ensure_db) + start = 0.0 + else: + start = time.monotonic() reply: Optional[_OpMsg] = None docs: list[dict[str, Any]] = [{"ok": 1}] @@ -211,10 +223,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()) @@ -222,7 +239,7 @@ def _run_command( "list[dict[str, Any]]", _decode_all_selective(decrypted, codec_options, user_fields) ) - return docs, reply, telemetry.duration + return docs, reply, duration_s def run_bulk_write_command( @@ -235,8 +252,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 +326,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 +346,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..3959455d3f 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 _generate_op_id_or_none, log_command_retry from pymongo.client_options import ClientOptions from pymongo.driver_info import DriverInfo from pymongo.errors import ( @@ -81,12 +80,10 @@ ) from pymongo.logger import ( _CLIENT_LOGGER, - _COMMAND_LOGGER, - _SERVER_SELECTION_LOGGER, _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, @@ -2879,16 +2876,9 @@ 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 - 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) - ): - operation_id = _randint() + # Only generate an operation id when APM/logging is enabled. + 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/pool.py b/pymongo/synchronous/pool.py index 1304921781..b75415eeeb 100644 --- a/pymongo/synchronous/pool.py +++ b/pymongo/synchronous/pool.py @@ -59,6 +59,7 @@ _create_condition, _create_lock, ) +from pymongo.logger import _CONNECTION_LOGGER, _is_debug_enabled from pymongo.monitoring import ( ConnectionCheckOutFailedReason, ConnectionClosedReason, @@ -1109,7 +1110,9 @@ 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._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() else: @@ -1226,12 +1229,20 @@ def __init__( def __enter__(self) -> Connection: pool = self._pool - checkout_started_time = pool._telemetry.checkout_started() + telemetry = pool._telemetry + # Fast path: skip telemetry calls when CMAP events/logging are disabled + if not telemetry._should_publish and not ( + telemetry._log and _is_debug_enabled(_CONNECTION_LOGGER) + ): + 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..4d53af1711 100644 --- a/pymongo/synchronous/topology.py +++ b/pymongo/synchronous/topology.py @@ -51,6 +51,7 @@ _create_condition, _create_lock, ) +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 ( @@ -283,10 +284,13 @@ 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 does not have APM events, gate only on logging + ss: Optional[_ServerSelectionTelemetry] = None + if _is_debug_enabled(_SERVER_SELECTION_LOGGER): + ss = _ServerSelectionTelemetry( + self._topology_id, selector, operation, operation_id, self.description + ) + ss.started() server_descriptions = self._description.apply_selector( selector, @@ -300,12 +304,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 +376,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 _is_debug_enabled(_SERVER_SELECTION_LOGGER): + 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_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/asynchronous/test_operation_id_retry.py b/test/asynchronous/test_operation_id_retry.py index e5a1d16c90..eedd25547a 100644 --- a/test/asynchronous/test_operation_id_retry.py +++ b/test/asynchronous/test_operation_id_retry.py @@ -24,16 +24,17 @@ 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 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 +140,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) @@ -151,10 +152,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}, @@ -166,7 +171,7 @@ def recording_init(self, topology_id, conn, listeners, cmd, dbname, request_id, } 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( @@ -178,10 +183,69 @@ def recording_init(self, topology_id, conn, listeners, cmd, dbname, request_id, ) 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_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(_telemetry, "_randint") as 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( + randint.call_count, 0, "generated an operation id without APM/logging enabled" + ) + + # Ensure we see randint() calls with APM enabled + with patch.object(_telemetry, "_randint", wraps=_randint) as wrapped_randint: + await self.coll.bulk_write([InsertOne({})]) + 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): + 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(_telemetry, "_randint") as 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( + randint.call_count, 0, "generated an operation id without APM/logging enabled" + ) + + # Ensure we see randint() calls with APM enabled + with patch.object(_telemetry, "_randint", wraps=_randint) as wrapped_randint: + await self.client.bulk_write([InsertOne(namespace=ns, document={})]) + self.assertEqual(wrapped_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_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) diff --git a/test/test_operation_id_retry.py b/test/test_operation_id_retry.py index cae8927a37..237447f994 100644 --- a/test/test_operation_id_retry.py +++ b/test/test_operation_id_retry.py @@ -24,16 +24,17 @@ 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 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 +138,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) @@ -149,10 +150,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}, @@ -164,7 +169,7 @@ def recording_init(self, topology_id, conn, listeners, cmd, dbname, request_id, } 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( @@ -176,10 +181,67 @@ def recording_init(self, topology_id, conn, listeners, cmd, dbname, request_id, ) 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_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(_telemetry, "_randint") as 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( + randint.call_count, 0, "generated an operation id without APM/logging enabled" + ) + + # Ensure we see randint() calls with APM enabled + with patch.object(_telemetry, "_randint", wraps=_randint) as wrapped_randint: + self.coll.bulk_write([InsertOne({})]) + 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): + 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(_telemetry, "_randint") as 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( + randint.call_count, 0, "generated an operation id without APM/logging enabled" + ) + + # Ensure we see randint() calls with APM enabled + with patch.object(_telemetry, "_randint", wraps=_randint) as wrapped_randint: + self.client.bulk_write([InsertOne(namespace=ns, document={})]) + self.assertEqual(wrapped_randint.call_count, 1) + def test_reauth_does_not_reuse_operation_id(self): class FakeConnection(Connection): def __init__(self):