diff --git a/pymongo/_op_id.py b/pymongo/_op_id.py index d0de439789..164ceafa62 100644 --- a/pymongo/_op_id.py +++ b/pymongo/_op_id.py @@ -14,7 +14,7 @@ """Internal helpers for the APM operation id. -The retryable read/write logic sets OP_ID for the duration of each attempt so +The retryable read/write logic sets OP_ID if APM/logging is enabled for the duration of each attempt so that every attempt of one logical operation publishes the same operation_id. Commands run outside that scope (handshake, auth, killCursors, pinned-cursor getMores) read the default None and fall back to their request_id. diff --git a/pymongo/asynchronous/mongo_client.py b/pymongo/asynchronous/mongo_client.py index cd1cefea13..3ed5f551d0 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 @@ -91,6 +92,8 @@ ) from pymongo.logger import ( _CLIENT_LOGGER, + _COMMAND_LOGGER, + _SERVER_SELECTION_LOGGER, _log_client_error, _log_or_warn, ) @@ -2790,7 +2793,17 @@ def __init__( self._server: Server = None # type: ignore self._deprioritized_servers: list[Server] = [] self._operation = operation - self._operation_id = operation_id if operation_id is not None else _randint() + # 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() + self._operation_id = operation_id self._attempt_number = 0 self._is_run_command = is_run_command self._is_aggregate_write = is_aggregate_write @@ -3019,7 +3032,9 @@ async def _write(self) -> T: self._retryable = False if self._retrying: self._log_retry(is_write=True) - # One operation id across all attempts of this operation. + # 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): return await self._func(self._session, conn, self._retryable) # type: ignore except PyMongoError as exc: @@ -3044,7 +3059,9 @@ async def _read(self) -> T: self._check_last_error() if self._retrying: self._log_retry(is_write=False) - # One operation id across all attempts of this operation. + # 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): return await self._func(self._session, self._server, conn, read_pref) # type: ignore diff --git a/pymongo/synchronous/mongo_client.py b/pymongo/synchronous/mongo_client.py index d3f6a1e217..3e2495f524 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 @@ -81,6 +82,8 @@ ) from pymongo.logger import ( _CLIENT_LOGGER, + _COMMAND_LOGGER, + _SERVER_SELECTION_LOGGER, _log_client_error, _log_or_warn, ) @@ -2781,7 +2784,17 @@ def __init__( self._server: Server = None # type: ignore self._deprioritized_servers: list[Server] = [] self._operation = operation - self._operation_id = operation_id if operation_id is not None else _randint() + # 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() + self._operation_id = operation_id self._attempt_number = 0 self._is_run_command = is_run_command self._is_aggregate_write = is_aggregate_write @@ -3010,7 +3023,9 @@ def _write(self) -> T: self._retryable = False if self._retrying: self._log_retry(is_write=True) - # One operation id across all attempts of this operation. + # 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): return self._func(self._session, conn, self._retryable) # type: ignore except PyMongoError as exc: @@ -3035,7 +3050,9 @@ def _read(self) -> T: self._check_last_error() if self._retrying: self._log_retry(is_write=False) - # One operation id across all attempts of this operation. + # 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): return self._func(self._session, self._server, conn, read_pref) # type: ignore diff --git a/test/asynchronous/test_operation_id_retry.py b/test/asynchronous/test_operation_id_retry.py index 91e3cd9011..e5a1d16c90 100644 --- a/test/asynchronous/test_operation_id_retry.py +++ b/test/asynchronous/test_operation_id_retry.py @@ -16,18 +16,23 @@ from __future__ import annotations +import logging import sys +from unittest.mock import patch sys.path[0:0] = [""] import pymongo 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.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.operations import InsertOne from test.asynchronous import AsyncIntegrationTest, async_client_context, unittest from test.utils_shared import AllowListEventListener @@ -134,6 +139,49 @@ 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): + appname = _APP_NAME + "noapm" + client = await self.async_rs_or_single_client(appname=appname) + + # 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) + + find_op_ids = [] + original_init = _CommandTelemetry.__init__ + + def recording_init(self, topology_id, conn, listeners, cmd, dbname, request_id, op_id): + if next(iter(cmd)) == "find": + find_op_ids.append(op_id) + original_init(self, topology_id, conn, listeners, cmd, dbname, request_id, op_id) + + fail_point = { + "mode": {"times": 1}, + "data": { + "failCommands": ["find"], + "closeConnection": True, + "appName": appname, + }, + } + async with self.fail_point(fail_point): + with ( + patch.object(mongo_client, "_randint") as randint, + patch.object(_CommandTelemetry, "__init__", recording_init), + ): + self.assertIsNotNone( + await client.pymongo_test.test_operation_id_retry.find_one({"_id": 1}) + ) + + self.assertEqual( + randint.call_count, 0, "generated an operation id without APM/logging enabled" + ) + self.assertEqual( + find_op_ids, + [None, None], + "expected two attempts, neither carrying a shared operation id", + ) + 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 7224d0626a..cae8927a37 100644 --- a/test/test_operation_id_retry.py +++ b/test/test_operation_id_retry.py @@ -16,16 +16,21 @@ from __future__ import annotations +import logging import sys +from unittest.mock import patch sys.path[0:0] = [""] import pymongo from bson.codec_options import DEFAULT_CODEC_OPTIONS from pymongo import _op_id +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.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 @@ -132,6 +137,49 @@ 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): + appname = _APP_NAME + "noapm" + client = self.rs_or_single_client(appname=appname) + + # 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) + + find_op_ids = [] + original_init = _CommandTelemetry.__init__ + + def recording_init(self, topology_id, conn, listeners, cmd, dbname, request_id, op_id): + if next(iter(cmd)) == "find": + find_op_ids.append(op_id) + original_init(self, topology_id, conn, listeners, cmd, dbname, request_id, op_id) + + fail_point = { + "mode": {"times": 1}, + "data": { + "failCommands": ["find"], + "closeConnection": True, + "appName": appname, + }, + } + with self.fail_point(fail_point): + with ( + patch.object(mongo_client, "_randint") as randint, + patch.object(_CommandTelemetry, "__init__", recording_init), + ): + self.assertIsNotNone( + client.pymongo_test.test_operation_id_retry.find_one({"_id": 1}) + ) + + self.assertEqual( + randint.call_count, 0, "generated an operation id without APM/logging enabled" + ) + self.assertEqual( + find_op_ids, + [None, None], + "expected two attempts, neither carrying a shared operation id", + ) + def test_reauth_does_not_reuse_operation_id(self): class FakeConnection(Connection): def __init__(self):