PYTHON-5947 Add OpenTelemetry Operation and Transaction Span Support

.evergreen/generated_configs/variants.yml

@@ -445,6 +445,7 @@ buildvariants:
   - name: otel-rhel8
     tasks:
       - name: .test-non-standard .standalone-noauth-nossl
+      - name: .test-non-standard .replica_set-noauth-ssl
     display_name: OTel RHEL8
     run_on:
       - rhel87-small

.evergreen/resync-specs.sh

@@ -127,6 +127,9 @@ do
       cpjson command-logging-and-monitoring/tests/logging command_logging
       cpjson command-logging-and-monitoring/tests/monitoring command_monitoring
       ;;
+    open-telemetry|otel|open_telemetry)
+      cpjson open-telemetry/tests open_telemetry
+      ;;
     crud|CRUD)
       cpjson crud/tests/ crud
       ;;

.evergreen/scripts/generate_config.py

@@ -457,7 +457,16 @@ def create_otel_variants():
     expansions = dict(TEST_NAME="otel", COVERAGE="1")
     return [
         create_variant(
-            [".test-non-standard .standalone-noauth-nossl"],
+            [
+                ".test-non-standard .standalone-noauth-nossl",
+                # Transaction spans (test/open_telemetry/transaction/*.json,
+                # test_otel.py's @require_transactions tests) need a replica set
+                # (they're skipped entirely on a standalone topology), so also
+                # run against one, mirroring how other variants in this file
+                # (e.g. PyOpenSSL's ".replica_set-noauth-ssl") pair a
+                # standalone/standard selector with a replica-set one.
+                ".test-non-standard .replica_set-noauth-ssl",
+            ],
             get_variant_name("OTel", host),
             host=host,
             tags=["pr"],

.gitignore

@@ -49,3 +49,6 @@ server.log
 
 # uv lockfiles
 uv.lock
+
+# Local git worktrees
+.worktrees/

bson/json_util.py

@@ -1109,20 +1109,18 @@ def _truncate_documents(obj: Any, max_length: int) -> tuple[Any, int]:
     if hasattr(obj, "items"):
         truncated: Any = {}
         for k, v in obj.items():
-            truncated_v, remaining = _truncate_documents(v, remaining)
-            if truncated_v:
-                truncated[k] = truncated_v
             if remaining <= 0:
                 break
+            truncated_v, remaining = _truncate_documents(v, remaining)
+            truncated[k] = truncated_v
         return truncated, remaining
     elif hasattr(obj, "__iter__") and not isinstance(obj, (str, bytes)):
         truncated: Any = []  # type:ignore[no-redef]
         for v in obj:
-            truncated_v, remaining = _truncate_documents(v, remaining)
-            if truncated_v:
-                truncated.append(truncated_v)
             if remaining <= 0:
                 break
+            truncated_v, remaining = _truncate_documents(v, remaining)
+            truncated.append(truncated_v)
         return truncated, remaining
     else:
         return _truncate(obj, remaining)

doc/changelog.rst

@@ -20,11 +20,14 @@ PyMongo 4.18 brings a number of changes including:
   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
+- Added optional OpenTelemetry tracing support, conforming to the
   `OpenTelemetry driver specification <https://github.com/mongodb/specifications/blob/master/source/open-telemetry/open-telemetry.md>`_.
-  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
+  Every public API call produces an operation span, which contains one span
+  per command sent to the server. Inside a transaction, those operation spans
+  nest under a ``transaction`` span. 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.
 - Fixed a potential out-of-bounds read in the C extension when decoding an
   array of BSON documents. An embedded document whose declared length exceeds

pymongo/_otel.py

@@ -21,8 +21,11 @@
 
 from __future__ import annotations
 
+import contextlib
 import os
-from collections.abc import Mapping, MutableMapping
+import traceback
+from collections.abc import Iterator, Mapping, MutableMapping
+from contextvars import ContextVar
 from typing import TYPE_CHECKING, Any, Optional, TypedDict
 
 from bson import json_util
@@ -31,7 +34,7 @@
 from pymongo.logger import _HELLO_COMMANDS, _JSON_OPTIONS, _SENSITIVE_COMMANDS
 
 try:
-    from opentelemetry import trace
+    from opentelemetry import context, trace
     from opentelemetry.trace import SpanKind, Status, StatusCode
 
     _HAS_OPENTELEMETRY = True
@@ -45,6 +48,14 @@
     _HAS_OPENTELEMETRY = False
     _TRACER = None
 
+# The operation name of whichever operation span is currently active (entered
+# by start_operation_span), so start_command_span can backfill the operation
+# span's name/namespace attributes from the first command executed inside it
+# (dbname/collection aren't known until then; see start_operation_span).
+_CURRENT_OPERATION_NAME: ContextVar[Optional[str]] = ContextVar(
+    "_CURRENT_OPERATION_NAME", default=None
+)
+
 if TYPE_CHECKING:
     from opentelemetry.trace import Span, Tracer
 
@@ -205,10 +216,27 @@ def start_command_span(
     """
     if not _is_tracing_enabled(tracing_options):
         return None
+
+    collection = _extract_collection_name(command_name, dbname, cmd)
+    # Backfill the ambient operation span's name/namespace/summary from the
+    # first command built inside it, before the sensitive-command early return
+    # below: the operation span still needs its (Required, per the OTel spec)
+    # db.namespace/db.operation.summary attributes even when the command
+    # itself is sensitive and gets no command span of its own.
+    current_operation = _CURRENT_OPERATION_NAME.get()
+    if current_operation is not None:
+        current_span = trace.get_current_span()
+        if current_span.is_recording():
+            summary = _build_query_summary(current_operation, dbname, collection)
+            current_span.update_name(summary)
+            current_span.set_attribute("db.namespace", dbname)
+            current_span.set_attribute("db.operation.summary", summary)
+            if collection:
+                current_span.set_attribute("db.collection.name", collection)
+
     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] = {
@@ -243,15 +271,40 @@ def start_command_span(
 
 
 def end_command_span_success(span: Optional[Span], reply: _DocumentOut) -> None:
-    """Set the cursor id (if any) and end the span."""
+    """Set the cursor id (if any open cursor) and end the span."""
     if span is None:
         return
     cursor = reply.get("cursor")
-    if isinstance(cursor, Mapping) and "id" in cursor:
+    if isinstance(cursor, Mapping) and cursor.get("id"):
+        # A cursor id of 0 means the cursor is already exhausted, i.e. there is
+        # no cursor left to track, so per the OTel spec ("If the command
+        # returns a cursor, or uses a cursor, the cursor_id attribute SHOULD
+        # be added") the attribute is only meaningful, and only added, when
+        # id is nonzero.
         span.set_attribute("db.mongodb.cursor_id", cursor["id"])
     span.end()
 
 
+def _set_exception_attributes(span: Span, exc: BaseException) -> None:
+    """Set exception.type/exception.message/exception.stacktrace span attributes.
+
+    ``span.record_exception`` only attaches these to an "exception" *event*,
+    but the OTel spec requires them as span *attributes* too ("drivers SHOULD
+    add the following attributes to the span"); mirror record_exception's own
+    formatting for consistency. Shared by the command-span and operation-span
+    failure paths, since the spec states the same requirement for both.
+    """
+    module = type(exc).__module__
+    qualname = type(exc).__qualname__
+    exception_type = f"{module}.{qualname}" if module and module != "builtins" else qualname
+    span.set_attribute("exception.type", exception_type)
+    span.set_attribute("exception.message", str(exc))
+    span.set_attribute(
+        "exception.stacktrace",
+        "".join(traceback.format_exception(type(exc), exc, exc.__traceback__)),
+    )
+
+
 def end_command_span_failure(
     span: Optional[Span],
     failure: _DocumentOut,
@@ -261,8 +314,190 @@ def end_command_span_failure(
     if span is None:
         return
     span.record_exception(exc)
+    _set_exception_attributes(span, 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()
+
+
+class _OperationSpanHandle:
+    """Bundles an operation span with what's needed to end it later.
+
+    ``_cm`` is the ``start_as_current_span`` context manager when the span was
+    made current at creation, or None in detached mode (see
+    ``start_operation_span``'s ``set_current``), where the span is made current
+    per-use by ``use_operation_span`` instead.
+    """
+
+    __slots__ = ("_cm", "_name_token", "operation_name", "span")
+
+    def __init__(
+        self,
+        span: Span,
+        cm: Any,
+        name_token: Any,
+        operation_name: str,
+    ) -> None:
+        self.span = span
+        self._cm = cm
+        self._name_token = name_token
+        self.operation_name = operation_name
+
+
+def start_operation_span(
+    tracing_options: Optional[TracingOptions],
+    operation: str,
+    parent_span: Optional[Span],
+    dbname: Optional[str] = None,
+    collection: Optional[str] = None,
+    set_current: bool = True,
+) -> Optional[_OperationSpanHandle]:
+    """Start a CLIENT-kind span for one logical operation, or None.
+
+    Spans all retry attempts of one call to _retry_internal. The (Required,
+    per the OTel spec) ``db.operation.summary`` is always set immediately,
+    to the bare operation name unless ``dbname`` is given, in which case it
+    (and, when ``collection`` is also given, the "if available"
+    ``db.namespace``/``db.collection.name``) are built from those instead.
+    This guarantees a conformant span even for an operation that fails before
+    any command is ever built, e.g. server selection timing out.
+    ``start_command_span`` still backfills these from the real command once
+    one is built, overwriting these values with the authoritative ones.
+
+    ``parent_span`` (the active transaction span, if any) becomes this span's
+    *explicit* parent; it is deliberately not read from ambient context, to
+    avoid a concurrently-running unrelated session's operations picking up
+    this transaction by accident. Pass None outside of a transaction.
+
+    With ``set_current=False`` the span is created but not made current, and
+    the operation-name contextvar is left alone. That suits a span whose
+    lifetime covers several ``_retry_internal`` calls (cursor getMores), where
+    the caller makes it current per call with ``use_operation_span``.
+    """
+    if not _is_tracing_enabled(tracing_options):
+        return None
+    assert _TRACER is not None  # _is_tracing_enabled already checked _HAS_OPENTELEMETRY
+    context = trace.set_span_in_context(parent_span) if parent_span is not None else None
+    attributes: dict[str, Any] = {
+        "db.system.name": "mongodb",
+        "db.operation.name": operation,
+    }
+    name = operation
+    if dbname is not None:
+        name = _build_query_summary(operation, dbname, collection)
+        attributes["db.namespace"] = dbname
+        if collection:
+            attributes["db.collection.name"] = collection
+    attributes["db.operation.summary"] = name
+    if not set_current:
+        span = _TRACER.start_span(
+            name, kind=SpanKind.CLIENT, context=context, attributes=attributes
+        )
+        return _OperationSpanHandle(span, None, None, operation)
+    cm = _TRACER.start_as_current_span(
+        name,
+        kind=SpanKind.CLIENT,
+        context=context,
+        attributes=attributes,
+    )
+    span = cm.__enter__()
+    name_token = _CURRENT_OPERATION_NAME.set(operation)
+    return _OperationSpanHandle(span, cm, name_token, operation)
+
+
+@contextlib.contextmanager
+def use_operation_span(handle: Optional[_OperationSpanHandle]) -> Iterator[None]:
+    """Make a detached operation span current for the duration of the block.
+
+    Does not end the span; the owner (e.g. a cursor, across all of its
+    getMore calls) ends it explicitly. A no-op when ``handle`` is None.
+    """
+    if handle is None:
+        yield
+        return
+    token = _CURRENT_OPERATION_NAME.set(handle.operation_name)
+    try:
+        # record_exception/set_status_on_exception default to True, which would
+        # auto-record any exception propagating out of the block and set ERROR
+        # status here, duplicating what the caller's own
+        # end_operation_span_failure does explicitly once the operation's
+        # final outcome is known. Disabled for the same reason the
+        # attached-mode path passes hardcoded Nones to cm.__exit__.
+        with trace.use_span(
+            handle.span,
+            end_on_exit=False,
+            record_exception=False,
+            set_status_on_exception=False,
+        ):
+            yield
+    finally:
+        _CURRENT_OPERATION_NAME.reset(token)
+
+
+def reset_context() -> None:
+    """Clear the OTel ambient span and operation-name contextvar.
+
+    For long-lived background tasks whose context was copied from whatever
+    happened to be running when they were created (``asyncio.create_task``
+    freezes the caller's ``contextvars.Context``). Without this, spans the task
+    emits are parented under an unrelated, long-since-ended operation and share
+    its trace id. Attaching an empty context makes ``get_current_span()`` return
+    the non-recording invalid span, so spans started afterwards become trace
+    roots. Deliberately does not detach: the task's context is wrong for its
+    whole life, and it dies with the task.
+    """
+    if not _HAS_OPENTELEMETRY:
+        return
+    _CURRENT_OPERATION_NAME.set(None)
+    context.attach(context.Context())
+
+
+def end_operation_span_success(handle: Optional[_OperationSpanHandle]) -> None:
+    """End the operation span with no error status."""
+    if handle is None:
+        return
+    if handle._cm is None:
+        handle.span.end()
+        return
+    _CURRENT_OPERATION_NAME.reset(handle._name_token)
+    handle._cm.__exit__(None, None, None)
+
+
+def end_operation_span_failure(handle: Optional[_OperationSpanHandle], exc: BaseException) -> None:
+    """Record the exception, set the error status, and end the operation span."""
+    if handle is None:
+        return
+    handle.span.record_exception(exc)
+    _set_exception_attributes(handle.span, exc)
+    handle.span.set_status(Status(StatusCode.ERROR, description=str(exc)))
+    if handle._cm is None:
+        handle.span.end()
+        return
+    _CURRENT_OPERATION_NAME.reset(handle._name_token)
+    handle._cm.__exit__(None, None, None)
+
+
+def start_transaction_span(tracing_options: Optional[TracingOptions]) -> Optional[Span]:
+    """Start (but do not make current) the ``"transaction"`` pseudo-span, or None.
+
+    Not pushed as ambient/current context; it's stored explicitly on
+    ``session._transaction.span`` and passed as the explicit ``parent_span``
+    wherever an operation span is started under this transaction (see
+    :func:`start_operation_span`). Per the OTel driver spec, this span has
+    exactly one attribute.
+    """
+    if not _is_tracing_enabled(tracing_options):
+        return None
+    assert _TRACER is not None
+    return _TRACER.start_span(
+        "transaction", kind=SpanKind.CLIENT, attributes={"db.system.name": "mongodb"}
+    )
+
+
+def end_transaction_span(span: Optional[Span]) -> None:
+    """End the transaction span, if any."""
+    if span is None:
+        return
+    span.end()