diff --git a/.evergreen/generated_configs/variants.yml b/.evergreen/generated_configs/variants.yml index b167508539..5fe79df1f0 100644 --- a/.evergreen/generated_configs/variants.yml +++ b/.evergreen/generated_configs/variants.yml @@ -441,6 +441,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: + - rhel87-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 6dfa014a16..d250580ce6 100644 --- a/.evergreen/scripts/generate_config.py +++ b/.evergreen/scripts/generate_config.py @@ -451,6 +451,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 e188dcaa9d..ccfcfafc25 100644 --- a/.evergreen/scripts/setup_tests.py +++ b/.evergreen/scripts/setup_tests.py @@ -42,6 +42,7 @@ "enterprise_auth": "gssapi", "kms": "encryption", "ocsp": "ocsp", + "otel": "opentelemetry", "pyopenssl": "ocsp", } @@ -439,6 +440,11 @@ 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). + UV_ARGS.append("--with opentelemetry-sdk") + if test_name == "perf": data_dir = ROOT / "specifications/source/benchmarking/data" if not data_dir.exists(): @@ -460,9 +466,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 914cd9ac60..4b1252ccca 100644 --- a/.evergreen/scripts/utils.py +++ b/.evergreen/scripts/utils.py @@ -43,6 +43,7 @@ class Distro: "load_balancer": "load_balancer", "mockupdb": "mockupdb", "ocsp": "ocsp", + "otel": "otel", "perf": "perf", "numpy": "", } diff --git a/doc/changelog.rst b/doc/changelog.rst index aa4feb0af9..de33220b31 100644 --- a/doc/changelog.rst +++ b/doc/changelog.rst @@ -17,6 +17,12 @@ Changes in Version 4.18.0 attempts, so consumers can correlate a retried operation's events. As a result, ``operation_id`` is no longer equal to the per-attempt ``request_id`` for these operations. +- 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 f6f10ca5a9..8b6f485056 100644 --- a/justfile +++ b/justfile @@ -5,7 +5,7 @@ set shell := ["bash", "-c"] export UV_NO_LOCK := "1" # 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..713f963bbc --- /dev/null +++ b/pymongo/_otel.py @@ -0,0 +1,268 @@ +# 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 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__) +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. + + ``query_text_max_length`` is None when the client didn't configure it, so + the environment variable can be consulted; any explicit value (including + 0, to force ``db.query.text`` off) overrides the environment variable. + """ + + enabled: 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"}) + +# 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" + + +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 _is_tracing_enabled(tracing_options: Optional[TracingOptions]) -> bool: + """Return True if OTel command spans should be created for this client. + + The ``MongoClient`` ``tracing.enabled`` option and the + ``OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED`` environment variable both + gate enablement; either one being truthy is sufficient. + """ + if not _HAS_OPENTELEMETRY: + return False + if tracing_options and tracing_options.get("enabled"): + return True + return _env_truthy(_OTEL_ENABLED_ENV) + + +def _get_query_text_max_length(tracing_options: Optional[TracingOptions]) -> int: + """Return the configured db.query.text truncation length, or 0 to omit the attribute. + + An explicit client value (including 0) always wins; the environment + variable is only consulted when the client didn't configure it at all. + """ + client_value = tracing_options.get("query_text_max_length") if tracing_options else None + if client_value is not None: + return max(0, client_value) + try: + return max(0, int(os.getenv(_OTEL_QUERY_TEXT_MAX_LENGTH_ENV, "0"))) + except ValueError: + return 0 + + +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: + suffix = "..." + text = text[: max(0, 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 == _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``.""" + if command_name in _SENSITIVE_COMMANDS: + return True + return command_name in _HELLO_COMMANDS 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). + """ + 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) + address = conn.address + transport = "unix" if address[1] is None else "tcp" + attributes: dict[str, Any] = { + "db.system.name": "mongodb", + "db.namespace": dbname, + "db.command.name": command_name, + "db.query.summary": _build_query_summary(command_name, dbname, collection), + "server.address": address[0], + "network.transport": transport, + "db.mongodb.driver_connection_id": conn.id, + } + if address[1] is not None: + attributes["server.port"] = address[1] + if collection: + attributes["db.collection.name"] = collection + if conn.server_connection_id is not None: + attributes["db.mongodb.server_connection_id"] = conn.server_connection_id + lsid = cmd.get("lsid") + if isinstance(lsid, Mapping): + formatted_lsid = _format_lsid(lsid) + if formatted_lsid is not None: + attributes["db.mongodb.lsid"] = formatted_lsid + txn_number = cmd.get("txnNumber") + if txn_number is not None: + attributes["db.mongodb.txn_number"] = txn_number + max_query_text_length = _get_query_text_max_length(tracing_options) + if max_query_text_length > 0: + attributes["db.query.text"] = _build_query_text(cmd, max_query_text_length) + + assert _TRACER is not None # _is_tracing_enabled already checked _HAS_OPENTELEMETRY + return _TRACER.start_span(command_name, kind=SpanKind.CLIENT, attributes=attributes) + + +def end_command_span_success(span: Optional[Span], reply: _DocumentOut) -> None: + """Set the cursor id (if any) and end the span.""" + if span is None: + return + cursor = reply.get("cursor") + if isinstance(cursor, Mapping) and "id" in cursor: + span.set_attribute("db.mongodb.cursor_id", cursor["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.""" + if span is None: + return + span.record_exception(exc) + code = failure.get("code") + if code is not None: + span.set_attribute("db.response.status_code", str(code)) + span.set_status(Status(StatusCode.ERROR, description=failure.get("errmsg"))) + span.end() diff --git a/pymongo/_telemetry.py b/pymongo/_telemetry.py index f972a402df..b770add9d6 100644 --- a/pymongo/_telemetry.py +++ b/pymongo/_telemetry.py @@ -23,6 +23,8 @@ from collections.abc import MutableMapping from typing import TYPE_CHECKING, Any, Optional +from pymongo import _otel +from pymongo.errors import OperationFailure from pymongo.logger import ( _COMMAND_LOGGER, _CONNECTION_LOGGER, @@ -75,8 +77,12 @@ class _CommandTelemetry: "_publish", "_request_id", "_should_log", + "_span", + "_speculative_hello", "_start", "_topology_id", + "_tracing_enabled", + "_tracing_options", ) def __init__( @@ -88,11 +94,15 @@ def __init__( dbname: str, request_id: int, op_id: Optional[int], + tracing_options: Optional[_otel.TracingOptions] = None, + speculative_hello: bool = False, ) -> None: self._topology_id = topology_id self._should_log = topology_id is not None and _COMMAND_LOGGER.isEnabledFor(logging.DEBUG) self._publish = listeners is not None and listeners.enabled_for_commands - self._active = self._should_log or self._publish + self._tracing_options = tracing_options + self._tracing_enabled = _otel._is_tracing_enabled(tracing_options) + self._active = self._should_log or self._publish or self._tracing_enabled self._listeners = listeners self._conn = conn self._cmd = cmd @@ -100,6 +110,8 @@ def __init__( self._dbname = dbname self._request_id = request_id self._op_id = op_id + self._speculative_hello = speculative_hello + self._span: Optional[Any] = None self._start: datetime.datetime self._duration: datetime.timedelta @@ -121,7 +133,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 = datetime.datetime.now() if not self._active: return @@ -140,6 +152,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, + self._cmd, + self._dbname, + self._name, + self._speculative_hello, + ) @property def duration(self) -> datetime.timedelta: @@ -152,7 +173,7 @@ 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 = datetime.datetime.now() - self._start if not self._active: return @@ -177,14 +198,16 @@ def succeeded( speculative_hello=speculative_hello, database_name=self._dbname, ) + if self._span is not None: + _otel.end_command_span_success(self._span, 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 = datetime.datetime.now() - self._start if not self._active: return @@ -193,7 +216,7 @@ def failed( _CommandStatusMessage.FAILED, durationMS=self._duration, failure=failure, - isServerSideError=is_server_side_error, + isServerSideError=isinstance(exc, OperationFailure), ) if self._publish: assert self._listeners is not None @@ -208,6 +231,8 @@ def failed( 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/command_runner.py b/pymongo/asynchronous/command_runner.py index 683dc0d8e8..6f9cce77aa 100644 --- a/pymongo/asynchronous/command_runner.py +++ b/pymongo/asynchronous/command_runner.py @@ -162,7 +162,18 @@ async def _run_command( if op_id is None: op_id = _op_id.OP_ID.get() - telemetry = _CommandTelemetry(topology_id, conn, listeners, cmd, dbname, request_id, op_id) + tracing_options = client.options.tracing if client is not None else None + telemetry = _CommandTelemetry( + topology_id, + conn, + listeners, + cmd, + dbname, + request_id, + op_id, + tracing_options, + speculative_hello, + ) telemetry.started(orig, ensure_db) reply: Optional[_OpMsg] = None @@ -211,7 +222,7 @@ async def _run_command( failure: _DocumentOut = exc.details # type: ignore[assignment] else: failure = _convert_exception(exc) - telemetry.failed(failure, command_name, isinstance(exc, OperationFailure)) + telemetry.failed(failure, command_name, exc) raise telemetry.succeeded(docs[0], command_name, speculative_hello) diff --git a/pymongo/asynchronous/mongo_client.py b/pymongo/asynchronous/mongo_client.py index cd1cefea13..90151acf3e 100644 --- a/pymongo/asynchronous/mongo_client.py +++ b/pymongo/asynchronous/mongo_client.py @@ -615,8 +615,27 @@ def __init__( 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. Defaults to ``False``. Also controlled by the + ``OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED`` environment variable; either + being enabled is sufficient. + - ``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.18 + Added the ``tracing`` keyword argument. + .. versionchanged:: 4.17 Added the ``max_adaptive_retries`` and ``enable_overload_retargeting`` URI and keyword arguments. diff --git a/pymongo/client_options.py b/pymongo/client_options.py index 99a2bcb4e4..d2d33f6805 100644 --- a/pymongo/client_options.py +++ b/pymongo/client_options.py @@ -40,6 +40,7 @@ if TYPE_CHECKING: from bson.codec_options import CodecOptions + from pymongo import _otel from pymongo.auth_shared import MongoCredential from pymongo.encryption_options import AutoEncryptionOpts from pymongo.pyopenssl_context import SSLContext @@ -247,6 +248,10 @@ def __init__( if "enable_overload_retargeting" in options else options.get("enableoverloadretargeting", common.ENABLE_OVERLOAD_RETARGETING) ) + self.__tracing = cast( + "_otel.TracingOptions", + options.get("tracing") or {"enabled": False, "query_text_max_length": None}, + ) @property def _options(self) -> Mapping[str, Any]: @@ -374,3 +379,11 @@ 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. + + .. versionadded:: 4.18 + """ + return self.__tracing diff --git a/pymongo/common.py b/pymongo/common.py index 57606a6317..053ddded76 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 @@ -607,6 +608,26 @@ 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", False) + 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") + 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: @@ -784,6 +805,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/pool_shared.py b/pymongo/pool_shared.py index 410ffd8189..ca7bf9370b 100644 --- a/pymongo/pool_shared.py +++ b/pymongo/pool_shared.py @@ -55,7 +55,7 @@ class _ConnectionTelemetryInfo(Protocol): id: int server_connection_id: Optional[int] - address: tuple[str, int] + address: _Address service_id: Optional[ObjectId] diff --git a/pymongo/synchronous/command_runner.py b/pymongo/synchronous/command_runner.py index e81f514003..06be4e51a2 100644 --- a/pymongo/synchronous/command_runner.py +++ b/pymongo/synchronous/command_runner.py @@ -162,7 +162,18 @@ def _run_command( if op_id is None: op_id = _op_id.OP_ID.get() - telemetry = _CommandTelemetry(topology_id, conn, listeners, cmd, dbname, request_id, op_id) + tracing_options = client.options.tracing if client is not None else None + telemetry = _CommandTelemetry( + topology_id, + conn, + listeners, + cmd, + dbname, + request_id, + op_id, + tracing_options, + speculative_hello, + ) telemetry.started(orig, ensure_db) reply: Optional[_OpMsg] = None @@ -211,7 +222,7 @@ def _run_command( failure: _DocumentOut = exc.details # type: ignore[assignment] else: failure = _convert_exception(exc) - telemetry.failed(failure, command_name, isinstance(exc, OperationFailure)) + telemetry.failed(failure, command_name, exc) raise telemetry.succeeded(docs[0], command_name, speculative_hello) diff --git a/pymongo/synchronous/mongo_client.py b/pymongo/synchronous/mongo_client.py index d3f6a1e217..4f01e03a7e 100644 --- a/pymongo/synchronous/mongo_client.py +++ b/pymongo/synchronous/mongo_client.py @@ -616,8 +616,27 @@ def __init__( 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. Defaults to ``False``. Also controlled by the + ``OTEL_PYTHON_INSTRUMENTATION_MONGODB_ENABLED`` environment variable; either + being enabled is sufficient. + - ``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.18 + Added the ``tracing`` keyword argument. + .. versionchanged:: 4.17 Added the ``max_adaptive_retries`` and ``enable_overload_retargeting`` URI and keyword arguments. diff --git a/pyproject.toml b/pyproject.toml index 05fd0658d8..3e0b464dd8 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -97,6 +97,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"] @@ -144,6 +145,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..7eaeafe734 --- /dev/null +++ b/test/asynchronous/test_otel.py @@ -0,0 +1,388 @@ +# 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 os +import sys +from typing import 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.typings import _Address +from test.asynchronous import AsyncIntegrationTest, 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.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_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_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 + + 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, {"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) + self.assertTrue(any(event.name == "exception" for event in span.events)) + + 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_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) + + +# 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": False, "query_text_max_length": None}, + ) + + 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}) + + +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 + + 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/test_otel.py b/test/test_otel.py new file mode 100644 index 0000000000..d0e3a55fe5 --- /dev/null +++ b/test/test_otel.py @@ -0,0 +1,382 @@ +# 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 os +import sys +from typing import 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.typings import _Address +from test import IntegrationTest, 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.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_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_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 + + 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, {"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) + self.assertTrue(any(event.name == "exception" for event in span.events)) + + 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_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) + + +# 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": False, "query_text_max_length": None}, + ) + + 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}) + + +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 + + 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()