feat(pymongo): aggregate and getMore capture statements support (#3601)

* feat(pymongo): aggregate and getMore capture statements

* chore: update changelog

* fix: tests

* fix: proper MockCommand and expectations

* chore: ruff-format

Author: smoke

CHANGELOG.md

@@ -22,7 +22,9 @@ and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0
 
 - `opentelemetry-util-http` Added support for redacting specific url query string values and url credentials in instrumentations
   ([#3508](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3508))
-  
+- `opentelemetry-instrumentation-pymongo` `aggregate` and `getMore` capture statements support
+  ([#3601](https://github.com/open-telemetry/opentelemetry-python-contrib/pull/3601))
+
 ## Version 1.34.0/0.55b0 (2025-06-04)
 
 ### Fixed

instrumentation/opentelemetry-instrumentation-pymongo/src/opentelemetry/instrumentation/pymongo/utils.py

@@ -17,4 +17,6 @@
     "delete": "deletes",
     "update": "updates",
     "find": "filter",
+    "getMore": "collection",
+    "aggregate": "pipeline",
 }

instrumentation/opentelemetry-instrumentation-pymongo/tests/test_pymongo.py

@@ -57,14 +57,12 @@ def test_started(self):
         # pylint: disable=protected-access
         span = command_tracer._pop_span(mock_event)
         self.assertIs(span.kind, trace_api.SpanKind.CLIENT)
-        self.assertEqual(span.name, "database_name.command_name")
+        self.assertEqual(span.name, "database_name.find")
         self.assertEqual(span.attributes[SpanAttributes.DB_SYSTEM], "mongodb")
         self.assertEqual(
             span.attributes[SpanAttributes.DB_NAME], "database_name"
         )
-        self.assertEqual(
-            span.attributes[SpanAttributes.DB_STATEMENT], "command_name"
-        )
+        self.assertEqual(span.attributes[SpanAttributes.DB_STATEMENT], "find")
         self.assertEqual(
             span.attributes[SpanAttributes.NET_PEER_NAME], "test.com"
         )
@@ -181,7 +179,7 @@ def test_int_command(self):
 
         self.assertEqual(len(spans_list), 1)
         span = spans_list[0]
-        self.assertEqual(span.name, "database_name.command_name")
+        self.assertEqual(span.name, "database_name.123")
 
     def test_no_op_tracer(self):
         mock_event = MockEvent({})
@@ -194,6 +192,90 @@ def test_no_op_tracer(self):
         spans_list = self.memory_exporter.get_finished_spans()
         self.assertEqual(len(spans_list), 0)
 
+    def test_capture_statement_getmore(self):
+        command_attrs = {
+            "command_name": "getMore",
+            "collection": "test_collection",
+        }
+        mock_event = MockEvent(command_attrs)
+
+        command_tracer = CommandTracer(self.tracer, capture_statement=True)
+        command_tracer.started(event=mock_event)
+        command_tracer.succeeded(event=mock_event)
+
+        spans_list = self.memory_exporter.get_finished_spans()
+
+        self.assertEqual(len(spans_list), 1)
+        span = spans_list[0]
+
+        self.assertEqual(
+            span.attributes[SpanAttributes.DB_STATEMENT],
+            "getMore test_collection",
+        )
+
+    def test_capture_statement_aggregate(self):
+        pipeline = [
+            {"$match": {"status": "active"}},
+            {"$group": {"_id": "$category", "count": {"$sum": 1}}},
+        ]
+        command_attrs = {
+            "command_name": "aggregate",
+            "pipeline": pipeline,
+        }
+        command_tracer = CommandTracer(self.tracer, capture_statement=True)
+        mock_event = MockEvent(command_attrs)
+        command_tracer.started(event=mock_event)
+        command_tracer.succeeded(event=mock_event)
+
+        spans_list = self.memory_exporter.get_finished_spans()
+
+        self.assertEqual(len(spans_list), 1)
+        span = spans_list[0]
+
+        expected_statement = f"aggregate {pipeline}"
+        self.assertEqual(
+            span.attributes[SpanAttributes.DB_STATEMENT], expected_statement
+        )
+
+    def test_capture_statement_disabled_getmore(self):
+        command_attrs = {
+            "command_name": "getMore",
+            "collection": "test_collection",
+        }
+        command_tracer = CommandTracer(self.tracer, capture_statement=False)
+        mock_event = MockEvent(command_attrs)
+        command_tracer.started(event=mock_event)
+        command_tracer.succeeded(event=mock_event)
+
+        spans_list = self.memory_exporter.get_finished_spans()
+
+        self.assertEqual(len(spans_list), 1)
+        span = spans_list[0]
+
+        self.assertEqual(
+            span.attributes[SpanAttributes.DB_STATEMENT], "getMore"
+        )
+
+    def test_capture_statement_disabled_aggregate(self):
+        pipeline = [{"$match": {"status": "active"}}]
+        command_attrs = {
+            "command_name": "aggregate",
+            "pipeline": pipeline,
+        }
+        command_tracer = CommandTracer(self.tracer, capture_statement=False)
+        mock_event = MockEvent(command_attrs)
+        command_tracer.started(event=mock_event)
+        command_tracer.succeeded(event=mock_event)
+
+        spans_list = self.memory_exporter.get_finished_spans()
+
+        self.assertEqual(len(spans_list), 1)
+        span = spans_list[0]
+
+        self.assertEqual(
+            span.attributes[SpanAttributes.DB_STATEMENT], "aggregate"
+        )
+
 
 class MockCommand:
     def __init__(self, command_attrs):
@@ -206,6 +288,7 @@ def get(self, key, default=""):
 class MockEvent:
     def __init__(self, command_attrs, connection_id=None, request_id=""):
         self.command = MockCommand(command_attrs)
+        self.command_name = self.command.get("command_name")
         self.connection_id = connection_id
         self.request_id = request_id