[mongo] skip explain administrative aggregation pipeline

mongo/changelog.d/18844.fixed

@@ -0,0 +1 @@
+Skip explain plan collection for mongo administrative aggregation pipeline, including `$collStats`, `$currentOp`, `$indexStats`, `$listSearchIndexes`, `$sample` and `$shardedDataDistribution`.

mongo/datadog_checks/mongo/dbm/utils.py

@@ -57,6 +57,17 @@
     ]
 )
 
+UNEXPLAINABLE_PIPELINE_STAGES = frozenset(
+    [
+        "$collStats",
+        "$currentOp",
+        "$indexStats",
+        "$listSearchIndexes",
+        "$sample",
+        "$shardedDataDistribution",
+    ]
+)
+
 COMMAND_KEYS_TO_REMOVE = frozenset(["comment", "lsid", "$clusterTime"])
 
 EXPLAIN_PLAN_KEYS_TO_REMOVE = frozenset(
@@ -108,6 +119,12 @@ def should_explain_operation(
     if any(command.get(key) for key in UNEXPLAINABLE_COMMANDS):
         return False
 
+    # if UNEXPLAINABLE_PIPELINE_STAGES in command pipeline stages, skip
+    if pipeline := command.get("pipeline"):
+        stages = [list(stage.keys())[0] for stage in pipeline if isinstance(stage, dict)]
+        if any(stage in UNEXPLAINABLE_PIPELINE_STAGES for stage in stages):
+            return False
+
     db, _ = namespace.split(".", 1)
     if db in MONGODB_SYSTEM_DATABASES:
         return False

mongo/tests/test_unit.py

@@ -10,12 +10,15 @@
 
 import mock
 import pytest
+from bson import json_util
 from pymongo.errors import ConnectionFailure, OperationFailure
 
 from datadog_checks.base import ConfigurationError
+from datadog_checks.base.utils.db.sql import compute_exec_plan_signature
 from datadog_checks.mongo.api import CRITICAL_FAILURE, MongoApi
 from datadog_checks.mongo.collectors import MongoCollector
 from datadog_checks.mongo.common import MongosDeployment, ReplicaSetDeployment, get_state_name
+from datadog_checks.mongo.dbm.utils import should_explain_operation
 from datadog_checks.mongo.mongo import HostingType, MongoDb, metrics
 from datadog_checks.mongo.utils import parse_mongo_uri
 
@@ -778,3 +781,114 @@ def seed_mock_client():
 def load_json_fixture(name):
     with open(os.path.join(common.HERE, "fixtures", name), 'r') as f:
         return json.load(f)
+
+
+@pytest.mark.parametrize(
+    'namespace,op,command,should_explain',
+    [
+        pytest.param(
+            "test.test",
+            "command",
+            {
+                "aggregate": "test",
+                "pipeline": [{"$collStats": {"latencyStats": {}, "storageStats": {}, "queryExecStats": {}}}],
+                "cursor": {},
+                "$db": "test",
+                "$readPreference": {"mode": "?"},
+            },
+            False,
+            id='no-explain $collStats',
+        ),
+        pytest.param(
+            "test.test",
+            "command",
+            {
+                "aggregate": "test",
+                "pipeline": [{"$sample": {"size": "?"}}],
+                "cursor": {},
+                "$db": "test",
+                "$readPreference": {"mode": "?"},
+            },
+            False,
+            id='no explain $sample',
+        ),
+        pytest.param(
+            "test.test",
+            "command",
+            {
+                "aggregate": "test",
+                "pipeline": [{"$indexStats": {}}],
+                "cursor": {},
+                "$db": "test",
+                "$readPreference": {"mode": "?"},
+            },
+            False,
+            id='no explain $indexStats',
+        ),
+        pytest.param(
+            "test.test",
+            "command",
+            {"getMore": "?", "collection": "test", "$db": "test", "$readPreference": {"mode": "?"}},
+            False,
+            id='no explain getMore',
+        ),
+        pytest.param(
+            "test.test",
+            "update",
+            {
+                "update": "test",
+                "updates": [{"q": {}, "u": {}, "multi": False, "upsert": False}],
+                "ordered": True,
+                "$db": "test",
+                "$readPreference": {"mode": "?"},
+            },
+            False,
+            id='no explain update',
+        ),
+        pytest.param(
+            "test.test",
+            "insert",
+            {
+                "insert": "test",
+                "documents": [{"_id": "?", "a": 1}],
+                "ordered": True,
+                "$db": "test",
+                "$readPreference": {"mode": "?"},
+            },
+            False,
+            id='no explain insert',
+        ),
+        pytest.param(
+            "test.test",
+            "remove",
+            {
+                "delete": "test",
+                "deletes": [{"q": {}, "limit": 1}],
+                "ordered": True,
+                "$db": "test",
+                "$readPreference": {"mode": "?"},
+            },
+            False,
+            id='no explain delete',
+        ),
+        pytest.param(
+            "test.test",
+            "query",
+            {"find": "test", "filter": {}, "$db": "test", "$readPreference": {"mode": "?"}},
+            True,
+            id='explain find',
+        ),
+    ],
+)
+def test_should_explain_operation(namespace, op, command, should_explain):
+    check = MongoDb('mongo', {}, [{'hosts': ['localhost']}])
+    assert (
+        should_explain_operation(
+            namespace,
+            op,
+            command,
+            explain_plan_rate_limiter=check._operation_samples._explained_operations_ratelimiter,
+            explain_plan_cache_key=(namespace, op, compute_exec_plan_signature(json_util.dumps(command))),
+        )
+        == should_explain
+    )