diff --git a/.evergreen/generated_configs/variants.yml b/.evergreen/generated_configs/variants.yml index 26027888fe..b9b054bda0 100644 --- a/.evergreen/generated_configs/variants.yml +++ b/.evergreen/generated_configs/variants.yml @@ -468,6 +468,18 @@ buildvariants: - windows-2022-latest-small batchtime: 1440 + # Otel tests + - name: otel-rhel8 + tasks: + - name: .test-non-standard .standalone-noauth-nossl + display_name: OTel RHEL8 + run_on: + - rhel8.10-small + expansions: + TEST_NAME: otel + COVERAGE: "1" + tags: [pr] + # Perf tests - name: performance-benchmarks tasks: diff --git a/.evergreen/scripts/generate_config.py b/.evergreen/scripts/generate_config.py index 9ec13d5037..2ef5a34b9e 100644 --- a/.evergreen/scripts/generate_config.py +++ b/.evergreen/scripts/generate_config.py @@ -514,6 +514,21 @@ def create_doctests_variants(): ] +def create_otel_variants(): + host = DEFAULT_HOST + # Merge otel's coverage into the combined report; see setup_tests.py's COVERAGE handling. + expansions = dict(TEST_NAME="otel", COVERAGE="1") + return [ + create_variant( + [".test-non-standard .standalone-noauth-nossl"], + get_variant_name("OTel", host), + host=host, + tags=["pr"], + expansions=expansions, + ) + ] + + def create_atlas_connect_variants(): host = DEFAULT_HOST return [ diff --git a/.evergreen/scripts/setup_tests.py b/.evergreen/scripts/setup_tests.py index c8a15ad4a6..724556997e 100644 --- a/.evergreen/scripts/setup_tests.py +++ b/.evergreen/scripts/setup_tests.py @@ -46,6 +46,7 @@ "enterprise_auth": "gssapi", "kms": "encryption", "ocsp": "ocsp", + "otel": "opentelemetry", "pyopenssl": "ocsp", } @@ -502,6 +503,16 @@ def handle_test_env() -> None: if test_name == "numpy": UV_ARGS.append("--with numpy") + if test_name == "otel": + # The SDK is test-only tooling (for the in-memory span exporter); the driver + # itself must not depend on it, only on opentelemetry-api (the "opentelemetry" extra). + # The lower bound matters: under --resolution=lowest-direct an unpinned + # --with dependency resolves to its oldest version (1.0.0), which pins + # opentelemetry-api==1.0.0 and drags the API below the driver's floor + # (>=1.20.0 in requirements/opentelemetry.txt). Keep the SDK's floor in + # sync with that file when bumping the supported API version. + UV_ARGS.append("--with opentelemetry-sdk>=1.20.0") + if test_name == "perf": data_dir = ROOT / "specifications/source/benchmarking/data" if not data_dir.exists(): @@ -527,9 +538,10 @@ def handle_test_env() -> None: else: TEST_ARGS = f"test/performance/async_perf_test.py {TEST_ARGS}" - # Add coverage if requested. + # Add coverage if requested, either via --cov or a pre-set COVERAGE expansion + # (e.g. a buildvariant that wants its coverage merged into the combined report). # Only cover CPython. PyPy reports suspiciously low coverage. - if opts.cov and platform.python_implementation() == "CPython": + if (opts.cov or is_set("COVERAGE")) and platform.python_implementation() == "CPython": # Keep in sync with combine-coverage.sh. # coverage >=5 is needed for relative_files=true. UV_ARGS.append("--group coverage") diff --git a/.evergreen/scripts/utils.py b/.evergreen/scripts/utils.py index fb69539f16..28ca274d05 100644 --- a/.evergreen/scripts/utils.py +++ b/.evergreen/scripts/utils.py @@ -47,6 +47,7 @@ class Distro: "load_balancer": "load_balancer", "mockupdb": "mockupdb", "ocsp": "ocsp", + "otel": "otel", "perf": "perf", "sfp": "sfp", "numpy": "", diff --git a/.gitignore b/.gitignore index 63813daf02..f65211826f 100644 --- a/.gitignore +++ b/.gitignore @@ -30,6 +30,7 @@ expansion.yml .evergreen/scripts/test-env.sh specifications/ results.json +otel-bench-results/ .synchro-modified .evergreen/atlas_x509_dev_client_certificate.pem .evergreen/atlas_x509_sfp_client_certificate.pem diff --git a/doc/changelog.rst b/doc/changelog.rst index fd8ef94334..f418453279 100644 --- a/doc/changelog.rst +++ b/doc/changelog.rst @@ -196,6 +196,12 @@ PyMongo 4.18 brings a number of changes including: :meth:`~pymongo.synchronous.database.Database.aggregate`, and :meth:`~pymongo.asynchronous.collection.AsyncCollection.list_search_indexes` and :meth:`~pymongo.synchronous.collection.Collection.list_search_indexes`. +- Added optional OpenTelemetry command-span support, conforming to the + `OpenTelemetry driver specification `_. + Enable it with the ``tracing`` :class:`~pymongo.mongo_client.MongoClient` + option or the ``OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED`` environment + variable. Install the ``opentelemetry-api`` package, or use the + ``pymongo[opentelemetry]`` extra, to enable this feature. Changes in Version 4.17.0 (2026/04/20) -------------------------------------- diff --git a/justfile b/justfile index 9b37109c74..328c4fda8e 100644 --- a/justfile +++ b/justfile @@ -2,7 +2,7 @@ set shell := ["bash", "-c"] # Commonly used command segments. -typing_run := "uv run --group typing --extra aws --extra encryption --with numpy --extra ocsp --extra snappy --extra test --extra zstd" +typing_run := "uv run --group typing --extra aws --extra encryption --with numpy --extra ocsp --extra opentelemetry --with opentelemetry-sdk --extra snappy --extra test --extra zstd" docs_run := "uv run --extra docs" doc_build := "./doc/_build" mypy_args := "--install-types --non-interactive" diff --git a/pymongo/_otel.py b/pymongo/_otel.py new file mode 100644 index 0000000000..b43f831418 --- /dev/null +++ b/pymongo/_otel.py @@ -0,0 +1,475 @@ +# Copyright 2026-present MongoDB, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# https://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Optional OpenTelemetry command-span support. + +Kept separate from :mod:`pymongo._telemetry` so that module stays free of +``opentelemetry`` import guards. Every function here is a no-op when +``opentelemetry`` isn't installed or tracing isn't enabled. +""" + +from __future__ import annotations + +import os +from collections.abc import Mapping, MutableMapping +from typing import TYPE_CHECKING, Any, Optional, TypedDict + +from bson import json_util +from bson.json_util import _truncate_documents +from pymongo._version import __version__ +from pymongo.logger import _HELLO_COMMANDS, _JSON_OPTIONS, _SENSITIVE_COMMANDS + +try: + from opentelemetry import context as otel_context + from opentelemetry import trace + from opentelemetry.trace import SpanKind, Status, StatusCode + + _HAS_OPENTELEMETRY = True + # Safe to cache at import time: opentelemetry.trace.get_tracer() returns a + # ProxyTracer when no real TracerProvider is registered yet, and that proxy + # transparently starts delegating to the real tracer once the application + # calls trace.set_tracer_provider() later, so this doesn't bind us to a + # permanently-inert no-op tracer. + _TRACER: Optional[Tracer] = trace.get_tracer("PyMongo", __version__) + # Command spans are always CLIENT kind; hoisted to avoid an attribute + # lookup on every span creation. + _SPAN_KIND_CLIENT = SpanKind.CLIENT +except ImportError: + _HAS_OPENTELEMETRY = False + _TRACER = None + +if TYPE_CHECKING: + from opentelemetry.trace import Span, Tracer + + from pymongo.pool_shared import _ConnectionTelemetryInfo + from pymongo.typings import _DocumentOut + + +class TracingOptions(TypedDict): + """The shape of the ``MongoClient`` ``tracing`` option. + + ``enabled`` and ``query_text_max_length`` are None when the client didn't + configure them, so the environment variables can be consulted; any explicit + value (including ``False``, to force tracing off, or 0, to force + ``db.query.text`` off) overrides the environment variable. + """ + + enabled: Optional[bool] + query_text_max_length: Optional[int] + + +_OTEL_ENABLED_ENV = "OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED" +_OTEL_QUERY_TEXT_MAX_LENGTH_ENV = "OTEL_PYTHON_INSTRUMENTATION_MONGODB_QUERY_TEXT_MAX_LENGTH" +_TRUTHY = frozenset({"1", "true", "yes"}) + +# Redaction compares command names case-insensitively, mirroring the +# normalization in command monitoring, so a differently-cased sensitive +# command still gets no span. +_SENSITIVE_COMMANDS_LOWER = frozenset(name.lower() for name in _SENSITIVE_COMMANDS) +_HELLO_COMMANDS_LOWER = frozenset(name.lower() for name in _HELLO_COMMANDS) + +# Fields redacted from the db.query.text attribute, mirroring the fields excluded +# from the equivalent CommandStartedEvent.command per the OpenTelemetry spec. +_QUERY_TEXT_EXCLUDED_FIELDS = frozenset({"lsid", "$db", "$clusterTime", "signature"}) + +# getMore's own command value is the cursor id, not the collection name; the +# collection lives under a separate "collection" key instead. +# See _gen_get_more_command in pymongo/message.py. +_GET_MORE = "getMore" + +# explain wraps the real command (e.g. find/aggregate) rather than naming a +# collection directly: {"explain": {"find": "coll", ...}}. See _Query.as_command +# in pymongo/message.py. +_EXPLAIN = "explain" + +# Commands against this database (e.g. user/role management, renameCollection) +# never have a real collection name, even when their command value is a string. +_ADMIN_DB = "admin" + +# Commands whose string command value names a user or role, not a collection +# (against any database): db.collection.name must be omitted rather than +# expose the username or role name as a collection. +_NOT_COLLECTION_COMMANDS = frozenset( + { + "createUser", + "dropAllRolesFromDatabase", + "dropAllUsersFromDatabase", + "dropRole", + "dropUser", + "grantPrivilegesToRole", + "grantRolesToRole", + "grantPrivilegesToUser", + "grantRolesToUser", + "invalidateUserCache", + "revokePrivilegesFromRole", + "revokePrivilegesFromUser", + "revokeRolesFromRole", + "revokeRolesFromUser", + "rolesInfo", + "createRole", + "updateRole", + "updateUser", + "usersInfo", + } +) + + +def _env_truthy(name: str) -> bool: + """Return True if the environment variable ``name`` is set to "1", "true", or "yes".""" + return os.getenv(name, "").strip().lower() in _TRUTHY + + +def _resolve_tracing_options(tracing: Optional[TracingOptions]) -> TracingOptions: + """Resolve a client's raw ``tracing`` option against the environment, once. + + Called once per ``MongoClient`` construction: an explicit value (including + ``False``, to force tracing off, or ``0``, to force ``db.query.text`` off) + wins over the environment variable; otherwise the environment variable + decides. When opentelemetry isn't installed, tracing is always disabled. + The resolved values are then consulted on every command without touching + the environment again. + """ + if not _HAS_OPENTELEMETRY: + return {"enabled": False, "query_text_max_length": 0} + enabled = tracing.get("enabled") if tracing is not None else None + if enabled is None: + enabled = _env_truthy(_OTEL_ENABLED_ENV) + max_length = tracing.get("query_text_max_length") if tracing is not None else None + if max_length is None: + try: + max_length = int(os.getenv(_OTEL_QUERY_TEXT_MAX_LENGTH_ENV, "0")) + except ValueError: + max_length = 0 + return {"enabled": bool(enabled), "query_text_max_length": max(0, max_length)} + + +def _is_tracing_enabled(tracing_options: Optional[TracingOptions]) -> bool: + """Return True if OTel command spans should be created for this client. + + ``tracing_options`` is the client's resolved ``tracing`` option (see + :func:`_resolve_tracing_options`); ``None`` means there is no client + context (connection handshakes, server monitoring), which is never traced. + This runs on every command and must be cheap: no environment lookups. + """ + if tracing_options is None: + return False + return bool(tracing_options.get("enabled")) + + +def _get_query_text_max_length(tracing_options: Optional[TracingOptions]) -> int: + """Return the resolved ``db.query.text`` truncation length, or 0 to omit the attribute.""" + if tracing_options is None: + return 0 + return tracing_options.get("query_text_max_length") or 0 + + +def _connection_attributes(conn: _ConnectionTelemetryInfo) -> dict[str, Any]: + """Return the connection-static span attributes as a fresh dict. + + The values are derived from the connection's address and ids, which do not + change during the life of a connection, so they are computed once and + cached on the connection (keyed by the server connection id, which changes + when the underlying socket is re-established, invalidating the cache). A + checked-out connection is owned by one caller at a time, so reading and + updating the cache needs no lock. + """ + server_connection_id = conn.server_connection_id + cached = getattr(conn, "_otel_connection_attributes", None) + if cached is not None and cached[0] == server_connection_id and cached[1]: + return cached[1].copy() + address = conn.address + attrs: dict[str, Any] = { + "db.system.name": "mongodb", + "server.address": address[0], + "network.transport": "unix" if address[1] is None else "tcp", + "db.mongodb.driver_connection_id": conn.id, + } + if address[1] is not None: + attrs["server.port"] = address[1] + if server_connection_id is not None: + attrs["db.mongodb.server_connection_id"] = server_connection_id + if cached is not None: + # Duck-typed fakes without the attribute (test doubles for the + # Protocol) skip caching rather than growing a new attribute. + conn._otel_connection_attributes = (server_connection_id, attrs) + return attrs.copy() + + +def _build_query_text(cmd: Mapping[str, Any], max_length: int) -> str: + """Serialize ``cmd`` to extended JSON, redacted and truncated to ``max_length``. + + Mirrors the truncation approach used for log messages: truncate field + values first, which usually keeps the result well-formed JSON (unlike a + blind cut of the fully-serialized string), then fall back to a hard + string cut as a safety net for whatever the field truncation's size + estimate still leaves over ``max_length``. The "..." marker is carved out + of the budget (not appended on top of it) so the result never exceeds + ``max_length``. + """ + filtered = {k: v for k, v in cmd.items() if k not in _QUERY_TEXT_EXCLUDED_FIELDS} + truncated_cmd = _truncate_documents(filtered, max_length)[0] + # default=repr mirrors the structured logger: tracing is best-effort and must + # not raise for commands containing custom/codec-managed Python types. + text = json_util.dumps(truncated_cmd, json_options=_JSON_OPTIONS, default=repr) + if len(text) > max_length: + # A budget smaller than the marker truncates without it so the result + # still never exceeds max_length. + suffix = "..." if max_length >= 3 else "" + text = text[: max_length - len(suffix)] + suffix + return text + + +def _extract_collection_name( + command_name: str, dbname: str, cmd: Mapping[str, Any] +) -> Optional[str]: + """Return the collection name targeted by ``cmd``, or None if it doesn't target one. + + Always None for commands against the admin database: several (e.g. dropUser, + renameCollection) carry a string command value that names a user, role, or + namespace rather than a collection. + """ + if dbname == _ADMIN_DB: + return None + if command_name in _NOT_COLLECTION_COMMANDS: + return None + if command_name == _EXPLAIN: + inner = cmd.get(_EXPLAIN) + if not isinstance(inner, Mapping) or not inner: + return None + inner_name = next(iter(inner)) + return _extract_collection_name(inner_name, dbname, inner) + key = "collection" if command_name == _GET_MORE else command_name + value = cmd.get(key) + return value if isinstance(value, str) else None + + +def _build_query_summary(command_name: str, dbname: str, collection: Optional[str]) -> str: + """Build the ``db.query.summary`` attribute value for a command.""" + if collection: + return f"{command_name} {dbname}.{collection}" + return f"{command_name} {dbname}" + + +def _is_sensitive_command(command_name: str, speculative_hello: bool) -> bool: + """Mirror the redaction rules in ``pymongo.logger.LogMessage._is_sensitive``.""" + name = command_name.lower() + if name in _SENSITIVE_COMMANDS_LOWER: + return True + return name in _HELLO_COMMANDS_LOWER and speculative_hello + + +def _format_lsid(lsid: Mapping[str, Any]) -> Optional[str]: + """Return the ``db.mongodb.lsid`` attribute value for a session id document.""" + id_value = lsid.get("id") + if id_value is None: + return None + try: + return str(id_value.as_uuid()) + except (AttributeError, ValueError): + return str(id_value) + + +def start_command_span( + tracing_options: Optional[TracingOptions], + conn: _ConnectionTelemetryInfo, + cmd: MutableMapping[str, Any], + dbname: str, + command_name: str, + speculative_hello: bool, +) -> Optional[Span]: + """Start and return a CLIENT-kind span for a server command, or None. + + Returns None when tracing is disabled/unavailable or the command is + sensitive (mirroring the redaction applied to logs). + + Only cheap attributes (constants and values already held by the + connection) are provided at span creation, so they are visible to the + sampler. The expensive ones are added only when the span is being + recorded, per the OpenTelemetry spec's performance guidelines, so + unsampled configurations pay little more than the span creation itself. + """ + if not _is_tracing_enabled(tracing_options): + return None + if _is_sensitive_command(command_name, speculative_hello): + return None + + collection = _extract_collection_name(command_name, dbname, cmd) + attributes = _connection_attributes(conn) + attributes["db.namespace"] = dbname + attributes["db.command.name"] = command_name + if collection: + attributes["db.collection.name"] = collection + + assert _TRACER is not None # _is_tracing_enabled already checked _HAS_OPENTELEMETRY + span = _TRACER.start_span(command_name, kind=_SPAN_KIND_CLIENT, attributes=attributes) + if not span.is_recording(): + # The span was dropped by the sampler (or tracing is a no-op): the + # spec says no further attributes are added to it. + return span + + # Expensive attributes, added after the sampling decision; they are not + # visible to samplers. + span.set_attribute("db.query.summary", _build_query_summary(command_name, dbname, collection)) + lsid = cmd.get("lsid") + if isinstance(lsid, Mapping): + formatted_lsid = _format_lsid(lsid) + if formatted_lsid is not None: + span.set_attribute("db.mongodb.lsid", formatted_lsid) + txn_number = cmd.get("txnNumber") + if txn_number is not None: + span.set_attribute("db.mongodb.txn_number", txn_number) + max_query_text_length = _get_query_text_max_length(tracing_options) + if max_query_text_length > 0: + span.set_attribute("db.query.text", _build_query_text(cmd, max_query_text_length)) + return span + + +def end_command_span_success( + span: Optional[Span], + cmd: Mapping[str, Any], + command_name: str, + reply: _DocumentOut, +) -> None: + """Set the cursor id (if any) and end the span. + + The ``db.mongodb.cursor_id`` rules (spec: "db.mongodb.cursor_id"): a + non-zero reply cursor id is recorded as-is; a zero reply id means the + cursor is exhausted, so the attribute is omitted unless the command + operated on an existing cursor (getMore), in which case the cursor id the + command sent is recorded; a literal zero is never recorded. + """ + if span is None: + return + cursor = reply.get("cursor") + if isinstance(cursor, Mapping) and "id" in cursor: + cursor_id = cursor["id"] + if cursor_id: + span.set_attribute("db.mongodb.cursor_id", cursor_id) + elif command_name == _GET_MORE: + sent_id = cmd.get(_GET_MORE) + if sent_id: + span.set_attribute("db.mongodb.cursor_id", sent_id) + span.end() + + +def end_command_span_failure( + span: Optional[Span], + failure: _DocumentOut, + exc: BaseException, +) -> None: + """Record the exception, set the error status, and end the span. + + ``error.type`` (OpenTelemetry convention): the server response code when + the failure came from a server response, otherwise the exception class + name, so server and transport failures are distinguishable. + """ + if span is None: + return + span.record_exception(exc) + code = failure.get("code") + if code is not None: + code = str(code) + span.set_attribute("db.response.status_code", code) + span.set_attribute("error.type", code) + else: + span.set_attribute("error.type", type(exc).__name__) + span.set_status(Status(StatusCode.ERROR, description=failure.get("errmsg"))) + span.end() + + +class _NoOperationSpan: + """No-op context manager used by :func:`operation_span` when tracing is off. + + A module-level singleton: the operation funnels enter it unconditionally, + so the disabled-tracing path pays only the ``with`` statement itself. + """ + + __slots__ = () + + def __enter__(self) -> None: + return None + + def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None: + return None + + +_NO_OPERATION_SPAN = _NoOperationSpan() + + +class _OperationSpanContext: + """Context manager for one operation span. + + Makes the span current while the operation runs, so the operation's + command spans (created with ``Tracer.start_span``, which parents from the + current context) nest beneath it. Ends the span on exit; on exception, + records the exception and sets the spec's ``error.type`` attribute first. + """ + + __slots__ = ("_span", "_token") + + def __init__(self, span: Span) -> None: + self._span = span + self._token: Optional[object] = None + + def __enter__(self) -> Span: + self._token = otel_context.attach(trace.set_span_in_context(self._span)) + return self._span + + def __exit__(self, exc_type: Any, exc_value: Any, traceback: Any) -> None: + otel_context.detach(self._token) + span = self._span + if exc_value is not None: + span.record_exception(exc_value) + # The spec's operation-span error.type: the name of the exception + # class raised to the application (a failed command does not carry + # its server code over: a retry of the same operation may succeed). + span.set_attribute("error.type", type(exc_value).__name__) + span.set_status(Status(StatusCode.ERROR, description=str(exc_value))) + span.end() + + +def operation_span( + tracing_options: Optional[TracingOptions], + operation_name: str, + dbname: Optional[str], + collection: Optional[str], +) -> _OperationSpanContext | _NoOperationSpan: + """Start an operation span and return a context manager that ends it. + + Per the spec ("Instrumenting Driver Operations"): one CLIENT-kind span per + driver operation, started before server selection so its duration reflects + all of the driver's activities, named ``operation_name db.collection_name`` + (or ``operation_name db`` when there is no collection) and carrying the + spec's required attributes. Returns the :data:`_NO_OPERATION_SPAN` + singleton when tracing is disabled so the disabled path stays at parity. + """ + if not _is_tracing_enabled(tracing_options): + return _NO_OPERATION_SPAN + if collection: + summary = f"{operation_name} {dbname}.{collection}" + else: + summary = f"{operation_name} {dbname}" + attributes: dict[str, Any] = { + "db.system.name": "mongodb", + "db.operation.name": operation_name, + "db.operation.summary": summary, + } + if dbname: + attributes["db.namespace"] = dbname + if collection: + attributes["db.collection.name"] = collection + assert _TRACER is not None # _is_tracing_enabled already checked _HAS_OPENTELEMETRY + return _OperationSpanContext( + _TRACER.start_span(summary, kind=_SPAN_KIND_CLIENT, attributes=attributes) + ) diff --git a/pymongo/_telemetry.py b/pymongo/_telemetry.py index b8c8fa00ff..2295f3bd80 100644 --- a/pymongo/_telemetry.py +++ b/pymongo/_telemetry.py @@ -22,7 +22,8 @@ from collections.abc import MutableMapping from typing import TYPE_CHECKING, Any, Optional -from pymongo import _op_id +from pymongo import _op_id, _otel +from pymongo.errors import OperationFailure from pymongo.logger import ( _COMMAND_LOGGER, _CONNECTION_LOGGER, @@ -70,6 +71,21 @@ def _generate_op_id_or_none(listeners: Optional[_EventListeners]) -> Optional[in ) +def operation_span( + tracing_options: Optional[_otel.TracingOptions], + operation_name: str, + dbname: Optional[str], + collection: Optional[str], +) -> Any: + """Start the spec's operation span for one driver operation, as a context manager. + + Thin wrapper over :func:`pymongo._otel.operation_span`, kept here so + callers already importing :mod:`pymongo._telemetry` need no new import + edges. See the ``_otel`` function for the span's shape. + """ + return _otel.operation_span(tracing_options, operation_name, dbname, collection) + + class _CommandTelemetry: """Combines structured logging and APM event publishing for a single command. @@ -94,8 +110,12 @@ class _CommandTelemetry: "_publish", "_request_id", "_should_log", + "_span", + "_speculative_hello", "_start", "_topology_id", + "_tracing_enabled", + "_tracing_options", ) def __init__( @@ -107,13 +127,18 @@ def __init__( dbname: str, request_id: int, op_id: Optional[int], + tracing_options: Optional[_otel.TracingOptions] = None, + speculative_hello: bool = False, 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 _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._tracing_options = tracing_options + self._tracing_enabled = _otel._is_tracing_enabled(tracing_options) + self._span: Optional[Any] = None + self._active = self._should_log or self._publish or self._tracing_enabled self._start = 0.0 self._duration_s = 0.0 if not self._active: @@ -126,6 +151,7 @@ def __init__( self._dbname = dbname self._request_id = request_id self._op_id = op_id if op_id is not None else _op_id.OP_ID.get() + self._speculative_hello = speculative_hello def _emit_log(self, message: _CommandStatusMessage, **extra: Any) -> None: _debug_log( @@ -145,7 +171,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.""" + """Emit the STARTED log entry and APM event, start the span, and start the duration clock.""" self._start = time.monotonic() if not self._active: return @@ -164,6 +190,15 @@ def started(self, orig: MutableMapping[str, Any], ensure_db: bool) -> None: self._op_id, service_id=self._conn.service_id, ) + if self._tracing_enabled: + self._span = _otel.start_command_span( + self._tracing_options, + self._conn, + orig, + self._dbname, + self._name, + self._speculative_hello, + ) @property def duration_s(self) -> float: @@ -176,64 +211,74 @@ def succeeded( command_name: str, speculative_hello: bool, ) -> None: - """Emit the SUCCEEDED log entry and APM event.""" + """Emit the SUCCEEDED log entry and APM event, and end the span.""" 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=duration, - reply=reply, - speculative_authenticate=speculative_hello, - ) - if self._publish: - assert self._listeners is not None - self._listeners.publish_command_success( - duration, - reply, - command_name, - self._request_id, - self._conn.address, - self._conn.server_connection_id, - self._op_id, - service_id=self._conn.service_id, - speculative_hello=speculative_hello, - database_name=self._dbname, - ) + if self._should_log or self._publish: + # Only logging and APM events format the duration; a tracing-only + # client doesn't pay for the timedelta. + duration = datetime.timedelta(seconds=self._duration_s) + if self._should_log: + self._emit_log( + _CommandStatusMessage.SUCCEEDED, + durationMS=duration, + reply=reply, + speculative_authenticate=speculative_hello, + ) + if self._publish: + assert self._listeners is not None + self._listeners.publish_command_success( + duration, + reply, + command_name, + self._request_id, + self._conn.address, + self._conn.server_connection_id, + self._op_id, + service_id=self._conn.service_id, + speculative_hello=speculative_hello, + database_name=self._dbname, + ) + if self._span is not None: + _otel.end_command_span_success(self._span, self._cmd, self._name, reply) def failed( self, failure: _DocumentOut, command_name: str, - is_server_side_error: bool, + exc: BaseException, ) -> None: - """Emit the FAILED log entry and APM event.""" + """Emit the FAILED log entry and APM event, and end the span.""" 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=duration, - failure=failure, - isServerSideError=is_server_side_error, - ) - if self._publish: - assert self._listeners is not None - self._listeners.publish_command_failure( - duration, - failure, - command_name, - self._request_id, - self._conn.address, - self._conn.server_connection_id, - self._op_id, - service_id=self._conn.service_id, - database_name=self._dbname, - ) + if self._should_log or self._publish: + # Only logging and APM events format the duration; a tracing-only + # client doesn't pay for the timedelta. + duration = datetime.timedelta(seconds=self._duration_s) + if self._should_log: + self._emit_log( + _CommandStatusMessage.FAILED, + durationMS=duration, + failure=failure, + isServerSideError=isinstance(exc, OperationFailure), + ) + if self._publish: + assert self._listeners is not None + self._listeners.publish_command_failure( + duration, + failure, + command_name, + self._request_id, + self._conn.address, + self._conn.server_connection_id, + self._op_id, + service_id=self._conn.service_id, + database_name=self._dbname, + ) + if self._span is not None: + _otel.end_command_span_failure(self._span, failure, exc) class _CmapTelemetry: diff --git a/pymongo/asynchronous/bulk.py b/pymongo/asynchronous/bulk.py index 81186499f6..436df3902c 100644 --- a/pymongo/asynchronous/bulk.py +++ b/pymongo/asynchronous/bulk.py @@ -312,6 +312,8 @@ async def retryable_bulk( retryable_bulk, session, operation, + db_name=self.collection.database.name, + coll_name=self.collection.name, bulk=self, # type: ignore[arg-type] operation_id=op_id, ) diff --git a/pymongo/asynchronous/client_bulk.py b/pymongo/asynchronous/client_bulk.py index bfbe9e2c6b..4340046de9 100644 --- a/pymongo/asynchronous/client_bulk.py +++ b/pymongo/asynchronous/client_bulk.py @@ -404,6 +404,9 @@ async def retryable_bulk( retryable_bulk, session, operation, + # The spec's client bulkWrite spans run against admin with no + # single collection. + db_name="admin", bulk=self, operation_id=op_id, ) diff --git a/pymongo/asynchronous/collection.py b/pymongo/asynchronous/collection.py index 924635d7dc..9de95fe6b2 100644 --- a/pymongo/asynchronous/collection.py +++ b/pymongo/asynchronous/collection.py @@ -650,7 +650,9 @@ async def inner( session=session, ) - await self.database.client._retryable_write(False, inner, session, _Op.CREATE) + await self.database.client._retryable_write( + False, inner, session, _Op.CREATE, db_name=self._database.name, coll_name=self._name + ) async def _create( self, @@ -813,7 +815,12 @@ async def _insert_command( _check_write_command_response(result) await self._database.client._retryable_write( - acknowledged, _insert_command, session, operation=_Op.INSERT + acknowledged, + _insert_command, + session, + operation=_Op.INSERT, + db_name=self._database.name, + coll_name=self._name, ) if not isinstance(doc, RawBSONDocument): @@ -1100,6 +1107,8 @@ async def _update( _update, session, operation, + db_name=self._database.name, + coll_name=self._name, ) async def replace_one( @@ -1562,6 +1571,8 @@ async def _delete( _delete, session, operation=_Op.DELETE, + db_name=self._database.name, + coll_name=self._name, ) async def delete_one( @@ -2251,7 +2262,12 @@ def gen_indexes() -> Iterator[Mapping[str, Any]]: return names return await self.database.client._retryable_write( - False, inner, session, _Op.CREATE_INDEXES + False, + inner, + session, + _Op.CREATE_INDEXES, + db_name=self._database.name, + coll_name=self._name, ) async def create_index( @@ -2485,7 +2501,14 @@ async def inner( session=session, ) - await self.database.client._retryable_write(False, inner, session, _Op.DROP_INDEXES) + await self.database.client._retryable_write( + False, + inner, + session, + _Op.DROP_INDEXES, + db_name=self._database.name, + coll_name=self._name, + ) async def list_indexes( self, @@ -2763,7 +2786,12 @@ async def inner( return [index["name"] for index in resp["indexesCreated"]] return await self.database.client._retryable_write( - False, inner, session, _Op.CREATE_SEARCH_INDEXES + False, + inner, + session, + _Op.CREATE_SEARCH_INDEXES, + db_name=self._database.name, + coll_name=self._name, ) async def drop_search_index( @@ -2804,7 +2832,14 @@ async def inner( session=session, ) - await self.database.client._retryable_write(False, inner, session, _Op.DROP_SEARCH_INDEXES) + await self.database.client._retryable_write( + False, + inner, + session, + _Op.DROP_SEARCH_INDEXES, + db_name=self._database.name, + coll_name=self._name, + ) async def update_search_index( self, @@ -2846,7 +2881,14 @@ async def inner( session=session, ) - await self.database.client._retryable_write(False, inner, session, _Op.UPDATE_SEARCH_INDEX) + await self.database.client._retryable_write( + False, + inner, + session, + _Op.UPDATE_SEARCH_INDEX, + db_name=self._database.name, + coll_name=self._name, + ) async def options( self, @@ -3142,7 +3184,9 @@ async def inner( client=client, ) - return await client._retryable_write(False, inner, session, _Op.RENAME) + return await client._retryable_write( + False, inner, session, _Op.RENAME, db_name=self._database.name, coll_name=self._name + ) async def distinct( self, diff --git a/pymongo/asynchronous/command_runner.py b/pymongo/asynchronous/command_runner.py index aa0839fa4a..9e94f9b0b7 100644 --- a/pymongo/asynchronous/command_runner.py +++ b/pymongo/asynchronous/command_runner.py @@ -48,7 +48,7 @@ ) from bson import _decode_all_selective -from pymongo import _csot, helpers_shared, message +from pymongo import _csot, _otel, helpers_shared, message from pymongo._telemetry import _CommandTelemetry from pymongo.compression_support import _NO_COMPRESSION from pymongo.errors import NotPrimaryError, OperationFailure @@ -163,14 +163,28 @@ async def _run_command( if orig is None: orig = cmd - # Fast path: skip telemetry construction when logging and APM are disabled - # Inline enabled check here for performance + # Fast path: skip telemetry construction when logging, APM, and tracing are all + # disabled. Inline enabled check here for performance; it must stay in sync with + # _CommandTelemetry.__init__'s own gate, including its tracing term, or a + # tracing-only client would build no telemetry and so emit no command spans. + tracing_options = client.options.tracing if client is not None else None 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 + if ( + (topology_id is not None and _is_debug_enabled(_COMMAND_LOGGER)) + or (listeners is not None and listeners.enabled_for_commands) + or _otel._is_tracing_enabled(tracing_options) ): telemetry = _CommandTelemetry( - topology_id, conn, listeners, cmd, dbname, request_id, op_id, name=name + topology_id, + conn, + listeners, + cmd, + dbname, + request_id, + op_id, + tracing_options=tracing_options, + speculative_hello=speculative_hello, + name=name, ) telemetry.started(orig, ensure_db) start = 0.0 @@ -218,13 +232,16 @@ async def _run_command( parse_write_concern_error=parse_write_concern_error, pool_opts=pool_opts, ) - except Exception as exc: + except BaseException as exc: + # CancelledError (a BaseException) must also end the span: task + # cancellation lands here mid-command, and the span is ended with an + # error status before the cancellation propagates unmasked. if isinstance(exc, (NotPrimaryError, OperationFailure)): failure: _DocumentOut = exc.details # type: ignore[assignment] else: failure = _convert_exception(exc) if telemetry is not None: - telemetry.failed(failure, command_name, isinstance(exc, OperationFailure)) + telemetry.failed(failure, command_name, exc) raise if telemetry is not None: diff --git a/pymongo/asynchronous/database.py b/pymongo/asynchronous/database.py index 88ca9cab5b..f9455f1a73 100644 --- a/pymongo/asynchronous/database.py +++ b/pymongo/asynchronous/database.py @@ -1268,7 +1268,9 @@ async def inner( session=session, ) - return await self.client._retryable_write(False, inner, session, _Op.DROP) + return await self.client._retryable_write( + False, inner, session, _Op.DROP, db_name=self.name + ) @_csot.apply async def drop_collection( diff --git a/pymongo/asynchronous/mongo_client.py b/pymongo/asynchronous/mongo_client.py index df390a218e..d423ca920a 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 _generate_op_id_or_none, log_command_retry +from pymongo._telemetry import _generate_op_id_or_none, log_command_retry, operation_span from pymongo.asynchronous import client_session, database, uri_parser from pymongo.asynchronous.change_stream import AsyncChangeStream, AsyncClusterChangeStream from pymongo.asynchronous.client_bulk import _AsyncClientBulk @@ -662,6 +662,23 @@ def validator(host: str) -> bool: If enabled, server overload errors will cause retry attempts to select a server that has not yet returned an overload error, if possible. Defaults to ``False``. + | **OpenTelemetry options:** + | (Requires the ``opentelemetry-api`` package; install with the ``pymongo[opentelemetry]`` extra.) + + - `tracing`: (dict) Configuration for OpenTelemetry command spans, with keys: + + - ``enabled``: (boolean) Whether to create spans for server commands issued by + this client. Unset by default, which defers to the + ``OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED`` environment variable. When set + explicitly, including to ``False``, it overrides the environment variable, so + an application can reliably opt out. + - ``query_text_max_length``: (int) The maximum length of the ``db.query.text`` + span attribute. Unset by default, which defers to the + ``OTEL_PYTHON_INSTRUMENTATION_MONGODB_QUERY_TEXT_MAX_LENGTH`` environment + variable (itself defaulting to ``0``, which omits the attribute). Setting + this explicitly, including to ``0``, always overrides the environment + variable. + .. seealso:: The MongoDB documentation on `connections `_. .. versionchanged:: 4.19 @@ -670,6 +687,9 @@ def validator(host: str) -> bool: .. versionchanged:: 4.18 Added the ``srvAllowedHostsSuffix`` URI and keyword argument. + .. versionchanged:: 4.18 + Added the ``tracing`` keyword argument. + .. versionchanged:: 4.17 Added the ``max_adaptive_retries`` and ``enable_overload_retargeting`` URI and keyword arguments. @@ -1035,6 +1055,10 @@ async def _resolve_srv(self) -> None: # Username and password passed as kwargs override user info in URI. username = opts.get("username", self._resolve_srv_info["username"]) password = opts.get("password", self._resolve_srv_info["password"]) + # The tracing option was resolved against the environment when the + # client was constructed; carry that snapshot into the rebuilt + # options rather than re-reading the (possibly changed) environment. + opts["tracing"] = self._options.tracing self._options = ClientOptions( username, password, self._resolve_srv_info["dbase"], opts, _IS_SYNC ) @@ -1968,43 +1992,51 @@ async def _run_operation( :param address: Optional address when sending a message to a specific server, used for getMore. """ - if operation.conn_mgr: - server = await self._select_server( + # The spec's operation span: one span per driver operation, started + # before server selection, with the operation's command spans nested + # beneath it. getMores are cursor territory (PYTHON-5947): no + # operation span yet. + # getMores are cursor territory (PYTHON-5947): no operation span yet. + tracing = self.options.tracing if not isinstance(operation, _GetMore) else None + op_span = operation_span(tracing, operation.name, operation.db, operation.coll) + with op_span: + if operation.conn_mgr: + server = await self._select_server( + operation.read_preference, + operation.session, # type: ignore[arg-type] + operation.name, + address=address, + ) + + async with operation.conn_mgr._lock: + async with _ClientCheckout.for_existing_conn( + self, + server, + operation.session, # type: ignore[arg-type] + operation.conn_mgr.conn, + ): + return await run_with_conn( + operation.conn_mgr.conn, operation, operation.read_preference + ) + + async def _cmd( + _session: Optional[AsyncClientSession], + _server: Server, + conn: AsyncConnection, + read_preference: _ServerMode, + ) -> Response: + operation.reset() # Reset op in case of retry. + return await run_with_conn(conn, operation, read_preference) + + return await self._retryable_read( + _cmd, operation.read_preference, operation.session, # type: ignore[arg-type] - operation.name, address=address, + retryable=isinstance(operation, _Query), + operation=operation.name, ) - async with operation.conn_mgr._lock: - async with _ClientCheckout.for_existing_conn( - self, - server, - operation.session, # type: ignore[arg-type] - operation.conn_mgr.conn, - ): - return await run_with_conn( - operation.conn_mgr.conn, operation, operation.read_preference - ) - - async def _cmd( - _session: Optional[AsyncClientSession], - _server: Server, - conn: AsyncConnection, - read_preference: _ServerMode, - ) -> Response: - operation.reset() # Reset op in case of retry. - return await run_with_conn(conn, operation, read_preference) - - return await self._retryable_read( - _cmd, - operation.read_preference, - operation.session, # type: ignore[arg-type] - address=address, - retryable=isinstance(operation, _Query), - operation=operation.name, - ) - async def _retry_with_session( self, retryable: bool, @@ -2140,6 +2172,8 @@ async def _retryable_write( operation: str, bulk: Optional[Union[_AsyncBulk, _AsyncClientBulk]] = None, operation_id: Optional[int] = None, + db_name: Optional[str] = None, + coll_name: Optional[str] = None, ) -> T: """Execute an operation with consecutive retries if possible @@ -2154,9 +2188,21 @@ async def _retryable_write( :param operation: The name of the operation that the server is being selected for :param bulk: bulk abstraction to execute operations in bulk, defaults to None :param operation_id: Stable operation id shared across retries, defaults to None + :param db_name: The database the operation targets, for the spec's + operation span. Defaults to None (attribute omitted). + :param coll_name: The collection the operation targets, for the spec's + operation span. Defaults to None (attribute omitted). """ async with self._tmp_session(session) as s: - return await self._retry_with_session(retryable, func, s, bulk, operation, operation_id) + with operation_span( + self.options.tracing, + getattr(operation, "value", operation), + db_name, + coll_name, + ): + return await self._retry_with_session( + retryable, func, s, bulk, operation, operation_id + ) def _cleanup_cursor_no_lock( self, diff --git a/pymongo/asynchronous/pool.py b/pymongo/asynchronous/pool.py index b7f7a1f6a0..f7e408acfc 100644 --- a/pymongo/asynchronous/pool.py +++ b/pymongo/asynchronous/pool.py @@ -170,6 +170,9 @@ def __init__( # For load balancer support. self.service_id: Optional[ObjectId] = None self.server_connection_id: Optional[int] = None + # Cache of connection-static span attributes for OpenTelemetry + # command spans; see pymongo._otel._connection_attributes. + self._otel_connection_attributes: tuple[Optional[int], dict[str, Any]] = (None, {}) # When executing a transaction in load balancing mode, this flag is # set to true to indicate that the session now owns the connection. self.pinned_txn = False diff --git a/pymongo/client_options.py b/pymongo/client_options.py index 99a2bcb4e4..1e505a58b4 100644 --- a/pymongo/client_options.py +++ b/pymongo/client_options.py @@ -23,7 +23,7 @@ from typing import TYPE_CHECKING, Any, Optional, cast from bson.codec_options import _parse_codec_options -from pymongo import common +from pymongo import _otel, common from pymongo.compression_support import CompressionSettings from pymongo.errors import ConfigurationError from pymongo.monitoring import _EventListener, _EventListeners @@ -247,6 +247,10 @@ def __init__( if "enable_overload_retargeting" in options else options.get("enableoverloadretargeting", common.ENABLE_OVERLOAD_RETARGETING) ) + # Resolve the tracing option against the environment variables once: + # tracing cannot be changed after the client is constructed, so no + # command needs to consult the environment again. + self.__tracing = _otel._resolve_tracing_options(options.get("tracing")) @property def _options(self) -> Mapping[str, Any]: @@ -374,3 +378,16 @@ def enable_overload_retargeting(self) -> bool: .. versionadded:: 4.17 """ return self.__enable_overload_retargeting + + @property + def tracing(self) -> _otel.TracingOptions: + """The configured ``tracing`` option for OpenTelemetry command spans. + + The values are resolved when the client is constructed: an explicit + value wins over the ``OTEL_PYTHON_INSTRUMENTATION_*`` environment + variables, and the environment variables decide when the client + didn't configure the value. + + .. versionadded:: 4.18 + """ + return self.__tracing diff --git a/pymongo/common.py b/pymongo/common.py index 9fa3bb15db..ebab02f47c 100644 --- a/pymongo/common.py +++ b/pymongo/common.py @@ -51,6 +51,7 @@ from pymongo.write_concern import DEFAULT_WRITE_CONCERN, WriteConcern, validate_boolean if TYPE_CHECKING: + from pymongo import _otel from pymongo.typings import _AgnosticClientSession @@ -610,6 +611,31 @@ def validate_server_api_or_none(option: Any, value: Any) -> Optional[ServerApi]: return value +def validate_tracing_or_none(option: str, value: Any) -> Optional[_otel.TracingOptions]: + """Validate the tracing keyword arg.""" + if value is None: + return value + validate_is_mapping(option, value) + unknown = set(value) - {"enabled", "query_text_max_length"} + if unknown: + raise ConfigurationError(f"Unknown tracing option(s): {sorted(unknown)}") + enabled = value.get("enabled") + if enabled is not None: + validate_boolean("tracing.enabled", enabled) + query_text_max_length = value.get("query_text_max_length") + if query_text_max_length is not None: + # bool is a subclass of int; reject it explicitly rather than silently + # treating True/False as 1/0. + if isinstance(query_text_max_length, bool): + raise TypeError("tracing.query_text_max_length must be an integer, not a boolean") + # Keep the validator's converted value: it accepts numeric strings and + # the stored option must be an int. + query_text_max_length = validate_non_negative_integer( + "tracing.query_text_max_length", query_text_max_length + ) + return {"enabled": enabled, "query_text_max_length": query_text_max_length} + + def validate_is_callable_or_none(option: Any, value: Any) -> Optional[Callable[..., Any]]: """Validates that 'value' is a callable.""" if value is None: @@ -789,6 +815,7 @@ def validate_server_monitoring_mode(option: str, value: str) -> str: "authoidcallowedhosts": validate_list, "max_adaptive_retries": validate_non_negative_integer, "enable_overload_retargeting": validate_boolean_or_string, + "tracing": validate_tracing_or_none, } # Dictionary where keys are any URI option name, and values are the diff --git a/pymongo/message.py b/pymongo/message.py index b4b3e485d1..7ca1a7583c 100644 --- a/pymongo/message.py +++ b/pymongo/message.py @@ -127,8 +127,12 @@ def _maybe_add_read_preference( return spec -def _convert_exception(exception: Exception) -> dict[str, Any]: - """Convert an Exception into a failure document for publishing.""" +def _convert_exception(exception: BaseException) -> dict[str, Any]: + """Convert an Exception into a failure document for publishing. + + Accepts BaseException so task cancellation (CancelledError) can be + recorded on a command span's failure path. + """ return {"errmsg": str(exception), "errtype": exception.__class__.__name__} diff --git a/pymongo/pool_shared.py b/pymongo/pool_shared.py index 04663f6fc9..69eb28d7f2 100644 --- a/pymongo/pool_shared.py +++ b/pymongo/pool_shared.py @@ -58,8 +58,12 @@ class _ConnectionTelemetryInfo(Protocol): id: int server_connection_id: Optional[int] - address: tuple[str, int] + address: _Address service_id: Optional[ObjectId] + # Cache of (server_connection_id, connection-static span attributes) + # consulted by pymongo._otel._connection_attributes; (None, {}) until the + # first span is created. + _otel_connection_attributes: tuple[Optional[int], dict[str, Any]] def _get_ssl_session(ssl_sock: Any) -> Optional[Any]: diff --git a/pymongo/synchronous/bulk.py b/pymongo/synchronous/bulk.py index 291965fca1..5a71879750 100644 --- a/pymongo/synchronous/bulk.py +++ b/pymongo/synchronous/bulk.py @@ -312,6 +312,8 @@ def retryable_bulk( retryable_bulk, session, operation, + db_name=self.collection.database.name, + coll_name=self.collection.name, bulk=self, # type: ignore[arg-type] operation_id=op_id, ) diff --git a/pymongo/synchronous/client_bulk.py b/pymongo/synchronous/client_bulk.py index 549c9d59ca..546474f9ea 100644 --- a/pymongo/synchronous/client_bulk.py +++ b/pymongo/synchronous/client_bulk.py @@ -402,6 +402,9 @@ def retryable_bulk( retryable_bulk, session, operation, + # The spec's client bulkWrite spans run against admin with no + # single collection. + db_name="admin", bulk=self, operation_id=op_id, ) diff --git a/pymongo/synchronous/collection.py b/pymongo/synchronous/collection.py index 931001e179..1984a34d44 100644 --- a/pymongo/synchronous/collection.py +++ b/pymongo/synchronous/collection.py @@ -652,7 +652,9 @@ def inner( session=session, ) - self.database.client._retryable_write(False, inner, session, _Op.CREATE) + self.database.client._retryable_write( + False, inner, session, _Op.CREATE, db_name=self._database.name, coll_name=self._name + ) def _create( self, @@ -813,7 +815,12 @@ def _insert_command( _check_write_command_response(result) self._database.client._retryable_write( - acknowledged, _insert_command, session, operation=_Op.INSERT + acknowledged, + _insert_command, + session, + operation=_Op.INSERT, + db_name=self._database.name, + coll_name=self._name, ) if not isinstance(doc, RawBSONDocument): @@ -1100,6 +1107,8 @@ def _update( _update, session, operation, + db_name=self._database.name, + coll_name=self._name, ) def replace_one( @@ -1562,6 +1571,8 @@ def _delete( _delete, session, operation=_Op.DELETE, + db_name=self._database.name, + coll_name=self._name, ) def delete_one( @@ -2248,7 +2259,14 @@ def gen_indexes() -> Iterator[Mapping[str, Any]]: ) return names - return self.database.client._retryable_write(False, inner, session, _Op.CREATE_INDEXES) + return self.database.client._retryable_write( + False, + inner, + session, + _Op.CREATE_INDEXES, + db_name=self._database.name, + coll_name=self._name, + ) def create_index( self, @@ -2481,7 +2499,14 @@ def inner( session=session, ) - self.database.client._retryable_write(False, inner, session, _Op.DROP_INDEXES) + self.database.client._retryable_write( + False, + inner, + session, + _Op.DROP_INDEXES, + db_name=self._database.name, + coll_name=self._name, + ) def list_indexes( self, @@ -2759,7 +2784,12 @@ def inner( return [index["name"] for index in resp["indexesCreated"]] return self.database.client._retryable_write( - False, inner, session, _Op.CREATE_SEARCH_INDEXES + False, + inner, + session, + _Op.CREATE_SEARCH_INDEXES, + db_name=self._database.name, + coll_name=self._name, ) def drop_search_index( @@ -2800,7 +2830,14 @@ def inner( session=session, ) - self.database.client._retryable_write(False, inner, session, _Op.DROP_SEARCH_INDEXES) + self.database.client._retryable_write( + False, + inner, + session, + _Op.DROP_SEARCH_INDEXES, + db_name=self._database.name, + coll_name=self._name, + ) def update_search_index( self, @@ -2842,7 +2879,14 @@ def inner( session=session, ) - self.database.client._retryable_write(False, inner, session, _Op.UPDATE_SEARCH_INDEX) + self.database.client._retryable_write( + False, + inner, + session, + _Op.UPDATE_SEARCH_INDEX, + db_name=self._database.name, + coll_name=self._name, + ) def options( self, @@ -3136,7 +3180,9 @@ def inner( client=client, ) - return client._retryable_write(False, inner, session, _Op.RENAME) + return client._retryable_write( + False, inner, session, _Op.RENAME, db_name=self._database.name, coll_name=self._name + ) def distinct( self, diff --git a/pymongo/synchronous/command_runner.py b/pymongo/synchronous/command_runner.py index 21da51e2fc..88d66fa78a 100644 --- a/pymongo/synchronous/command_runner.py +++ b/pymongo/synchronous/command_runner.py @@ -48,7 +48,7 @@ ) from bson import _decode_all_selective -from pymongo import _csot, helpers_shared, message +from pymongo import _csot, _otel, helpers_shared, message from pymongo._telemetry import _CommandTelemetry from pymongo.compression_support import _NO_COMPRESSION from pymongo.errors import NotPrimaryError, OperationFailure @@ -163,14 +163,28 @@ def _run_command( if orig is None: orig = cmd - # Fast path: skip telemetry construction when logging and APM are disabled - # Inline enabled check here for performance + # Fast path: skip telemetry construction when logging, APM, and tracing are all + # disabled. Inline enabled check here for performance; it must stay in sync with + # _CommandTelemetry.__init__'s own gate, including its tracing term, or a + # tracing-only client would build no telemetry and so emit no command spans. + tracing_options = client.options.tracing if client is not None else None 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 + if ( + (topology_id is not None and _is_debug_enabled(_COMMAND_LOGGER)) + or (listeners is not None and listeners.enabled_for_commands) + or _otel._is_tracing_enabled(tracing_options) ): telemetry = _CommandTelemetry( - topology_id, conn, listeners, cmd, dbname, request_id, op_id, name=name + topology_id, + conn, + listeners, + cmd, + dbname, + request_id, + op_id, + tracing_options=tracing_options, + speculative_hello=speculative_hello, + name=name, ) telemetry.started(orig, ensure_db) start = 0.0 @@ -218,13 +232,16 @@ def _run_command( parse_write_concern_error=parse_write_concern_error, pool_opts=pool_opts, ) - except Exception as exc: + except BaseException as exc: + # CancelledError (a BaseException) must also end the span: task + # cancellation lands here mid-command, and the span is ended with an + # error status before the cancellation propagates unmasked. if isinstance(exc, (NotPrimaryError, OperationFailure)): failure: _DocumentOut = exc.details # type: ignore[assignment] else: failure = _convert_exception(exc) if telemetry is not None: - telemetry.failed(failure, command_name, isinstance(exc, OperationFailure)) + telemetry.failed(failure, command_name, exc) raise if telemetry is not None: diff --git a/pymongo/synchronous/database.py b/pymongo/synchronous/database.py index 65c5d2a7f0..aa4da7452f 100644 --- a/pymongo/synchronous/database.py +++ b/pymongo/synchronous/database.py @@ -1265,7 +1265,7 @@ def inner( session=session, ) - return self.client._retryable_write(False, inner, session, _Op.DROP) + return self.client._retryable_write(False, inner, session, _Op.DROP, db_name=self.name) @_csot.apply def drop_collection( diff --git a/pymongo/synchronous/mongo_client.py b/pymongo/synchronous/mongo_client.py index 6aaaaed20a..6dcc20cf8c 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 _generate_op_id_or_none, log_command_retry +from pymongo._telemetry import _generate_op_id_or_none, log_command_retry, operation_span from pymongo.client_options import ClientOptions from pymongo.client_session_shared import SessionOptions, TransactionOptions, _EmptyServerSession from pymongo.driver_info import DriverInfo @@ -663,6 +663,23 @@ def validator(host: str) -> bool: If enabled, server overload errors will cause retry attempts to select a server that has not yet returned an overload error, if possible. Defaults to ``False``. + | **OpenTelemetry options:** + | (Requires the ``opentelemetry-api`` package; install with the ``pymongo[opentelemetry]`` extra.) + + - `tracing`: (dict) Configuration for OpenTelemetry command spans, with keys: + + - ``enabled``: (boolean) Whether to create spans for server commands issued by + this client. Unset by default, which defers to the + ``OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED`` environment variable. When set + explicitly, including to ``False``, it overrides the environment variable, so + an application can reliably opt out. + - ``query_text_max_length``: (int) The maximum length of the ``db.query.text`` + span attribute. Unset by default, which defers to the + ``OTEL_PYTHON_INSTRUMENTATION_MONGODB_QUERY_TEXT_MAX_LENGTH`` environment + variable (itself defaulting to ``0``, which omits the attribute). Setting + this explicitly, including to ``0``, always overrides the environment + variable. + .. seealso:: The MongoDB documentation on `connections `_. .. versionchanged:: 4.19 @@ -671,6 +688,9 @@ def validator(host: str) -> bool: .. versionchanged:: 4.18 Added the ``srvAllowedHostsSuffix`` URI and keyword argument. + .. versionchanged:: 4.18 + Added the ``tracing`` keyword argument. + .. versionchanged:: 4.17 Added the ``max_adaptive_retries`` and ``enable_overload_retargeting`` URI and keyword arguments. @@ -1036,6 +1056,10 @@ def _resolve_srv(self) -> None: # Username and password passed as kwargs override user info in URI. username = opts.get("username", self._resolve_srv_info["username"]) password = opts.get("password", self._resolve_srv_info["password"]) + # The tracing option was resolved against the environment when the + # client was constructed; carry that snapshot into the rebuilt + # options rather than re-reading the (possibly changed) environment. + opts["tracing"] = self._options.tracing self._options = ClientOptions( username, password, self._resolve_srv_info["dbase"], opts, _IS_SYNC ) @@ -1963,43 +1987,51 @@ def _run_operation( :param address: Optional address when sending a message to a specific server, used for getMore. """ - if operation.conn_mgr: - server = self._select_server( + # The spec's operation span: one span per driver operation, started + # before server selection, with the operation's command spans nested + # beneath it. getMores are cursor territory (PYTHON-5947): no + # operation span yet. + # getMores are cursor territory (PYTHON-5947): no operation span yet. + tracing = self.options.tracing if not isinstance(operation, _GetMore) else None + op_span = operation_span(tracing, operation.name, operation.db, operation.coll) + with op_span: + if operation.conn_mgr: + server = self._select_server( + operation.read_preference, + operation.session, # type: ignore[arg-type] + operation.name, + address=address, + ) + + with operation.conn_mgr._lock: + with _ClientCheckout.for_existing_conn( + self, + server, + operation.session, # type: ignore[arg-type] + operation.conn_mgr.conn, + ): + return run_with_conn( + operation.conn_mgr.conn, operation, operation.read_preference + ) + + def _cmd( + _session: Optional[ClientSession], + _server: Server, + conn: Connection, + read_preference: _ServerMode, + ) -> Response: + operation.reset() # Reset op in case of retry. + return run_with_conn(conn, operation, read_preference) + + return self._retryable_read( + _cmd, operation.read_preference, operation.session, # type: ignore[arg-type] - operation.name, address=address, + retryable=isinstance(operation, _Query), + operation=operation.name, ) - with operation.conn_mgr._lock: - with _ClientCheckout.for_existing_conn( - self, - server, - operation.session, # type: ignore[arg-type] - operation.conn_mgr.conn, - ): - return run_with_conn( - operation.conn_mgr.conn, operation, operation.read_preference - ) - - def _cmd( - _session: Optional[ClientSession], - _server: Server, - conn: Connection, - read_preference: _ServerMode, - ) -> Response: - operation.reset() # Reset op in case of retry. - return run_with_conn(conn, operation, read_preference) - - return self._retryable_read( - _cmd, - operation.read_preference, - operation.session, # type: ignore[arg-type] - address=address, - retryable=isinstance(operation, _Query), - operation=operation.name, - ) - def _retry_with_session( self, retryable: bool, @@ -2135,6 +2167,8 @@ def _retryable_write( operation: str, bulk: Optional[Union[_Bulk, _ClientBulk]] = None, operation_id: Optional[int] = None, + db_name: Optional[str] = None, + coll_name: Optional[str] = None, ) -> T: """Execute an operation with consecutive retries if possible @@ -2149,9 +2183,19 @@ def _retryable_write( :param operation: The name of the operation that the server is being selected for :param bulk: bulk abstraction to execute operations in bulk, defaults to None :param operation_id: Stable operation id shared across retries, defaults to None + :param db_name: The database the operation targets, for the spec's + operation span. Defaults to None (attribute omitted). + :param coll_name: The collection the operation targets, for the spec's + operation span. Defaults to None (attribute omitted). """ with self._tmp_session(session) as s: - return self._retry_with_session(retryable, func, s, bulk, operation, operation_id) + with operation_span( + self.options.tracing, + getattr(operation, "value", operation), + db_name, + coll_name, + ): + return self._retry_with_session(retryable, func, s, bulk, operation, operation_id) def _cleanup_cursor_no_lock( self, diff --git a/pymongo/synchronous/pool.py b/pymongo/synchronous/pool.py index dd562ac68e..f881e31046 100644 --- a/pymongo/synchronous/pool.py +++ b/pymongo/synchronous/pool.py @@ -170,6 +170,9 @@ def __init__( # For load balancer support. self.service_id: Optional[ObjectId] = None self.server_connection_id: Optional[int] = None + # Cache of connection-static span attributes for OpenTelemetry + # command spans; see pymongo._otel._connection_attributes. + self._otel_connection_attributes: tuple[Optional[int], dict[str, Any]] = (None, {}) # When executing a transaction in load balancing mode, this flag is # set to true to indicate that the session now owns the connection. self.pinned_txn = False diff --git a/pyproject.toml b/pyproject.toml index 661c4d01e6..9a3f9574cd 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -104,6 +104,7 @@ docs = ["requirements/docs.txt"] encryption = ["requirements/encryption.txt"] gssapi = ["requirements/gssapi.txt"] ocsp = ["requirements/ocsp.txt"] +opentelemetry = ["requirements/opentelemetry.txt"] snappy = ["requirements/snappy.txt"] test = ["requirements/test.txt"] zstd = ["requirements/zstd.txt"] @@ -152,6 +153,7 @@ markers = [ "encryption: encryption tests", "load_balancer: load balancer tests", "mockupdb: tests that rely on mockupdb", + "otel: tests that rely on opentelemetry", "default: default test suite", "default_async: default async test suite", ] diff --git a/requirements/opentelemetry.txt b/requirements/opentelemetry.txt new file mode 100644 index 0000000000..d20388d07f --- /dev/null +++ b/requirements/opentelemetry.txt @@ -0,0 +1 @@ +opentelemetry-api>=1.20.0 diff --git a/test/asynchronous/test_otel.py b/test/asynchronous/test_otel.py new file mode 100644 index 0000000000..c2d7777252 --- /dev/null +++ b/test/asynchronous/test_otel.py @@ -0,0 +1,777 @@ +# Copyright 2026-present MongoDB, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Test OpenTelemetry command-span support.""" + +from __future__ import annotations + +import asyncio +import os +import sys +import time +from typing import Any, Optional +from unittest.mock import patch + +sys.path[0:0] = [""] + +import pytest + +import pymongo._otel as _otel +from pymongo import common +from pymongo.errors import ConfigurationError, OperationFailure +from pymongo.monitoring import CommandListener +from pymongo.operations import InsertOne +from pymongo.typings import _Address +from test.asynchronous import AsyncIntegrationTest, async_client_context, unittest + +_HAS_OTEL_TEST_DEPS = False +if _otel._HAS_OPENTELEMETRY: + try: + from opentelemetry import trace + from opentelemetry.sdk.trace import TracerProvider + from opentelemetry.sdk.trace.export import SimpleSpanProcessor + from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + + _HAS_OTEL_TEST_DEPS = True + except ImportError: + pass + +_IS_SYNC = False + +pytestmark = pytest.mark.otel + + +def _shared_test_provider() -> TracerProvider: + """Return a process-wide SDK TracerProvider for tests to attach exporters to. + + ``trace.set_tracer_provider`` only takes effect once per process (later calls + are silently ignored), so tests must share one provider and each register + their own span processor rather than trying to install a fresh provider. + """ + current = trace.get_tracer_provider() + if isinstance(current, TracerProvider): + return current + provider = TracerProvider() + trace.set_tracer_provider(provider) + return provider + + +@unittest.skipUnless(_HAS_OTEL_TEST_DEPS, "opentelemetry-sdk is not installed") +class TestOTelSpans(AsyncIntegrationTest): + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.exporter = InMemorySpanExporter() + _shared_test_provider().add_span_processor(SimpleSpanProcessor(cls.exporter)) + + async def asyncSetUp(self): + await super().asyncSetUp() + self.exporter.clear() + + def spans(self, name: str | None = None): + finished = self.exporter.get_finished_spans() + if name is None: + return list(finished) + return [s for s in finished if s.name == name] + + # TODO(PYTHON-5947): once the unified test format runner supports + # expectTracingMessages/operation spans, this is superseded by the spec's + # find_without_query_text.yml and insert.yml. + async def test_span_created_for_insert_and_find(self): + client = await self.async_rs_or_single_client(tracing={"enabled": True}) + coll = client[self.db.name].test_otel + await coll.drop() + self.exporter.clear() + await coll.insert_one({"x": 1}) + + insert_spans = self.spans("insert") + self.assertEqual(len(insert_spans), 1) + attrs = insert_spans[0].attributes + self.assertEqual(attrs["db.system.name"], "mongodb") + self.assertEqual(attrs["db.namespace"], self.db.name) + self.assertEqual(attrs["db.collection.name"], "test_otel") + self.assertEqual(attrs["db.command.name"], "insert") + self.assertEqual(attrs["db.query.summary"], f"insert {self.db.name}.test_otel") + self.assertIn("server.address", attrs) + self.assertIn("server.port", attrs) + self.assertIn(attrs["network.transport"], ("tcp", "unix")) + self.assertIn("db.mongodb.driver_connection_id", attrs) + self.assertIn("db.mongodb.server_connection_id", attrs) + self.assertIn("db.mongodb.lsid", attrs) + self.assertNotIn("db.query.text", attrs) + + self.exporter.clear() + docs = await coll.find({}).to_list() + self.assertEqual(len(docs), 1) + find_spans = self.spans("find") + self.assertEqual(len(find_spans), 1) + self.assertEqual(find_spans[0].attributes["db.command.name"], "find") + + async def test_span_created_for_get_more(self): + client = await self.async_rs_or_single_client(tracing={"enabled": True}) + coll = client[self.db.name].test_otel_getmore + await coll.drop() + await coll.insert_many([{"x": i} for i in range(5)]) + self.exporter.clear() + + docs = await coll.find({}, batch_size=2).to_list() + self.assertEqual(len(docs), 5) + + get_more_spans = self.spans("getMore") + self.assertGreater(len(get_more_spans), 0) + for span in get_more_spans: + self.assertEqual(span.attributes["db.collection.name"], "test_otel_getmore") + self.assertEqual(span.attributes["db.command.name"], "getMore") + + async def test_cursor_id_omitted_when_first_batch_exhausts_cursor(self): + # A find whose reply carries a cursor id of 0 (everything fit in the + # first batch) must not record db.mongodb.cursor_id: the spec forbids + # a literal 0 (spec: "db.mongodb.cursor_id"). + client = await self.async_rs_or_single_client(tracing={"enabled": True}) + coll = client[self.db.name].test_otel_cursor_id + await coll.drop() + await coll.insert_one({"x": 1}) + self.exporter.clear() + + docs = await coll.find({}).to_list() + self.assertEqual(len(docs), 1) + + find_spans = self.spans("find") + self.assertEqual(len(find_spans), 1) + self.assertNotIn("db.mongodb.cursor_id", find_spans[0].attributes) + + async def test_get_more_records_sent_cursor_id_when_reply_exhausts(self): + # The final getMore's reply carries a cursor id of 0 (cursor exhausted); + # its span must record the nonzero cursor id the command sent instead + # (spec: "db.mongodb.cursor_id"). + class StartedListener(CommandListener): + def __init__(self): + self.sent_ids: list = [] + + def started(self, event): + if event.command_name == "getMore": + self.sent_ids.append(event.command["getMore"]) + + def succeeded(self, event): + pass + + def failed(self, event): + pass + + listener = StartedListener() + client = await self.async_rs_or_single_client( + tracing={"enabled": True}, event_listeners=[listener] + ) + coll = client[self.db.name].test_otel_cursor_id + await coll.drop() + await coll.insert_many([{"x": i} for i in range(5)]) + self.exporter.clear() + + docs = await coll.find({}, batch_size=2).to_list() + self.assertEqual(len(docs), 5) + + get_more_spans = self.spans("getMore") + self.assertGreater(len(get_more_spans), 0) + self.assertEqual(len(get_more_spans), len(listener.sent_ids)) + # Every getMore span records the cursor id the command sent; the final + # one exhausts the cursor (reply id 0) but must not record 0. + for span, sent_id in zip(get_more_spans, listener.sent_ids): + self.assertEqual(span.attributes["db.mongodb.cursor_id"], sent_id) + self.assertNotEqual(listener.sent_ids[-1], 0) + # The find's reply carried a nonzero cursor id, which its span records. + find_spans = self.spans("find") + self.assertEqual(len(find_spans), 1) + self.assertEqual(find_spans[0].attributes["db.mongodb.cursor_id"], listener.sent_ids[0]) + + async def test_explain_retains_collection_name(self): + # explain wraps the real command ({"explain": {"find": "coll", ...}}), the + # same shape as getMore's indirection, so it needs the same handling. + client = await self.async_rs_or_single_client(tracing={"enabled": True}) + self.exporter.clear() + await client[self.db.name].command("explain", {"find": "test_otel", "filter": {}}) + + spans = self.spans("explain") + self.assertEqual(len(spans), 1) + attrs = spans[0].attributes + self.assertEqual(attrs["db.collection.name"], "test_otel") + self.assertEqual(attrs["db.query.summary"], f"explain {self.db.name}.test_otel") + + async def test_user_management_commands_omit_collection_name(self): + # usersInfo's string command value names a user, not a collection: + # db.collection.name must be omitted so usernames aren't mislabeled + # (and exposed) as collections. + client = await self.async_rs_or_single_client(tracing={"enabled": True}) + self.exporter.clear() + await client[self.db.name].command("usersInfo", "someuser") + + spans = self.spans("usersInfo") + self.assertEqual(len(spans), 1) + attrs = spans[0].attributes + self.assertNotIn("db.collection.name", attrs) + self.assertEqual(attrs["db.query.summary"], f"usersInfo {self.db.name}") + + async def test_server_port_omitted_for_unix_socket(self): + class _FakeUnixConn: + id = 1 + server_connection_id: Optional[int] = None + address: _Address = ("/tmp/fake-otel-test.sock", None) + service_id = None + _otel_connection_attributes: tuple[Optional[int], dict[str, Any]] = (None, {}) + + self.exporter.clear() + span = _otel.start_command_span( + {"enabled": True, "query_text_max_length": None}, + _FakeUnixConn(), + {"ping": 1}, + "admin", + "ping", + False, + ) + _otel.end_command_span_success(span, {"ping": 1}, "ping", {"ok": 1}) + + spans = self.spans("ping") + self.assertEqual(len(spans), 1) + attrs = spans[0].attributes + self.assertNotIn("server.port", attrs) + self.assertEqual(attrs["network.transport"], "unix") + + async def test_sensitive_command_produces_no_span(self): + client = await self.async_rs_or_single_client(tracing={"enabled": True}) + self.exporter.clear() + with self.assertRaises(OperationFailure): + await client.admin.command("saslStart", mechanism="SCRAM-SHA-256", payload=b"") + + names = [s.name for s in self.spans()] + self.assertNotIn("saslStart", names) + + async def test_admin_command_omits_collection_name(self): + # usersInfo's command value is a username string, not a collection, and + # it always runs against admin; querying a nonexistent user is a no-op. + client = await self.async_rs_or_single_client(tracing={"enabled": True}) + self.exporter.clear() + await client.admin.command("usersInfo", "pymongo_otel_nonexistent_user") + + spans = self.spans("usersInfo") + self.assertEqual(len(spans), 1) + attrs = spans[0].attributes + self.assertEqual(attrs["db.namespace"], "admin") + self.assertNotIn("db.collection.name", attrs) + self.assertEqual(attrs["db.query.summary"], "usersInfo admin") + + async def test_failure_records_exception_and_status_code(self): + client = await self.async_rs_or_single_client(tracing={"enabled": True}) + self.exporter.clear() + with self.assertRaises(OperationFailure): + await client[self.db.name].command("thisCommandDoesNotExist") + + spans = self.spans() + self.assertEqual(len(spans), 1) + span = spans[0] + self.assertEqual(span.status.status_code, trace.StatusCode.ERROR) + self.assertIn("db.response.status_code", span.attributes) + # error.type mirrors the server response code for server failures. + self.assertEqual(span.attributes["error.type"], span.attributes["db.response.status_code"]) + self.assertTrue(any(event.name == "exception" for event in span.events)) + + async def test_transport_failure_records_exception_class_as_error_type(self): + class _FakeConn: + id = 1 + server_connection_id: Optional[int] = None + address: _Address = ("localhost", 27017) + service_id = None + _otel_connection_attributes: tuple[Optional[int], dict[str, Any]] = (None, {}) + + options: _otel.TracingOptions = {"enabled": True, "query_text_max_length": None} + self.exporter.clear() + + # Transport failure: no server response, error.type is the exception + # class name and db.response.status_code is absent. + span = _otel.start_command_span(options, _FakeConn(), {"ping": 1}, "admin", "ping", False) + _otel.end_command_span_failure(span, {}, TimeoutError("timed out")) + spans = self.spans() + self.assertEqual(len(spans), 1) + attrs = spans[0].attributes + self.assertEqual(spans[0].status.status_code, trace.StatusCode.ERROR) + self.assertNotIn("db.response.status_code", attrs) + self.assertEqual(attrs["error.type"], "TimeoutError") + + # Server failure: error.type mirrors db.response.status_code. + self.exporter.clear() + span = _otel.start_command_span(options, _FakeConn(), {"ping": 1}, "admin", "ping", False) + _otel.end_command_span_failure( + span, {"code": 26, "errmsg": "not found"}, OperationFailure("") + ) + spans = self.spans() + self.assertEqual(len(spans), 1) + attrs = spans[0].attributes + self.assertEqual(spans[0].status.status_code, trace.StatusCode.ERROR) + self.assertEqual(attrs["db.response.status_code"], "26") + self.assertEqual(attrs["error.type"], "26") + + async def test_operation_span_wraps_command_span(self): + client = await self.async_rs_or_single_client(tracing={"enabled": True}) + coll = client[self.db.name].test_otel_operation + await coll.drop() + await coll.insert_one({"x": 1}) + self.exporter.clear() + + await coll.insert_one({"x": 2}) + + # One operation span per public-API call, named and attributed per the + # spec, with the operation's command span nested beneath it. + op_name = f"insert {self.db.name}.test_otel_operation" + op_spans = self.spans(op_name) + self.assertEqual(len(op_spans), 1) + op_span = op_spans[0] + attrs = op_span.attributes + self.assertEqual(attrs["db.system.name"], "mongodb") + self.assertEqual(attrs["db.namespace"], self.db.name) + self.assertEqual(attrs["db.collection.name"], "test_otel_operation") + self.assertEqual(attrs["db.operation.name"], "insert") + self.assertEqual(attrs["db.operation.summary"], op_name) + + cmd_spans = self.spans("insert") + self.assertEqual(len(cmd_spans), 1) + self.assertEqual(cmd_spans[0].parent.span_id, op_span.context.span_id) + + async def test_operation_failure_records_exception_class_as_error_type(self): + client = await self.async_rs_or_single_client(tracing={"enabled": True}) + coll = client[self.db.name].test_otel_operation + await coll.drop() + await coll.create_index("x", unique=True) + await coll.insert_one({"x": 1}) + self.exporter.clear() + + with self.assertRaises(OperationFailure): + await coll.insert_one({"x": 1}) + + # The operation span's error.type is the exception class raised to the + # application (the specific subclass pymongo raises); the failed + # command span keeps the server response code. + op_span = self.spans(f"insert {self.db.name}.test_otel_operation")[0] + self.assertEqual(op_span.status.status_code, trace.StatusCode.ERROR) + self.assertEqual(op_span.attributes["error.type"], "DuplicateKeyError") + # The duplicate key arrived as writeErrors in an ok:1 reply: the + # command succeeded at the protocol level, so its span carries no + # error; the failure belongs to the operation span alone. + cmd_spans = self.spans("insert") + self.assertEqual(len(cmd_spans), 1) + self.assertEqual(cmd_spans[0].status.status_code, trace.StatusCode.UNSET) + + async def test_operation_span_absent_when_tracing_disabled(self): + client = await self.async_rs_or_single_client() + coll = client[self.db.name].test_otel_operation + await coll.drop() + await coll.insert_one({"x": 1}) + self.assertEqual(self.spans(), []) + + async def test_tracing_disabled_by_default(self): + client = await self.async_rs_or_single_client() + self.exporter.clear() + await client.admin.command("ping") + self.assertEqual(self.spans(), []) + + # TODO(PYTHON-5947): once operation spans exist, also assert that the + # "ping" *operation* span (not just the command span) is absent/present + # here, and that self.spans() counts both. + async def test_prose_1_tracing_enable_disable_via_env_var(self): + """Prose Test 1: Tracing Enable/Disable via Environment Variable.""" + with patch.dict(os.environ, {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED": "false"}): + client = await self.async_rs_or_single_client() + self.exporter.clear() + await client.admin.command("ping") + self.assertEqual(self.spans(), []) + + with patch.dict(os.environ, {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED": "true"}): + client = await self.async_rs_or_single_client() + self.exporter.clear() + await client.admin.command("ping") + self.assertIn("ping", [s.name for s in self.spans()]) + + # TODO(PYTHON-5947): once operation spans exist, self.spans("find") will + # also match the outer find *operation* span; disambiguate (e.g. by + # db.command.name vs db.operation.name) so this only asserts on the + # command span's db.query.text attribute. + async def test_prose_2_command_payload_emission_via_env_var(self): + """Prose Test 2: Command Payload Emission via Environment Variable.""" + env = { + "OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED": "true", + "OTEL_PYTHON_INSTRUMENTATION_MONGODB_QUERY_TEXT_MAX_LENGTH": "1024", + } + with patch.dict(os.environ, env): + client = await self.async_rs_or_single_client() + self.exporter.clear() + await client[self.db.name].test_otel.find({}).to_list() + spans = self.spans("find") + self.assertEqual(len(spans), 1) + self.assertIn("db.query.text", spans[0].attributes) + + with patch.dict(os.environ, {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED": "true"}): + client = await self.async_rs_or_single_client() + self.exporter.clear() + await client[self.db.name].test_otel.find({}).to_list() + spans = self.spans("find") + self.assertEqual(len(spans), 1) + self.assertNotIn("db.query.text", spans[0].attributes) + + # TODO(PYTHON-5947): once the unified test format runner supports + # expectTracingMessages/operation spans, this is superseded by the spec's + # find.yml (db.query.text assertion). + async def test_query_text_included_when_configured(self): + client = await self.async_rs_or_single_client( + tracing={"enabled": True, "query_text_max_length": 1000} + ) + coll = client[self.db.name].test_otel + await coll.drop() + self.exporter.clear() + await coll.insert_one({"x": 1}) + + spans = self.spans("insert") + self.assertEqual(len(spans), 1) + self.assertIn("db.query.text", spans[0].attributes) + self.assertNotIn("lsid", spans[0].attributes["db.query.text"]) + + async def test_explicit_query_text_max_length_zero_overrides_env_var(self): + # An explicit client-side 0 must win over the environment variable, unlike + # unset (which defers to it) - otherwise an app can't reliably opt out. + env = {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_QUERY_TEXT_MAX_LENGTH": "1024"} + with patch.dict(os.environ, env): + client = await self.async_rs_or_single_client( + tracing={"enabled": True, "query_text_max_length": 0} + ) + self.exporter.clear() + await client.admin.command("ping") + + spans = self.spans("ping") + self.assertEqual(len(spans), 1) + self.assertNotIn("db.query.text", spans[0].attributes) + + async def test_explicit_enabled_false_overrides_env_var(self): + # An explicit client-side disable must win over the environment + # variable, unlike unset (which defers to it) - otherwise an app + # can't reliably opt out. + with patch.dict(os.environ, {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED": "true"}): + client = await self.async_rs_or_single_client(tracing={"enabled": False}) + self.exporter.clear() + await client.admin.command("ping") + # Only the client's own commands are gated by the client option; + # connection handshakes consult the environment variable directly. + self.assertNotIn("ping", [s.name for s in self.spans()]) + + async def test_unset_enabled_defers_to_env_var(self): + # ``tracing={}`` leaves enabled unconfigured, so the environment + # variable decides. + with patch.dict(os.environ, {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED": "true"}): + client = await self.async_rs_or_single_client(tracing={}) + self.exporter.clear() + await client.admin.command("ping") + self.assertIn("ping", [s.name for s in self.spans()]) + + async def test_srv_resolution_preserves_construction_time_snapshot(self): + # An SRV client rebuilds its ClientOptions at first use (_resolve_srv); + # the rebuilt options must keep the tracing values resolved at + # construction rather than re-reading the (possibly changed) environment. + with ( + patch("pymongo.asynchronous.srv_resolver._SrvResolver.get_hosts") as mock_hosts, + patch("pymongo.asynchronous.srv_resolver._SrvResolver.get_options", return_value=None), + ): + mock_hosts.return_value = [("localhost", 27017)] + with patch.dict(os.environ, {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED": "true"}): + client = self.unmanaged_simple_client( + "mongodb+srv://test.example.com", connect=False + ) + self.addAsyncCleanup(client.close) + self.assertEqual(client.options.tracing["enabled"], True) + # The environment changes after construction; the snapshot must not. + os.environ["OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED"] = "false" + await client._get_topology() + self.assertEqual(client.options.tracing["enabled"], True) + + async def test_query_text_truncation_shrinks_oversized_field_values(self): + client = await self.async_rs_or_single_client( + tracing={"enabled": True, "query_text_max_length": 200} + ) + coll = client[self.db.name].test_otel + await coll.drop() + self.exporter.clear() + await coll.insert_one({"x": "a" * 500}) + + spans = self.spans("insert") + self.assertEqual(len(spans), 1) + query_text = spans[0].attributes["db.query.text"] + # The oversized field value must be truncated at the field level (not + # just a blind cut of the fully-serialized string), and the result must + # never exceed the configured bound, even when a "..." marker is added. + self.assertLessEqual(len(query_text), 200) + self.assertNotIn("a" * 500, query_text) + + def test_query_text_tiny_max_length_truncates_without_suffix(self): + # Budgets smaller than the "..." marker must truncate without it so + # the result still honors the configured bound. + for max_length in (1, 2): + text = _otel._build_query_text({"ping": 1}, max_length) + self.assertEqual(len(text), max_length) + + def test_is_sensitive_command_case_insensitive(self): + # Redaction normalizes the command name, mirroring the comparison in + # command monitoring: a differently-cased sensitive command gets no + # span, so its payload cannot leak through db.query.text. + for name in ("saslStart", "SASLSTART", "saslstart", "CreateUser", "createuser"): + self.assertTrue(_otel._is_sensitive_command(name, False)) + self.assertFalse(_otel._is_sensitive_command("find", False)) + + def test_no_client_options_is_never_traced(self): + # ``None`` tracing options mean no client context (connection + # handshakes, server monitoring): never traced, even when the + # environment variable enables driver-level tracing. + with patch.dict(os.environ, {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED": "true"}): + self.assertFalse(_otel._is_tracing_enabled(None)) + + # Client-level bulk_write requires MongoDB 8.0+ (wire version 25). + @async_client_context.require_version_min(8, 0, 0, -24) + async def test_unacknowledged_bulk_write_query_text_includes_documents(self): + # db.query.text is built from the document published in + # CommandStartedEvent, which carries the write documents that the + # wire command omits for unacknowledged bulk writes. + client = await self.async_rs_or_single_client( + tracing={"enabled": True, "query_text_max_length": 1000}, w=0 + ) + self.exporter.clear() + await client.bulk_write( + [InsertOne(namespace=f"{self.db.name}.test_otel", document={"x": 1})], ordered=False + ) + (span,) = self.spans("bulkWrite") + self.assertIn('"x": 1', span.attributes["db.query.text"]) + + @async_client_context.require_failCommand_blockConnection + @async_client_context.require_async + async def test_span_ended_on_task_cancellation(self): + # Task cancellation raises CancelledError (a BaseException), which the + # command runner's cleanup must handle: the span ends with an error + # status and the cancellation still propagates. + client = await self.async_rs_or_single_client(tracing={"enabled": True}) + coll = client[self.db.name].test_otel + await coll.drop() + await coll.insert_many([{"x": i} for i in range(5)]) + self.exporter.clear() + + fail_command = { + "configureFailPoint": "failCommand", + "mode": "alwaysOn", + "data": {"failCommands": ["getMore"], "blockConnection": True, "blockTimeMS": 5000}, + } + + async def task(): + cursor = coll.find({}, batch_size=1) + await cursor.next() + async with self.fail_point(fail_command): + await cursor.next() + + running = asyncio.create_task(task()) + await asyncio.sleep(0.1) + start = time.monotonic() + running.cancel() + with self.assertRaises(asyncio.CancelledError): + await running + # The cancellation surfaces once the failPoint's block on the + # connection releases; it must not hang indefinitely. + self.assertLess(time.monotonic() - start, 7) + + (span,) = self.spans("getMore") + self.assertEqual(span.status.status_code, trace.StatusCode.ERROR) + + +# TODO(PYTHON-5947): superseded once the unified test format's +# expectTracingMessages/observeTracingMessages tests exercise this validator +# indirectly through real client construction; remove this class then. +class TestValidateTracingOrNone(unittest.TestCase): + def test_none(self): + self.assertIsNone(common.validate_tracing_or_none("tracing", None)) + + def test_defaults(self): + self.assertEqual( + common.validate_tracing_or_none("tracing", {}), + {"enabled": None, "query_text_max_length": None}, + ) + + def test_explicit_enabled_false_preserved(self): + # False must stay distinct from "unset" (None) so it can override the + # environment variable instead of deferring to it. + result = common.validate_tracing_or_none("tracing", {"enabled": False}) + self.assertIs(result["enabled"], False) + + def test_enabled_and_query_text_max_length(self): + self.assertEqual( + common.validate_tracing_or_none( + "tracing", {"enabled": True, "query_text_max_length": 500} + ), + {"enabled": True, "query_text_max_length": 500}, + ) + + def test_explicit_zero_query_text_max_length_preserved(self): + # 0 must stay distinct from "unset" (None) so it can override the + # environment variable instead of being treated as not configured. + result = common.validate_tracing_or_none( + "tracing", {"enabled": True, "query_text_max_length": 0} + ) + self.assertEqual(result["query_text_max_length"], 0) + + def test_rejects_non_mapping(self): + with self.assertRaises(TypeError): + common.validate_tracing_or_none("tracing", "enabled") + + def test_rejects_unknown_option(self): + with self.assertRaisesRegex(ConfigurationError, "Unknown tracing option"): + common.validate_tracing_or_none("tracing", {"bogus": True}) + + def test_rejects_non_boolean_enabled(self): + with self.assertRaises(TypeError): + common.validate_tracing_or_none("tracing", {"enabled": "yes"}) + + def test_rejects_non_integer_query_text_max_length(self): + with self.assertRaises(TypeError): + common.validate_tracing_or_none("tracing", {"query_text_max_length": [1]}) + + def test_rejects_negative_query_text_max_length(self): + with self.assertRaises(ValueError): + common.validate_tracing_or_none("tracing", {"query_text_max_length": -1}) + + def test_numeric_string_query_text_max_length_converted(self): + # The validator's converted value must be kept: a numeric string that + # passes validation cannot flow into max(0, value) at command time. + result = common.validate_tracing_or_none("tracing", {"query_text_max_length": "100"}) + self.assertEqual(result["query_text_max_length"], 100) + + +class TestResolveTracingOptions(unittest.TestCase): + """The client's ``tracing`` option is resolved against the environment once, at construction.""" + + def _without_otel_env(self): + env = os.environ.copy() + env.pop("OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED", None) + env.pop("OTEL_PYTHON_INSTRUMENTATION_MONGODB_QUERY_TEXT_MAX_LENGTH", None) + return env + + def test_none_without_env(self): + with patch.dict(os.environ, self._without_otel_env(), clear=True): + self.assertEqual( + _otel._resolve_tracing_options(None), + {"enabled": False, "query_text_max_length": 0}, + ) + + # The environment-deferral semantics only exist when opentelemetry is + # importable; without it _resolve_tracing_options always disables tracing. + @unittest.skipUnless(_otel._HAS_OPENTELEMETRY, "opentelemetry is not installed") + def test_unset_defers_to_env(self): + with patch.dict(os.environ, {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED": "true"}): + self.assertTrue( + _otel._resolve_tracing_options({"enabled": None, "query_text_max_length": None})[ + "enabled" + ] + ) + + def test_env_false(self): + with patch.dict(os.environ, {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED": "false"}): + self.assertFalse( + _otel._resolve_tracing_options({"enabled": None, "query_text_max_length": None})[ + "enabled" + ] + ) + + # Explicit values only win over the environment when tracing is available. + @unittest.skipUnless(_otel._HAS_OPENTELEMETRY, "opentelemetry is not installed") + def test_explicit_enabled_overrides_env(self): + env = {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED": "true"} + with patch.dict(os.environ, env): + resolved = _otel._resolve_tracing_options( + {"enabled": False, "query_text_max_length": None} + ) + self.assertIs(resolved["enabled"], False) + resolved = _otel._resolve_tracing_options( + {"enabled": True, "query_text_max_length": None} + ) + self.assertIs(resolved["enabled"], True) + + # The environment-deferral semantics only exist when opentelemetry is + # importable; without it _resolve_tracing_options always omits query text. + @unittest.skipUnless(_otel._HAS_OPENTELEMETRY, "opentelemetry is not installed") + def test_query_text_max_length_from_env(self): + with patch.dict( + os.environ, {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_QUERY_TEXT_MAX_LENGTH": "1024"} + ): + self.assertEqual( + _otel._resolve_tracing_options({"enabled": None, "query_text_max_length": None})[ + "query_text_max_length" + ], + 1024, + ) + + def test_explicit_zero_query_text_overrides_env(self): + env = {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_QUERY_TEXT_MAX_LENGTH": "1024"} + with patch.dict(os.environ, env): + resolved = _otel._resolve_tracing_options({"enabled": None, "query_text_max_length": 0}) + self.assertEqual(resolved["query_text_max_length"], 0) + + def test_invalid_env_query_text_is_zero(self): + with patch.dict( + os.environ, {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_QUERY_TEXT_MAX_LENGTH": "bogus"} + ): + self.assertEqual( + _otel._resolve_tracing_options({"enabled": None, "query_text_max_length": None})[ + "query_text_max_length" + ], + 0, + ) + + def test_negative_query_text_clamped(self): + resolved = _otel._resolve_tracing_options({"enabled": None, "query_text_max_length": -5}) + self.assertEqual(resolved["query_text_max_length"], 0) + + +class TestOTelTracerCaching(unittest.TestCase): + """Regression test for the tracer-caching implementation in ``pymongo/_otel.py``. + + ``opentelemetry.trace.get_tracer()`` must only be called once, at import + time (cached as module-level ``_otel._TRACER``). Calling it per command + allocates two objects, takes a process-wide lock, and mutates the global + ``warnings`` filter list on every call, even on a cache hit. + """ + + @unittest.skipUnless(_otel._HAS_OPENTELEMETRY, "opentelemetry is not installed") + def test_start_command_span_does_not_call_get_tracer(self): + class _FakeConn: + id = 1 + server_connection_id: Optional[int] = None + address: _Address = ("localhost", 27017) + service_id = None + _otel_connection_attributes: tuple[Optional[int], dict[str, Any]] = (None, {}) + + with patch.object(_otel, "trace") as mock_trace: + for _ in range(3): + span = _otel.start_command_span( + {"enabled": True, "query_text_max_length": None}, + _FakeConn(), + {"ping": 1}, + "admin", + "ping", + False, + ) + self.assertIsNotNone(span) + span.end() + + mock_trace.get_tracer.assert_not_called() + + +if __name__ == "__main__": + unittest.main() diff --git a/test/performance/async_perf_test.py b/test/performance/async_perf_test.py index e4a68e228e..fb2069c429 100644 --- a/test/performance/async_perf_test.py +++ b/test/performance/async_perf_test.py @@ -87,6 +87,11 @@ OUTPUT_FILE = os.environ.get("OUTPUT_FILE") +# When set to a truthy value, also record the median process CPU time per +# iteration in the results. Off by default so the standard results.json +# schema is unchanged. +RECORD_CPU_TIME = os.environ.get("PERF_CPU_TIME", "").lower() in ("1", "true", "yes") + result_data: list = [] @@ -102,11 +107,15 @@ def tearDownModule(): class Timer: def __enter__(self): self.start = time.monotonic() + if RECORD_CPU_TIME: + self.cpu_start = time.process_time() return self def __exit__(self, *args): self.end = time.monotonic() self.interval = self.end - self.start + if RECORD_CPU_TIME: + self.cpu_interval = time.process_time() - self.cpu_start async def concurrent(n_tasks, func): @@ -135,6 +144,29 @@ async def asyncTearDown(self): f"Completed {self.__class__.__name__} {megabytes_per_sec:.3f} MB/s, MEDIAN={self.percentile(50):.3f}s, " f"total time={duration:.3f}s, iterations={len(self.results)}" ) + metrics = [ + { + "name": "megabytes_per_sec", + "type": "MEDIAN", + "value": megabytes_per_sec, + "metadata": { + "improvement_direction": "up", + "measurement_unit": "megabytes_per_second", + }, + } + ] + if RECORD_CPU_TIME: + metrics.append( + { + "name": "cpu_time_median", + "type": "MEDIAN", + "value": self.percentile(50, self.cpu_results), + "metadata": { + "improvement_direction": "down", + "measurement_unit": "cpu_seconds_per_iteration", + }, + } + ) result_data.append( { "info": { @@ -143,17 +175,7 @@ async def asyncTearDown(self): "tasks": self.n_tasks, }, }, - "metrics": [ - { - "name": "megabytes_per_sec", - "type": "MEDIAN", - "value": megabytes_per_sec, - "metadata": { - "improvement_direction": "up", - "measurement_unit": "megabytes_per_second", - }, - }, - ], + "metrics": metrics, } ) @@ -166,9 +188,11 @@ async def do_task(self): async def after(self): pass - def percentile(self, percentile): - if hasattr(self, "results"): - sorted_results = sorted(self.results) + def percentile(self, percentile, values=None): + if values is None: + values = getattr(self, "results", None) + if values: + sorted_results = sorted(values) percentile_index = int(len(sorted_results) * percentile / 100) - 1 return sorted_results[percentile_index] else: @@ -177,6 +201,7 @@ def percentile(self, percentile): async def runTest(self): results = [] + cpu_results = [] start = time.monotonic() i = 0 while True: @@ -189,6 +214,8 @@ async def runTest(self): await concurrent(self.n_tasks, self.do_task) await self.after() results.append(timer.interval) + if RECORD_CPU_TIME: + cpu_results.append(timer.cpu_interval) duration = time.monotonic() - start if duration > MIN_ITERATION_TIME and i >= NUM_ITERATIONS: break @@ -204,6 +231,7 @@ async def runTest(self): break self.results = results + self.cpu_results = cpu_results # SINGLE-DOC BENCHMARKS diff --git a/test/performance/perf_test.py b/test/performance/perf_test.py index 5ec13a2b56..1b8109225f 100644 --- a/test/performance/perf_test.py +++ b/test/performance/perf_test.py @@ -89,6 +89,11 @@ OUTPUT_FILE = os.environ.get("OUTPUT_FILE") +# When set to a truthy value, also record the median process CPU time per +# iteration in the results. Off by default so the standard results.json +# schema is unchanged. +RECORD_CPU_TIME = os.environ.get("PERF_CPU_TIME", "").lower() in ("1", "true", "yes") + result_data: list = [] @@ -104,11 +109,15 @@ def tearDownModule(): class Timer: def __enter__(self): self.start = time.monotonic() + if RECORD_CPU_TIME: + self.cpu_start = time.process_time() return self def __exit__(self, *args): self.end = time.monotonic() self.interval = self.end - self.start + if RECORD_CPU_TIME: + self.cpu_interval = time.process_time() - self.cpu_start def threaded(n_threads, func): @@ -142,6 +151,29 @@ def tearDown(self): f"Completed {self.__class__.__name__} {megabytes_per_sec:.3f} MB/s, MEDIAN={self.percentile(50):.3f}s, " f"total time={duration:.3f}s, iterations={len(self.results)}" ) + metrics = [ + { + "name": "megabytes_per_sec", + "type": "MEDIAN", + "value": megabytes_per_sec, + "metadata": { + "improvement_direction": "up", + "measurement_unit": "megabytes_per_second", + }, + } + ] + if RECORD_CPU_TIME: + metrics.append( + { + "name": "cpu_time_median", + "type": "MEDIAN", + "value": self.percentile(50, self.cpu_results), + "metadata": { + "improvement_direction": "down", + "measurement_unit": "cpu_seconds_per_iteration", + }, + } + ) result_data.append( { "info": { @@ -150,17 +182,7 @@ def tearDown(self): "threads": self.n_threads, }, }, - "metrics": [ - { - "name": "megabytes_per_sec", - "type": "MEDIAN", - "value": megabytes_per_sec, - "metadata": { - "improvement_direction": "up", - "measurement_unit": "megabytes_per_second", - }, - }, - ], + "metrics": metrics, } ) @@ -173,9 +195,11 @@ def do_task(self): def after(self): pass - def percentile(self, percentile): - if hasattr(self, "results"): - sorted_results = sorted(self.results) + def percentile(self, percentile, values=None): + if values is None: + values = getattr(self, "results", None) + if values: + sorted_results = sorted(values) percentile_index = int(len(sorted_results) * percentile / 100) - 1 return sorted_results[percentile_index] else: @@ -184,6 +208,7 @@ def percentile(self, percentile): def runTest(self): results = [] + cpu_results = [] start = time.monotonic() i = 0 while True: @@ -196,6 +221,8 @@ def runTest(self): threaded(self.n_threads, self.do_task) self.after() results.append(timer.interval) + if RECORD_CPU_TIME: + cpu_results.append(timer.cpu_interval) duration = time.monotonic() - start if duration > MIN_ITERATION_TIME and i >= NUM_ITERATIONS: break @@ -209,6 +236,7 @@ def runTest(self): break self.results = results + self.cpu_results = cpu_results def mp_map(self, map_func, files): with mp.Pool(initializer=proc_init, initargs=(client_context.client_options,)) as pool: diff --git a/test/test_otel.py b/test/test_otel.py new file mode 100644 index 0000000000..b4be281931 --- /dev/null +++ b/test/test_otel.py @@ -0,0 +1,769 @@ +# Copyright 2026-present MongoDB, Inc. +# +# Licensed under the Apache License, Version 2.0 (the "License"); +# you may not use this file except in compliance with the License. +# You may obtain a copy of the License at +# +# http://www.apache.org/licenses/LICENSE-2.0 +# +# Unless required by applicable law or agreed to in writing, software +# distributed under the License is distributed on an "AS IS" BASIS, +# WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. +# See the License for the specific language governing permissions and +# limitations under the License. + +"""Test OpenTelemetry command-span support.""" + +from __future__ import annotations + +import asyncio +import os +import sys +import time +from typing import Any, Optional +from unittest.mock import patch + +sys.path[0:0] = [""] + +import pytest + +import pymongo._otel as _otel +from pymongo import common +from pymongo.errors import ConfigurationError, OperationFailure +from pymongo.monitoring import CommandListener +from pymongo.operations import InsertOne +from pymongo.typings import _Address +from test import IntegrationTest, client_context, unittest + +_HAS_OTEL_TEST_DEPS = False +if _otel._HAS_OPENTELEMETRY: + try: + from opentelemetry import trace + from opentelemetry.sdk.trace import TracerProvider + from opentelemetry.sdk.trace.export import SimpleSpanProcessor + from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + + _HAS_OTEL_TEST_DEPS = True + except ImportError: + pass + +_IS_SYNC = True + +pytestmark = pytest.mark.otel + + +def _shared_test_provider() -> TracerProvider: + """Return a process-wide SDK TracerProvider for tests to attach exporters to. + + ``trace.set_tracer_provider`` only takes effect once per process (later calls + are silently ignored), so tests must share one provider and each register + their own span processor rather than trying to install a fresh provider. + """ + current = trace.get_tracer_provider() + if isinstance(current, TracerProvider): + return current + provider = TracerProvider() + trace.set_tracer_provider(provider) + return provider + + +@unittest.skipUnless(_HAS_OTEL_TEST_DEPS, "opentelemetry-sdk is not installed") +class TestOTelSpans(IntegrationTest): + @classmethod + def setUpClass(cls): + super().setUpClass() + cls.exporter = InMemorySpanExporter() + _shared_test_provider().add_span_processor(SimpleSpanProcessor(cls.exporter)) + + def setUp(self): + super().setUp() + self.exporter.clear() + + def spans(self, name: str | None = None): + finished = self.exporter.get_finished_spans() + if name is None: + return list(finished) + return [s for s in finished if s.name == name] + + # TODO(PYTHON-5947): once the unified test format runner supports + # expectTracingMessages/operation spans, this is superseded by the spec's + # find_without_query_text.yml and insert.yml. + def test_span_created_for_insert_and_find(self): + client = self.rs_or_single_client(tracing={"enabled": True}) + coll = client[self.db.name].test_otel + coll.drop() + self.exporter.clear() + coll.insert_one({"x": 1}) + + insert_spans = self.spans("insert") + self.assertEqual(len(insert_spans), 1) + attrs = insert_spans[0].attributes + self.assertEqual(attrs["db.system.name"], "mongodb") + self.assertEqual(attrs["db.namespace"], self.db.name) + self.assertEqual(attrs["db.collection.name"], "test_otel") + self.assertEqual(attrs["db.command.name"], "insert") + self.assertEqual(attrs["db.query.summary"], f"insert {self.db.name}.test_otel") + self.assertIn("server.address", attrs) + self.assertIn("server.port", attrs) + self.assertIn(attrs["network.transport"], ("tcp", "unix")) + self.assertIn("db.mongodb.driver_connection_id", attrs) + self.assertIn("db.mongodb.server_connection_id", attrs) + self.assertIn("db.mongodb.lsid", attrs) + self.assertNotIn("db.query.text", attrs) + + self.exporter.clear() + docs = coll.find({}).to_list() + self.assertEqual(len(docs), 1) + find_spans = self.spans("find") + self.assertEqual(len(find_spans), 1) + self.assertEqual(find_spans[0].attributes["db.command.name"], "find") + + def test_span_created_for_get_more(self): + client = self.rs_or_single_client(tracing={"enabled": True}) + coll = client[self.db.name].test_otel_getmore + coll.drop() + coll.insert_many([{"x": i} for i in range(5)]) + self.exporter.clear() + + docs = coll.find({}, batch_size=2).to_list() + self.assertEqual(len(docs), 5) + + get_more_spans = self.spans("getMore") + self.assertGreater(len(get_more_spans), 0) + for span in get_more_spans: + self.assertEqual(span.attributes["db.collection.name"], "test_otel_getmore") + self.assertEqual(span.attributes["db.command.name"], "getMore") + + def test_cursor_id_omitted_when_first_batch_exhausts_cursor(self): + # A find whose reply carries a cursor id of 0 (everything fit in the + # first batch) must not record db.mongodb.cursor_id: the spec forbids + # a literal 0 (spec: "db.mongodb.cursor_id"). + client = self.rs_or_single_client(tracing={"enabled": True}) + coll = client[self.db.name].test_otel_cursor_id + coll.drop() + coll.insert_one({"x": 1}) + self.exporter.clear() + + docs = coll.find({}).to_list() + self.assertEqual(len(docs), 1) + + find_spans = self.spans("find") + self.assertEqual(len(find_spans), 1) + self.assertNotIn("db.mongodb.cursor_id", find_spans[0].attributes) + + def test_get_more_records_sent_cursor_id_when_reply_exhausts(self): + # The final getMore's reply carries a cursor id of 0 (cursor exhausted); + # its span must record the nonzero cursor id the command sent instead + # (spec: "db.mongodb.cursor_id"). + class StartedListener(CommandListener): + def __init__(self): + self.sent_ids: list = [] + + def started(self, event): + if event.command_name == "getMore": + self.sent_ids.append(event.command["getMore"]) + + def succeeded(self, event): + pass + + def failed(self, event): + pass + + listener = StartedListener() + client = self.rs_or_single_client(tracing={"enabled": True}, event_listeners=[listener]) + coll = client[self.db.name].test_otel_cursor_id + coll.drop() + coll.insert_many([{"x": i} for i in range(5)]) + self.exporter.clear() + + docs = coll.find({}, batch_size=2).to_list() + self.assertEqual(len(docs), 5) + + get_more_spans = self.spans("getMore") + self.assertGreater(len(get_more_spans), 0) + self.assertEqual(len(get_more_spans), len(listener.sent_ids)) + # Every getMore span records the cursor id the command sent; the final + # one exhausts the cursor (reply id 0) but must not record 0. + for span, sent_id in zip(get_more_spans, listener.sent_ids): + self.assertEqual(span.attributes["db.mongodb.cursor_id"], sent_id) + self.assertNotEqual(listener.sent_ids[-1], 0) + # The find's reply carried a nonzero cursor id, which its span records. + find_spans = self.spans("find") + self.assertEqual(len(find_spans), 1) + self.assertEqual(find_spans[0].attributes["db.mongodb.cursor_id"], listener.sent_ids[0]) + + def test_explain_retains_collection_name(self): + # explain wraps the real command ({"explain": {"find": "coll", ...}}), the + # same shape as getMore's indirection, so it needs the same handling. + client = self.rs_or_single_client(tracing={"enabled": True}) + self.exporter.clear() + client[self.db.name].command("explain", {"find": "test_otel", "filter": {}}) + + spans = self.spans("explain") + self.assertEqual(len(spans), 1) + attrs = spans[0].attributes + self.assertEqual(attrs["db.collection.name"], "test_otel") + self.assertEqual(attrs["db.query.summary"], f"explain {self.db.name}.test_otel") + + def test_user_management_commands_omit_collection_name(self): + # usersInfo's string command value names a user, not a collection: + # db.collection.name must be omitted so usernames aren't mislabeled + # (and exposed) as collections. + client = self.rs_or_single_client(tracing={"enabled": True}) + self.exporter.clear() + client[self.db.name].command("usersInfo", "someuser") + + spans = self.spans("usersInfo") + self.assertEqual(len(spans), 1) + attrs = spans[0].attributes + self.assertNotIn("db.collection.name", attrs) + self.assertEqual(attrs["db.query.summary"], f"usersInfo {self.db.name}") + + def test_server_port_omitted_for_unix_socket(self): + class _FakeUnixConn: + id = 1 + server_connection_id: Optional[int] = None + address: _Address = ("/tmp/fake-otel-test.sock", None) + service_id = None + _otel_connection_attributes: tuple[Optional[int], dict[str, Any]] = (None, {}) + + self.exporter.clear() + span = _otel.start_command_span( + {"enabled": True, "query_text_max_length": None}, + _FakeUnixConn(), + {"ping": 1}, + "admin", + "ping", + False, + ) + _otel.end_command_span_success(span, {"ping": 1}, "ping", {"ok": 1}) + + spans = self.spans("ping") + self.assertEqual(len(spans), 1) + attrs = spans[0].attributes + self.assertNotIn("server.port", attrs) + self.assertEqual(attrs["network.transport"], "unix") + + def test_sensitive_command_produces_no_span(self): + client = self.rs_or_single_client(tracing={"enabled": True}) + self.exporter.clear() + with self.assertRaises(OperationFailure): + client.admin.command("saslStart", mechanism="SCRAM-SHA-256", payload=b"") + + names = [s.name for s in self.spans()] + self.assertNotIn("saslStart", names) + + def test_admin_command_omits_collection_name(self): + # usersInfo's command value is a username string, not a collection, and + # it always runs against admin; querying a nonexistent user is a no-op. + client = self.rs_or_single_client(tracing={"enabled": True}) + self.exporter.clear() + client.admin.command("usersInfo", "pymongo_otel_nonexistent_user") + + spans = self.spans("usersInfo") + self.assertEqual(len(spans), 1) + attrs = spans[0].attributes + self.assertEqual(attrs["db.namespace"], "admin") + self.assertNotIn("db.collection.name", attrs) + self.assertEqual(attrs["db.query.summary"], "usersInfo admin") + + def test_failure_records_exception_and_status_code(self): + client = self.rs_or_single_client(tracing={"enabled": True}) + self.exporter.clear() + with self.assertRaises(OperationFailure): + client[self.db.name].command("thisCommandDoesNotExist") + + spans = self.spans() + self.assertEqual(len(spans), 1) + span = spans[0] + self.assertEqual(span.status.status_code, trace.StatusCode.ERROR) + self.assertIn("db.response.status_code", span.attributes) + # error.type mirrors the server response code for server failures. + self.assertEqual(span.attributes["error.type"], span.attributes["db.response.status_code"]) + self.assertTrue(any(event.name == "exception" for event in span.events)) + + def test_transport_failure_records_exception_class_as_error_type(self): + class _FakeConn: + id = 1 + server_connection_id: Optional[int] = None + address: _Address = ("localhost", 27017) + service_id = None + _otel_connection_attributes: tuple[Optional[int], dict[str, Any]] = (None, {}) + + options: _otel.TracingOptions = {"enabled": True, "query_text_max_length": None} + self.exporter.clear() + + # Transport failure: no server response, error.type is the exception + # class name and db.response.status_code is absent. + span = _otel.start_command_span(options, _FakeConn(), {"ping": 1}, "admin", "ping", False) + _otel.end_command_span_failure(span, {}, TimeoutError("timed out")) + spans = self.spans() + self.assertEqual(len(spans), 1) + attrs = spans[0].attributes + self.assertEqual(spans[0].status.status_code, trace.StatusCode.ERROR) + self.assertNotIn("db.response.status_code", attrs) + self.assertEqual(attrs["error.type"], "TimeoutError") + + # Server failure: error.type mirrors db.response.status_code. + self.exporter.clear() + span = _otel.start_command_span(options, _FakeConn(), {"ping": 1}, "admin", "ping", False) + _otel.end_command_span_failure( + span, {"code": 26, "errmsg": "not found"}, OperationFailure("") + ) + spans = self.spans() + self.assertEqual(len(spans), 1) + attrs = spans[0].attributes + self.assertEqual(spans[0].status.status_code, trace.StatusCode.ERROR) + self.assertEqual(attrs["db.response.status_code"], "26") + self.assertEqual(attrs["error.type"], "26") + + def test_operation_span_wraps_command_span(self): + client = self.rs_or_single_client(tracing={"enabled": True}) + coll = client[self.db.name].test_otel_operation + coll.drop() + coll.insert_one({"x": 1}) + self.exporter.clear() + + coll.insert_one({"x": 2}) + + # One operation span per public-API call, named and attributed per the + # spec, with the operation's command span nested beneath it. + op_name = f"insert {self.db.name}.test_otel_operation" + op_spans = self.spans(op_name) + self.assertEqual(len(op_spans), 1) + op_span = op_spans[0] + attrs = op_span.attributes + self.assertEqual(attrs["db.system.name"], "mongodb") + self.assertEqual(attrs["db.namespace"], self.db.name) + self.assertEqual(attrs["db.collection.name"], "test_otel_operation") + self.assertEqual(attrs["db.operation.name"], "insert") + self.assertEqual(attrs["db.operation.summary"], op_name) + + cmd_spans = self.spans("insert") + self.assertEqual(len(cmd_spans), 1) + self.assertEqual(cmd_spans[0].parent.span_id, op_span.context.span_id) + + def test_operation_failure_records_exception_class_as_error_type(self): + client = self.rs_or_single_client(tracing={"enabled": True}) + coll = client[self.db.name].test_otel_operation + coll.drop() + coll.create_index("x", unique=True) + coll.insert_one({"x": 1}) + self.exporter.clear() + + with self.assertRaises(OperationFailure): + coll.insert_one({"x": 1}) + + # The operation span's error.type is the exception class raised to the + # application (the specific subclass pymongo raises); the failed + # command span keeps the server response code. + op_span = self.spans(f"insert {self.db.name}.test_otel_operation")[0] + self.assertEqual(op_span.status.status_code, trace.StatusCode.ERROR) + self.assertEqual(op_span.attributes["error.type"], "DuplicateKeyError") + # The duplicate key arrived as writeErrors in an ok:1 reply: the + # command succeeded at the protocol level, so its span carries no + # error; the failure belongs to the operation span alone. + cmd_spans = self.spans("insert") + self.assertEqual(len(cmd_spans), 1) + self.assertEqual(cmd_spans[0].status.status_code, trace.StatusCode.UNSET) + + def test_operation_span_absent_when_tracing_disabled(self): + client = self.rs_or_single_client() + coll = client[self.db.name].test_otel_operation + coll.drop() + coll.insert_one({"x": 1}) + self.assertEqual(self.spans(), []) + + def test_tracing_disabled_by_default(self): + client = self.rs_or_single_client() + self.exporter.clear() + client.admin.command("ping") + self.assertEqual(self.spans(), []) + + # TODO(PYTHON-5947): once operation spans exist, also assert that the + # "ping" *operation* span (not just the command span) is absent/present + # here, and that self.spans() counts both. + def test_prose_1_tracing_enable_disable_via_env_var(self): + """Prose Test 1: Tracing Enable/Disable via Environment Variable.""" + with patch.dict(os.environ, {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED": "false"}): + client = self.rs_or_single_client() + self.exporter.clear() + client.admin.command("ping") + self.assertEqual(self.spans(), []) + + with patch.dict(os.environ, {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED": "true"}): + client = self.rs_or_single_client() + self.exporter.clear() + client.admin.command("ping") + self.assertIn("ping", [s.name for s in self.spans()]) + + # TODO(PYTHON-5947): once operation spans exist, self.spans("find") will + # also match the outer find *operation* span; disambiguate (e.g. by + # db.command.name vs db.operation.name) so this only asserts on the + # command span's db.query.text attribute. + def test_prose_2_command_payload_emission_via_env_var(self): + """Prose Test 2: Command Payload Emission via Environment Variable.""" + env = { + "OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED": "true", + "OTEL_PYTHON_INSTRUMENTATION_MONGODB_QUERY_TEXT_MAX_LENGTH": "1024", + } + with patch.dict(os.environ, env): + client = self.rs_or_single_client() + self.exporter.clear() + client[self.db.name].test_otel.find({}).to_list() + spans = self.spans("find") + self.assertEqual(len(spans), 1) + self.assertIn("db.query.text", spans[0].attributes) + + with patch.dict(os.environ, {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED": "true"}): + client = self.rs_or_single_client() + self.exporter.clear() + client[self.db.name].test_otel.find({}).to_list() + spans = self.spans("find") + self.assertEqual(len(spans), 1) + self.assertNotIn("db.query.text", spans[0].attributes) + + # TODO(PYTHON-5947): once the unified test format runner supports + # expectTracingMessages/operation spans, this is superseded by the spec's + # find.yml (db.query.text assertion). + def test_query_text_included_when_configured(self): + client = self.rs_or_single_client(tracing={"enabled": True, "query_text_max_length": 1000}) + coll = client[self.db.name].test_otel + coll.drop() + self.exporter.clear() + coll.insert_one({"x": 1}) + + spans = self.spans("insert") + self.assertEqual(len(spans), 1) + self.assertIn("db.query.text", spans[0].attributes) + self.assertNotIn("lsid", spans[0].attributes["db.query.text"]) + + def test_explicit_query_text_max_length_zero_overrides_env_var(self): + # An explicit client-side 0 must win over the environment variable, unlike + # unset (which defers to it) - otherwise an app can't reliably opt out. + env = {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_QUERY_TEXT_MAX_LENGTH": "1024"} + with patch.dict(os.environ, env): + client = self.rs_or_single_client(tracing={"enabled": True, "query_text_max_length": 0}) + self.exporter.clear() + client.admin.command("ping") + + spans = self.spans("ping") + self.assertEqual(len(spans), 1) + self.assertNotIn("db.query.text", spans[0].attributes) + + def test_explicit_enabled_false_overrides_env_var(self): + # An explicit client-side disable must win over the environment + # variable, unlike unset (which defers to it) - otherwise an app + # can't reliably opt out. + with patch.dict(os.environ, {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED": "true"}): + client = self.rs_or_single_client(tracing={"enabled": False}) + self.exporter.clear() + client.admin.command("ping") + # Only the client's own commands are gated by the client option; + # connection handshakes consult the environment variable directly. + self.assertNotIn("ping", [s.name for s in self.spans()]) + + def test_unset_enabled_defers_to_env_var(self): + # ``tracing={}`` leaves enabled unconfigured, so the environment + # variable decides. + with patch.dict(os.environ, {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED": "true"}): + client = self.rs_or_single_client(tracing={}) + self.exporter.clear() + client.admin.command("ping") + self.assertIn("ping", [s.name for s in self.spans()]) + + def test_srv_resolution_preserves_construction_time_snapshot(self): + # An SRV client rebuilds its ClientOptions at first use (_resolve_srv); + # the rebuilt options must keep the tracing values resolved at + # construction rather than re-reading the (possibly changed) environment. + with ( + patch("pymongo.synchronous.srv_resolver._SrvResolver.get_hosts") as mock_hosts, + patch("pymongo.synchronous.srv_resolver._SrvResolver.get_options", return_value=None), + ): + mock_hosts.return_value = [("localhost", 27017)] + with patch.dict(os.environ, {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED": "true"}): + client = self.unmanaged_simple_client( + "mongodb+srv://test.example.com", connect=False + ) + self.addCleanup(client.close) + self.assertEqual(client.options.tracing["enabled"], True) + # The environment changes after construction; the snapshot must not. + os.environ["OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED"] = "false" + client._get_topology() + self.assertEqual(client.options.tracing["enabled"], True) + + def test_query_text_truncation_shrinks_oversized_field_values(self): + client = self.rs_or_single_client(tracing={"enabled": True, "query_text_max_length": 200}) + coll = client[self.db.name].test_otel + coll.drop() + self.exporter.clear() + coll.insert_one({"x": "a" * 500}) + + spans = self.spans("insert") + self.assertEqual(len(spans), 1) + query_text = spans[0].attributes["db.query.text"] + # The oversized field value must be truncated at the field level (not + # just a blind cut of the fully-serialized string), and the result must + # never exceed the configured bound, even when a "..." marker is added. + self.assertLessEqual(len(query_text), 200) + self.assertNotIn("a" * 500, query_text) + + def test_query_text_tiny_max_length_truncates_without_suffix(self): + # Budgets smaller than the "..." marker must truncate without it so + # the result still honors the configured bound. + for max_length in (1, 2): + text = _otel._build_query_text({"ping": 1}, max_length) + self.assertEqual(len(text), max_length) + + def test_is_sensitive_command_case_insensitive(self): + # Redaction normalizes the command name, mirroring the comparison in + # command monitoring: a differently-cased sensitive command gets no + # span, so its payload cannot leak through db.query.text. + for name in ("saslStart", "SASLSTART", "saslstart", "CreateUser", "createuser"): + self.assertTrue(_otel._is_sensitive_command(name, False)) + self.assertFalse(_otel._is_sensitive_command("find", False)) + + def test_no_client_options_is_never_traced(self): + # ``None`` tracing options mean no client context (connection + # handshakes, server monitoring): never traced, even when the + # environment variable enables driver-level tracing. + with patch.dict(os.environ, {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED": "true"}): + self.assertFalse(_otel._is_tracing_enabled(None)) + + # Client-level bulk_write requires MongoDB 8.0+ (wire version 25). + @client_context.require_version_min(8, 0, 0, -24) + def test_unacknowledged_bulk_write_query_text_includes_documents(self): + # db.query.text is built from the document published in + # CommandStartedEvent, which carries the write documents that the + # wire command omits for unacknowledged bulk writes. + client = self.rs_or_single_client( + tracing={"enabled": True, "query_text_max_length": 1000}, w=0 + ) + self.exporter.clear() + client.bulk_write( + [InsertOne(namespace=f"{self.db.name}.test_otel", document={"x": 1})], ordered=False + ) + (span,) = self.spans("bulkWrite") + self.assertIn('"x": 1', span.attributes["db.query.text"]) + + @client_context.require_failCommand_blockConnection + @client_context.require_async + def test_span_ended_on_task_cancellation(self): + # Task cancellation raises CancelledError (a BaseException), which the + # command runner's cleanup must handle: the span ends with an error + # status and the cancellation still propagates. + client = self.rs_or_single_client(tracing={"enabled": True}) + coll = client[self.db.name].test_otel + coll.drop() + coll.insert_many([{"x": i} for i in range(5)]) + self.exporter.clear() + + fail_command = { + "configureFailPoint": "failCommand", + "mode": "alwaysOn", + "data": {"failCommands": ["getMore"], "blockConnection": True, "blockTimeMS": 5000}, + } + + def task(): + cursor = coll.find({}, batch_size=1) + cursor.next() + with self.fail_point(fail_command): + cursor.next() + + running = asyncio.create_task(task()) + time.sleep(0.1) + start = time.monotonic() + running.cancel() + with self.assertRaises(asyncio.CancelledError): + running + # The cancellation surfaces once the failPoint's block on the + # connection releases; it must not hang indefinitely. + self.assertLess(time.monotonic() - start, 7) + + (span,) = self.spans("getMore") + self.assertEqual(span.status.status_code, trace.StatusCode.ERROR) + + +# TODO(PYTHON-5947): superseded once the unified test format's +# expectTracingMessages/observeTracingMessages tests exercise this validator +# indirectly through real client construction; remove this class then. +class TestValidateTracingOrNone(unittest.TestCase): + def test_none(self): + self.assertIsNone(common.validate_tracing_or_none("tracing", None)) + + def test_defaults(self): + self.assertEqual( + common.validate_tracing_or_none("tracing", {}), + {"enabled": None, "query_text_max_length": None}, + ) + + def test_explicit_enabled_false_preserved(self): + # False must stay distinct from "unset" (None) so it can override the + # environment variable instead of deferring to it. + result = common.validate_tracing_or_none("tracing", {"enabled": False}) + self.assertIs(result["enabled"], False) + + def test_enabled_and_query_text_max_length(self): + self.assertEqual( + common.validate_tracing_or_none( + "tracing", {"enabled": True, "query_text_max_length": 500} + ), + {"enabled": True, "query_text_max_length": 500}, + ) + + def test_explicit_zero_query_text_max_length_preserved(self): + # 0 must stay distinct from "unset" (None) so it can override the + # environment variable instead of being treated as not configured. + result = common.validate_tracing_or_none( + "tracing", {"enabled": True, "query_text_max_length": 0} + ) + self.assertEqual(result["query_text_max_length"], 0) + + def test_rejects_non_mapping(self): + with self.assertRaises(TypeError): + common.validate_tracing_or_none("tracing", "enabled") + + def test_rejects_unknown_option(self): + with self.assertRaisesRegex(ConfigurationError, "Unknown tracing option"): + common.validate_tracing_or_none("tracing", {"bogus": True}) + + def test_rejects_non_boolean_enabled(self): + with self.assertRaises(TypeError): + common.validate_tracing_or_none("tracing", {"enabled": "yes"}) + + def test_rejects_non_integer_query_text_max_length(self): + with self.assertRaises(TypeError): + common.validate_tracing_or_none("tracing", {"query_text_max_length": [1]}) + + def test_rejects_negative_query_text_max_length(self): + with self.assertRaises(ValueError): + common.validate_tracing_or_none("tracing", {"query_text_max_length": -1}) + + def test_numeric_string_query_text_max_length_converted(self): + # The validator's converted value must be kept: a numeric string that + # passes validation cannot flow into max(0, value) at command time. + result = common.validate_tracing_or_none("tracing", {"query_text_max_length": "100"}) + self.assertEqual(result["query_text_max_length"], 100) + + +class TestResolveTracingOptions(unittest.TestCase): + """The client's ``tracing`` option is resolved against the environment once, at construction.""" + + def _without_otel_env(self): + env = os.environ.copy() + env.pop("OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED", None) + env.pop("OTEL_PYTHON_INSTRUMENTATION_MONGODB_QUERY_TEXT_MAX_LENGTH", None) + return env + + def test_none_without_env(self): + with patch.dict(os.environ, self._without_otel_env(), clear=True): + self.assertEqual( + _otel._resolve_tracing_options(None), + {"enabled": False, "query_text_max_length": 0}, + ) + + # The environment-deferral semantics only exist when opentelemetry is + # importable; without it _resolve_tracing_options always disables tracing. + @unittest.skipUnless(_otel._HAS_OPENTELEMETRY, "opentelemetry is not installed") + def test_unset_defers_to_env(self): + with patch.dict(os.environ, {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED": "true"}): + self.assertTrue( + _otel._resolve_tracing_options({"enabled": None, "query_text_max_length": None})[ + "enabled" + ] + ) + + def test_env_false(self): + with patch.dict(os.environ, {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED": "false"}): + self.assertFalse( + _otel._resolve_tracing_options({"enabled": None, "query_text_max_length": None})[ + "enabled" + ] + ) + + # Explicit values only win over the environment when tracing is available. + @unittest.skipUnless(_otel._HAS_OPENTELEMETRY, "opentelemetry is not installed") + def test_explicit_enabled_overrides_env(self): + env = {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED": "true"} + with patch.dict(os.environ, env): + resolved = _otel._resolve_tracing_options( + {"enabled": False, "query_text_max_length": None} + ) + self.assertIs(resolved["enabled"], False) + resolved = _otel._resolve_tracing_options( + {"enabled": True, "query_text_max_length": None} + ) + self.assertIs(resolved["enabled"], True) + + # The environment-deferral semantics only exist when opentelemetry is + # importable; without it _resolve_tracing_options always omits query text. + @unittest.skipUnless(_otel._HAS_OPENTELEMETRY, "opentelemetry is not installed") + def test_query_text_max_length_from_env(self): + with patch.dict( + os.environ, {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_QUERY_TEXT_MAX_LENGTH": "1024"} + ): + self.assertEqual( + _otel._resolve_tracing_options({"enabled": None, "query_text_max_length": None})[ + "query_text_max_length" + ], + 1024, + ) + + def test_explicit_zero_query_text_overrides_env(self): + env = {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_QUERY_TEXT_MAX_LENGTH": "1024"} + with patch.dict(os.environ, env): + resolved = _otel._resolve_tracing_options({"enabled": None, "query_text_max_length": 0}) + self.assertEqual(resolved["query_text_max_length"], 0) + + def test_invalid_env_query_text_is_zero(self): + with patch.dict( + os.environ, {"OTEL_PYTHON_INSTRUMENTATION_MONGODB_QUERY_TEXT_MAX_LENGTH": "bogus"} + ): + self.assertEqual( + _otel._resolve_tracing_options({"enabled": None, "query_text_max_length": None})[ + "query_text_max_length" + ], + 0, + ) + + def test_negative_query_text_clamped(self): + resolved = _otel._resolve_tracing_options({"enabled": None, "query_text_max_length": -5}) + self.assertEqual(resolved["query_text_max_length"], 0) + + +class TestOTelTracerCaching(unittest.TestCase): + """Regression test for the tracer-caching implementation in ``pymongo/_otel.py``. + + ``opentelemetry.trace.get_tracer()`` must only be called once, at import + time (cached as module-level ``_otel._TRACER``). Calling it per command + allocates two objects, takes a process-wide lock, and mutates the global + ``warnings`` filter list on every call, even on a cache hit. + """ + + @unittest.skipUnless(_otel._HAS_OPENTELEMETRY, "opentelemetry is not installed") + def test_start_command_span_does_not_call_get_tracer(self): + class _FakeConn: + id = 1 + server_connection_id: Optional[int] = None + address: _Address = ("localhost", 27017) + service_id = None + _otel_connection_attributes: tuple[Optional[int], dict[str, Any]] = (None, {}) + + with patch.object(_otel, "trace") as mock_trace: + for _ in range(3): + span = _otel.start_command_span( + {"enabled": True, "query_text_max_length": None}, + _FakeConn(), + {"ping": 1}, + "admin", + "ping", + False, + ) + self.assertIsNotNone(span) + span.end() + + mock_trace.get_tracer.assert_not_called() + + +if __name__ == "__main__": + unittest.main() diff --git a/tools/otel_bench.py b/tools/otel_bench.py new file mode 100755 index 0000000000..afa4666899 --- /dev/null +++ b/tools/otel_bench.py @@ -0,0 +1,410 @@ +#!/usr/bin/env python +"""Benchmark the overhead of the driver's OpenTelemetry instrumentation. + +Runs the DriverBench "Small doc insertOne" and "Find one by ID" tasks from +``test/performance`` (both the synchronous and the asynchronous API) under +each tracing configuration, interleaved with a rotated configuration order +across repetitions on a single host, and reports the overhead of each +configuration relative to the untraced baseline. + +Implements the benchmarking requirements of the OpenTelemetry +specification's Performance Implications section (DRIVERS-3620): +https://github.com/mongodb/specifications/pull/1986 + +Configurations (per the spec's table; the harness, not the driver, installs +and configures the SDK and its sampler, and the SDK configurations install +no exporter or span processor. No configuration has an active parent span): + +- ``off`` tracing disabled; the baseline. +- ``api-only`` tracing enabled; no SDK, so every tracing call is a no-op. +- ``sdk-ratio`` tracing enabled; SDK installed; ``TraceIdRatioBased(1%)`` sampler. +- ``sdk-always`` tracing enabled; SDK installed; every span sampled. +- ``main`` the commit before OpenTelemetry support was added, tracing + disabled (the spec's one-time pre-OTel comparison). Requires + ``--worktree-main`` pointing at a worktree of that commit. + +Requires a MongoDB server on localhost:27017 (or via DB_IP/DB_PORT), the +DriverBench datasets (``--data-dir``), and ``opentelemetry-sdk`` installed +in the current environment (the SDK is benchmark tooling only, the driver +itself depends only on ``opentelemetry-api``). + +Examples:: + + python tools/otel_bench.py --verify + python tools/otel_bench.py --fast --reps 1 + python tools/otel_bench.py --worktree-main /path/to/main-worktree +""" + +from __future__ import annotations + +import argparse +import json +import os +import statistics +import subprocess +import sys +import time +from collections import deque +from pathlib import Path +from typing import Any, Optional + +ROOT = Path(__file__).resolve().parent.parent + +# The two micro-benchmarks the spec requires, defined in both perf_test.py +# and async_perf_test.py. +TASKS = ("SmallDocInsertOne", "FindOneByID") + +OFF = "off" +API_ONLY = "api-only" +SDK_RATIO = "sdk-ratio" +SDK_ALWAYS = "sdk-always" +MAIN = "main" + +SDK_CONFIGS = frozenset({SDK_RATIO, SDK_ALWAYS}) +ENABLED_CONFIGS = frozenset({API_ONLY, SDK_RATIO, SDK_ALWAYS}) + +BASE_CONFIGS = (OFF, API_ONLY, SDK_RATIO, SDK_ALWAYS, MAIN) + +# Child pytest invocation; `-m perf` overrides the repo config's +# `-m default or default_async` addopts (the last -m wins). +PYTEST_ARGS = ["-m", "perf", "-v", "--durations=5"] + +# opentelemetry.sdk is imported only for the SDK configurations. The provider +# is installed before pytest runs; the driver's cached ProxyTracer delegates +# to it regardless of import order. No span processor is added. +BOOTSTRAP = """import sys +config = sys.argv[1] +del sys.argv[1] +if config in ("sdk-ratio", "sdk-always"): + from opentelemetry import trace + from opentelemetry.sdk.trace import TracerProvider + from opentelemetry.sdk.trace.sampling import ALWAYS_ON, TraceIdRatioBased + + sampler = TraceIdRatioBased(0.01) if config == "sdk-ratio" else ALWAYS_ON + trace.set_tracer_provider(TracerProvider(sampler=sampler)) +import pytest + +sys.exit(pytest.main(sys.argv[1:])) +""" + + +def parse_args() -> argparse.Namespace: + parser = argparse.ArgumentParser(description=__doc__.splitlines()[0]) + parser.add_argument("--reps", type=int, default=5, help="Number of interleaved repetitions") + parser.add_argument( + "--fast", + action="store_true", + help="Set FASTBENCH=1 in the benchmark children (quick, unrepresentative)", + ) + parser.add_argument( + "--data-dir", + type=Path, + default=ROOT / "specifications" / "source" / "benchmarking" / "data", + help="Directory containing the DriverBench datasets", + ) + parser.add_argument( + "--worktree-main", + type=Path, + help="Worktree of the commit before OpenTelemetry support was added, " + "for the spec's one-time pre-OTel comparison (the 'main' configuration)", + ) + parser.add_argument( + "--output-dir", + type=Path, + default=ROOT / "otel-bench-results", + help="Directory for results and logs", + ) + parser.add_argument("--configs", help="Comma-separated subset of: " + ",".join(BASE_CONFIGS)) + parser.add_argument( + "--python", default=sys.executable, help="Python interpreter for the benchmark children" + ) + parser.add_argument( + "--verify", + action="store_true", + help="Only check that tracing produces spans end-to-end (with an in-memory exporter) and exit", + ) + return parser.parse_args() + + +def resolve_configs(args: argparse.Namespace) -> tuple[str, ...]: + configs = list(BASE_CONFIGS) if args.worktree_main else [c for c in BASE_CONFIGS if c != MAIN] + if args.configs: + selected = args.configs.split(",") + unknown = set(selected) - set(configs) + if unknown: + raise SystemExit(f"Unknown or unavailable configurations: {sorted(unknown)}") + configs = selected + if OFF not in configs: + raise SystemExit("The 'off' configuration is required as the baseline") + return tuple(configs) + + +def build_command(config: str, api: str, python: str) -> list[str]: + perf_file = ( + "test/performance/perf_test.py" if api == "sync" else "test/performance/async_perf_test.py" + ) + nodes = [f"{perf_file}::Test{task}" for task in TASKS] + if config in SDK_CONFIGS: + return [python, "-c", BOOTSTRAP, config, *PYTEST_ARGS, *nodes] + return [python, "-m", "pytest", *PYTEST_ARGS, *nodes] + + +def child_env(config: str, output_file: Path, args: argparse.Namespace) -> dict[str, str]: + # The harness owns the tracing configuration: strip any ambient OTEL_ + # variables, then set exactly what the configuration requires. + env = {k: v for k, v in os.environ.items() if not k.startswith("OTEL_")} + env["TEST_PATH"] = str(args.data_dir) + env["OUTPUT_FILE"] = str(output_file) + env["PERF_CPU_TIME"] = "1" + if args.fast: + env["FASTBENCH"] = "1" + if config in ENABLED_CONFIGS: + env["OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED"] = "1" + return env + + +def run_one(config: str, api: str, rep: int, args: argparse.Namespace) -> tuple[Path, float]: + output_file = args.output_dir / f"{api}-{config}-rep{rep:02d}.json" + if output_file.exists(): + output_file.unlink() + cmd = build_command(config, api, args.python) + cwd = args.worktree_main if config == MAIN else ROOT + log_file = output_file.with_suffix(".log") + start = time.monotonic() + with open(log_file, "w") as log: + proc = subprocess.run( # noqa: S603 + cmd, + cwd=cwd, + env=child_env(config, output_file, args), + stdout=log, + stderr=subprocess.STDOUT, + check=False, + ) + elapsed = time.monotonic() - start + if proc.returncode != 0: + print(f"FAILED (exit {proc.returncode}): {' '.join(cmd)}\nLog tail:") + print("\n".join(log_file.read_text().splitlines()[-40:])) + raise SystemExit(1) + return output_file, elapsed + + +def parse_results(output_file: Path) -> dict[str, dict[str, float]]: + entries = json.loads(output_file.read_text()) + parsed: dict[str, dict[str, float]] = {} + for entry in entries: + name = entry["info"]["test_name"] + parsed[name] = {metric["name"]: metric["value"] for metric in entry["metrics"]} + return parsed + + +def verify() -> None: + """Check end-to-end that tracing produces spans, using an in-memory exporter. + + This is a wiring check only; measured runs never install an exporter. + """ + os.environ["OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED"] = "1" + from opentelemetry import trace + from opentelemetry.sdk.trace import TracerProvider + from opentelemetry.sdk.trace.export import SimpleSpanProcessor + + try: + from opentelemetry.sdk.trace.export.in_memory_span_exporter import InMemorySpanExporter + except ImportError: + # The module name changed across SDK versions. + from opentelemetry.sdk.trace.export.in_memory_exporter import ( # type: ignore[import-not-found,no-redef] + InMemorySpanExporter, + ) + + provider = TracerProvider() + exporter = InMemorySpanExporter() + provider.add_span_processor(SimpleSpanProcessor(exporter)) + trace.set_tracer_provider(provider) + + sys.path.insert(0, str(ROOT)) + from pymongo import MongoClient + + client: MongoClient[dict[str, Any]] = MongoClient() + client.perftest_otel_verify.coll.insert_one({"x": 1}) + client.drop_database("perftest_otel_verify") + client.close() + + spans = exporter.get_finished_spans() + names = [span.name for span in spans] + assert "insert" in names, f"Expected an 'insert' span, got: {names}" + span = next(s for s in spans if s.name == "insert") + attrs = dict(span.attributes or {}) + assert attrs.get("db.system.name") == "mongodb", attrs + assert attrs.get("db.command.name") == "insert", attrs + # The operation span nests the command span beneath it. + op_span = next(s for s in spans if s.name == "insert perftest_otel_verify.coll") + assert span.parent is not None + assert span.parent.span_id == op_span.context.span_id, names + assert op_span.attributes.get("db.operation.name") == "insert", dict(op_span.attributes or {}) + print(f"OK: {len(names)} span(s) recorded: {names}") + + +def report( + configs: tuple[str, ...], + scores: dict[tuple[str, str, str], list[float]], + cpu_scores: dict[tuple[str, str, str], list[float]], + args: argparse.Namespace, +) -> dict[str, Any]: + num_docs = 1000 if args.fast else 10000 + summary: dict[str, Any] = { + "reps": args.reps, + "fast": args.fast, + "configs": {}, + "pre_otel_compare": {}, + } + + for api in ("sync", "async"): + print(f"\n## {api} (median MB/s across repetitions, overhead vs 'off')") + print("| config | " + " | ".join(TASKS) + " |") + print("|" + "---|" * (len(TASKS) + 1)) + off_medians = { + task: statistics.median(scores[(OFF, api, task)]) + for task in TASKS + if scores.get((OFF, api, task)) + } + for config in configs: + if config == MAIN: + continue + cells = [] + for task in TASKS: + values = scores.get((config, api, task), []) + if not values: + cells.append("n/a") + continue + median = statistics.median(values) + if config == OFF: + cells.append(f"{median:.2f} MB/s (baseline)") + else: + overhead = (1 - median / off_medians[task]) * 100 + cells.append(f"{median:.2f} MB/s ({overhead:+.1f}%)") + print(f"| {config} | " + " | ".join(cells) + " |") + + if MAIN in configs and all(scores.get((MAIN, api, task)) for task in TASKS): + row: dict[str, Any] = {} + for task in TASKS: + main_median = statistics.median(scores[(MAIN, api, task)]) + off_median = statistics.median(scores[(OFF, api, task)]) + row[task] = { + "main_median_mb_s": main_median, + "off_median_mb_s": off_median, + "overhead_pct": (1 - off_median / main_median) * 100, + } + # Key by api: both rows must survive into the persisted results. + summary["pre_otel_compare"][api] = row + overheads = ", ".join(f"{task}: {row[task]['overhead_pct']:+.1f}%" for task in TASKS) + print(f"\nPre-OTel comparison (off on this branch vs main, {api}): {overheads}") + + print("\n## CPU time per operation (us/op median, delta vs 'off' in parens)") + print( + "| config | " + + " | ".join(f"{api}/{task}" for api in ("sync", "async") for task in TASKS) + + " |" + ) + print("|" + "---|" * (len(TASKS) * 2 + 1)) + for config in configs: + if config == MAIN: + continue + cells = [] + for api in ("sync", "async"): + for task in TASKS: + values = cpu_scores.get((config, api, task), []) + base = cpu_scores.get((OFF, api, task), []) + if not values or not base: + cells.append("n/a") + continue + per_op = statistics.median(values) / num_docs * 1e6 + per_op_off = statistics.median(base) / num_docs * 1e6 + delta = "" if config == OFF else f" ({per_op - per_op_off:+.0f})" + cells.append(f"{per_op:.0f}{delta}") + print(f"| {config} | " + " | ".join(cells) + " |") + + for config in configs: + summary["configs"][config] = {} + for api in ("sync", "async"): + for task in TASKS: + values = scores.get((config, api, task), []) + if values: + entry: dict[str, Any] = { + "mb_per_sec_reps": values, + "mb_per_sec_median": statistics.median(values), + } + cpu = cpu_scores.get((config, api, task), []) + if cpu: + entry["cpu_seconds_per_iteration_reps"] = cpu + entry["cpu_us_per_op_median"] = statistics.median(cpu) / num_docs * 1e6 + summary["configs"][config][f"{api}/{task}"] = entry + return summary + + +def main() -> None: + args = parse_args() + if args.verify: + verify() + return + + if args.reps < 1: + raise SystemExit("--reps must be at least 1") + # Children run with a different working directory (the repo root, or the + # main worktree); interpret user-supplied relative paths against the + # harness's cwd once, so the harness's own file operations and the paths + # passed to children in the environment agree. + args.data_dir = args.data_dir.absolute() + args.output_dir = args.output_dir.absolute() + if not args.data_dir.exists(): + raise SystemExit( + f"Dataset directory not found: {args.data_dir}\n" + "Clone the specifications repo and extract the data tarballs, or pass --data-dir." + ) + configs = resolve_configs(args) + args.output_dir.mkdir(parents=True, exist_ok=True) + + scores: dict[tuple[str, str, str], list[float]] = {} + cpu_scores: dict[tuple[str, str, str], list[float]] = {} + started = time.monotonic() + total_runs = args.reps * len(configs) * 2 + done_runs = 0 + summary: Optional[dict[str, Any]] = None + + for rep in range(1, args.reps + 1): + # Rotate the configuration order each repetition so host drift affects + # all configurations equally (the spec's interleaving requirement). + order: deque[str] = deque(configs) + order.rotate(-(rep - 1)) + for config in order: + for api in ("sync", "async"): + output_file, elapsed = run_one(config, api, rep, args) + parsed = parse_results(output_file) + for task in TASKS: + metrics = parsed.get(task, {}) + mbps = metrics.get("megabytes_per_sec") + if mbps is not None: + scores.setdefault((config, api, task), []).append(mbps) + if "cpu_time_median" in metrics: + cpu_scores.setdefault((config, api, task), []).append( + metrics["cpu_time_median"] + ) + done_runs += 1 + results_mb = ", ".join( + f"{task}={parsed[task]['megabytes_per_sec']:.2f}MB/s" for task in TASKS + ) + print( + f"[{done_runs}/{total_runs}] rep {rep}/{args.reps} {config:>9} {api:<5} " + f"({elapsed:.0f}s): {results_mb}", + flush=True, + ) + # Crash-safe: persist intermediate results after every repetition. + summary = report(configs, scores, cpu_scores, args) + (args.output_dir / "otel_bench_results.json").write_text(json.dumps(summary, indent=2)) + + assert summary is not None + print(f"\nTotal: {time.monotonic() - started:.0f}s. Results in {args.output_dir}/") + if summary["pre_otel_compare"]: + print(json.dumps(summary["pre_otel_compare"], indent=2)) + + +if __name__ == "__main__": + main() diff --git a/tools/synchro.py b/tools/synchro.py index b5c6b2873a..2060b2a448 100644 --- a/tools/synchro.py +++ b/tools/synchro.py @@ -140,6 +140,7 @@ "async_simple_test_client": "simple_test_client", "_async_create_connection": "_create_connection", "pymongo.asynchronous.srv_resolver._SrvResolver.get_hosts": "pymongo.synchronous.srv_resolver._SrvResolver.get_hosts", + "pymongo.asynchronous.srv_resolver._SrvResolver.get_options": "pymongo.synchronous.srv_resolver._SrvResolver.get_options", "dns.asyncresolver.resolve": "dns.resolver.resolve", "__aenter__": "__enter__", "__aexit__": "__exit__", diff --git a/uv.lock b/uv.lock index 577a943912..1b59238b32 100644 --- a/uv.lock +++ b/uv.lock @@ -1246,6 +1246,18 @@ wheels = [ { url = "https://files.pythonhosted.org/packages/54/c8/12811c9b48fde162bb72b6f2e78fada9a247a09d7bb5be2050a5d099c77b/nodeenv-1.11.0-py2.py3-none-any.whl", hash = "sha256:edaa16e6c14d7cf395d75d4bbd5a26390f4dc06501a33b4e76282b02cc688a25", size = 34277, upload-time = "2026-09-26T11:29:19.933Z" }, ] +[[package]] +name = "opentelemetry-api" +version = "1.45.0" +source = { registry = "https://pypi.org/simple" } +dependencies = [ + { name = "typing-extensions" }, +] +sdist = { url = "https://files.pythonhosted.org/packages/1f/dc/e12c1fe1ed8a7b7149777127b1a0e12ce5bd5a81d97408bedc2128c260f5/opentelemetry_api-1.45.0.tar.gz", hash = "sha256:711ede81773c8025c2c03dac0450bc89f3d30aea6eabcc815c570d4e35a963f7", size = 72115, upload-time = "2026-09-25T12:32:28.399Z" } +wheels = [ + { url = "https://files.pythonhosted.org/packages/44/b9/040d1a1c7836922828e6480cd2366bb8fe0ebf75b413d2bb51a9b0e7f78f/opentelemetry_api-1.45.0-py3-none-any.whl", hash = "sha256:80e068aba7cd56c8b58512d6a36f8d25cb1dfaa0c0a4cc1c938ccf9f362d9cb3", size = 60020, upload-time = "2026-09-25T12:32:03.192Z" }, +] + [[package]] name = "packaging" version = "26.3" @@ -1341,6 +1353,9 @@ ocsp = [ { name = "requests" }, { name = "service-identity" }, ] +opentelemetry = [ + { name = "opentelemetry-api" }, +] snappy = [ { name = "python-snappy" }, ] @@ -1392,6 +1407,7 @@ requires-dist = [ { name = "dnspython", specifier = ">=2.7.0,<3.0.0" }, { name = "furo", marker = "extra == 'docs'", specifier = "==2025.12.19" }, { name = "importlib-metadata", marker = "python_full_version < '3.13' and extra == 'test'", specifier = ">=7.0" }, + { name = "opentelemetry-api", marker = "extra == 'opentelemetry'", specifier = ">=1.20.0" }, { name = "pykerberos", marker = "os_name != 'nt' and extra == 'gssapi'", specifier = ">=1.2.4" }, { name = "pymongo-auth-aws", marker = "extra == 'aws'", specifier = ">=1.3.0,<2.0.0" }, { name = "pymongo-auth-aws", marker = "extra == 'encryption'", specifier = ">=1.3.0,<2.0.0" }, @@ -1409,7 +1425,7 @@ requires-dist = [ { name = "sphinxcontrib-shellcheck", marker = "extra == 'docs'", specifier = ">=1.1.2,<2" }, { name = "winkerberos", marker = "os_name == 'nt' and extra == 'gssapi'", specifier = ">=0.12.2" }, ] -provides-extras = ["aws", "docs", "encryption", "gssapi", "ocsp", "snappy", "test", "zstd"] +provides-extras = ["aws", "docs", "encryption", "gssapi", "ocsp", "opentelemetry", "snappy", "test", "zstd"] [package.metadata.requires-dev] coverage = [{ name = "coverage", extras = ["toml"], specifier = ">=5,<=7.15.0" }]