PYTHON-5963 Only attach a shared operation id to retry attempts when APM/logging is enabled

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.

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
 

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
 

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):

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):