PYTHON-6045 Add error.type OpenTelemetry command span attribute

.evergreen/scripts/setup_tests.py

@@ -458,7 +458,9 @@ def handle_test_env() -> None:
     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")
+        # The floor matches requirements/opentelemetry.txt: with --resolution=lowest-direct an
+        # unpinned sdk resolves to 1.0.0 and drags opentelemetry-api down to it. PYTHON-5947.
+        UV_ARGS.append('--with "opentelemetry-sdk>=1.20.0"')
 
     if test_name == "perf":
         data_dir = ROOT / "specifications/source/benchmarking/data"

pymongo/_otel.py

@@ -411,12 +411,14 @@ def end_command_span_success(span: Optional[Span], reply: _DocumentOut) -> None:
         span.end()
 
 
-def _set_exception_attributes(span: Span, exc: BaseException) -> None:
+def _set_exception_attributes(span: Span, exc: BaseException) -> str:
     """Set exception.type/exception.message/exception.stacktrace span attributes.
 
     ``record_exception`` attaches these to an "exception" *event* only, but the
     spec requires them as span *attributes* too, for both command and operation
     spans. Formatting mirrors ``record_exception``.
+
+    :return: The ``exception.type`` value.
     """
     module = type(exc).__module__
     qualname = type(exc).__qualname__
@@ -427,6 +429,7 @@ def _set_exception_attributes(span: Span, exc: BaseException) -> None:
         "exception.stacktrace",
         "".join(traceback.format_exception(type(exc), exc, exc.__traceback__)),
     )
+    return exception_type
 
 
 def end_command_span_failure(
@@ -439,10 +442,14 @@ def end_command_span_failure(
         return
     try:
         span.record_exception(exc)
-        _set_exception_attributes(span, exc)
+        exception_type = _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_attribute("error.type", str(code))
+        else:
+            # A network failure gets no server reply, so there is no code to report.
+            span.set_attribute("error.type", exception_type)
         span.set_status(Status(StatusCode.ERROR, description=failure.get("errmsg")))
     finally:
         # End even if recording raised, so a failure here costs the attributes
@@ -590,7 +597,8 @@ def end_operation_span_failure(handle: Optional[_OperationSpanHandle], exc: Base
         return
     try:
         handle.span.record_exception(exc)
-        _set_exception_attributes(handle.span, exc)
+        exception_type = _set_exception_attributes(handle.span, exc)
+        handle.span.set_attribute("error.type", exception_type)
         handle.span.set_status(Status(StatusCode.ERROR, description=str(exc)))
     finally:
         # Unwind even if recording raised, since a span left current would

test/asynchronous/test_otel.py

@@ -34,7 +34,9 @@
 from pymongo.errors import (
     ClientBulkWriteException,
     ConfigurationError,
+    ConnectionFailure,
     InvalidOperation,
+    NetworkTimeout,
     OperationFailure,
     ServerSelectionTimeoutError,
 )
@@ -76,6 +78,11 @@ def test_result_never_exceeds_max_length(self):
             self.assertLessEqual(len(text), max_length, (max_length, text))
 
 
+def _qualified_name(exc_type: type) -> str:
+    """Format an exception class the way the spans do: ``module.QualName``."""
+    return f"{exc_type.__module__}.{exc_type.__qualname__}"
+
+
 @unittest.skipUnless(_HAS_OTEL_TEST_DEPS, "opentelemetry-sdk is not installed")
 class TestOTelOperationSpanPrimitives(unittest.TestCase):
     """Unit tests for the pymongo._otel operation-span primitives."""
@@ -548,8 +555,81 @@ async def test_failure_records_exception_and_status_code(self):
         span = spans[0]
         self.assertEqual(span.status.status_code, trace.StatusCode.ERROR)
         self.assertIn("db.response.status_code", span.attributes)
+        # For a server error the spec has error.type mirror the status code.
+        self.assertEqual(span.attributes["error.type"], span.attributes["db.response.status_code"])
         self.assertTrue(any(event.name == "exception" for event in span.events))
 
+    @async_client_context.require_failCommand_fail_point
+    async def test_operation_span_error_type_is_exception_class_name_for_server_error(self):
+        # A non-retryable server error names the exception class on the operation
+        # span, not the server error code.
+        client = await self.async_rs_or_single_client(tracing={"enabled": True}, retryReads=False)
+        fail_command = {
+            "configureFailPoint": "failCommand",
+            "mode": {"times": 1},
+            "data": {"failCommands": ["find"], "errorCode": 2},
+        }
+        async with self.fail_point(fail_command):
+            self.exporter.clear()
+            with self.assertRaises(OperationFailure):
+                await client[self.db.name].test.find_one({})
+
+        finished = self.exporter.get_finished_spans()
+        (cmd_span,) = self.command_spans(finished, "find")
+        (op_span,) = self.operation_spans(finished, "find")
+        self.assertEqual(op_span.attributes["error.type"], op_span.attributes["exception.type"])
+        self.assertNotEqual(
+            op_span.attributes["error.type"], cmd_span.attributes["db.response.status_code"]
+        )
+
+    @async_client_context.require_failCommand_fail_point
+    async def test_error_type_is_exception_class_name_for_connection_failure(self):
+        # A closed connection produces no server reply, so error.type uses the class name.
+        client = await self.async_rs_or_single_client(tracing={"enabled": True}, retryReads=False)
+        fail_command = {
+            "configureFailPoint": "failCommand",
+            "mode": {"times": 1},
+            "data": {"failCommands": ["find"], "closeConnection": True},
+        }
+        async with self.fail_point(fail_command):
+            self.exporter.clear()
+            with self.assertRaises(ConnectionFailure):
+                await client[self.db.name].test.find_one({})
+
+        finished = self.exporter.get_finished_spans()
+        (cmd_span,) = self.command_spans(finished, "find")
+        self.assertNotIn("db.response.status_code", cmd_span.attributes)
+        self.assertEqual(cmd_span.attributes["error.type"], cmd_span.attributes["exception.type"])
+        (op_span,) = self.operation_spans(finished, "find")
+        self.assertEqual(op_span.attributes["error.type"], op_span.attributes["exception.type"])
+
+    @async_client_context.require_failCommand_blockConnection
+    async def test_error_type_is_exception_class_name_for_network_timeout(self):
+        # socketTimeoutMS trips before any reply, so there is no server error code.
+        client = await self.async_rs_or_single_client(
+            tracing={"enabled": True}, socketTimeoutMS=200, retryReads=False
+        )
+        fail_command = {
+            "configureFailPoint": "failCommand",
+            "mode": {"times": 1},
+            "data": {
+                "failCommands": ["find"],
+                "blockConnection": True,
+                "blockTimeMS": 1000,
+            },
+        }
+        async with self.fail_point(fail_command):
+            self.exporter.clear()
+            with self.assertRaises(NetworkTimeout) as ctx:
+                await client[self.db.name].test.find_one({})
+
+        spans = [s for s in self.spans() if s.attributes.get("db.command.name") == "find"]
+        self.assertEqual(len(spans), 1)
+        attrs = spans[0].attributes
+        self.assertNotIn("db.response.status_code", attrs)
+        self.assertEqual(attrs["error.type"], _qualified_name(NetworkTimeout))
+        self.assertIsInstance(ctx.exception, NetworkTimeout)
+
     async def test_tracing_disabled_by_default(self):
         client = await self.async_rs_or_single_client()
         self.exporter.clear()