[SQLServer] Filter query metrics by database_name if check is configured to a single non-master db (#17717)

* filter query metrics by database_name if monitoring is configured to single non-master db

* add changelog

* only filter unmonitored database on azure sql database

Author: lu-zhengda

sqlserver/changelog.d/17717.fixed

@@ -0,0 +1 @@
+Filter query metrics by database_name if check is configured to a single non-master db

sqlserver/datadog_checks/sqlserver/statements.py

@@ -22,7 +22,7 @@
 from datadog_checks.base.utils.serialization import json
 from datadog_checks.base.utils.tracking import tracked_method
 from datadog_checks.sqlserver.config import SQLServerConfig
-from datadog_checks.sqlserver.utils import extract_sql_comments_and_procedure_name
+from datadog_checks.sqlserver.utils import extract_sql_comments_and_procedure_name, is_azure_sql_database
 
 try:
     import datadog_agent
@@ -304,10 +304,29 @@ def _load_raw_query_metrics_rows(self, cursor):
         self.log.debug("loaded sql server statement metrics len(rows)=%s", len(rows))
         return rows
 
+    def _should_include_query_metrics_row(self, row):
+        if 'database_name' not in row:
+            # Unaggregated query metrics row without a database name
+            return True
+
+        engine_edition = self._check.static_info_cache.get(STATIC_INFO_ENGINE_EDITION, "")
+        if is_azure_sql_database(engine_edition):
+            configured_database = self._check.instance.get('database')
+            if not configured_database or configured_database == 'master':
+                # If no database configured or the configured database is master, include all rows
+                return True
+            if not row['database_name'] or str(row['database_name']).lower() != configured_database.lower():
+                # If the row's database name does not match the configured database, exclude the row
+                return False
+
+        return True
+
     def _normalize_queries(self, rows):
         normalized_rows = []
 
         for row in rows:
+            if not self._should_include_query_metrics_row(row):
+                continue
             # Attempt to obfuscate SQL statement with metadata
             procedure_statement = None
             try:

sqlserver/tests/test_statements.py

@@ -26,7 +26,9 @@
     ENGINE_EDITION_ENTERPRISE,
     ENGINE_EDITION_EXPRESS,
     ENGINE_EDITION_PERSONAL,
+    ENGINE_EDITION_SQL_DATABASE,
     ENGINE_EDITION_STANDARD,
+    STATIC_INFO_ENGINE_EDITION,
 )
 from datadog_checks.sqlserver.statements import SQL_SERVER_QUERY_METRICS_COLUMNS, obfuscate_xml_plan
 
@@ -1014,3 +1016,64 @@ def test_metrics_lookback_multiplier(instance_docker):
 
     check.statement_metrics._load_raw_query_metrics_rows(mock_cursor)
     mock_cursor.execute.assert_called_with(ANY, (6,))
+
+
+@pytest.mark.integration
+@pytest.mark.usefixtures('dd_environment')
+@pytest.mark.parametrize(
+    "configured_database,engine_edition,filter_to_configured_database",
+    [
+        pytest.param(None, None, False, id="no_database_configured_not_azure_sql"),  # should default to master
+        pytest.param(
+            "datadog_test", None, False, id="configured_database_datadog_test_not_azure_sql"
+        ),  # should not filter to configured database datadog_test
+        pytest.param(
+            "master", None, False, id="configured_database_master_not_azure_sql"
+        ),  # should use configured database master
+        pytest.param(
+            "datadog_test", ENGINE_EDITION_SQL_DATABASE, True, id="configured_database_datadog_test_azure_sql"
+        ),  # should filter to configured database datadog_test
+    ],
+)
+def test_statement_with_metrics_azure_sql_filtered_to_configured_database(
+    aggregator,
+    dd_run_check,
+    dbm_instance,
+    bob_conn,
+    configured_database,
+    engine_edition,
+    filter_to_configured_database,
+):
+    if configured_database:
+        dbm_instance['database'] = configured_database
+    check = SQLServer(CHECK_NAME, {}, [dbm_instance])
+
+    if engine_edition:
+        check.static_info_cache[STATIC_INFO_ENGINE_EDITION] = engine_edition
+
+    def _execute_queries():
+        bob_conn.execute_with_retries("SELECT * FROM ϑings", (), database="datadog_test")
+        bob_conn.execute_with_retries("SELECT count(*) from sys.databases", (), database="master")
+
+    dd_run_check(check)
+    _execute_queries()
+    dd_run_check(check)
+    aggregator.reset()
+    _execute_queries()
+    dd_run_check(check)
+
+    # dbm-metrics
+    dbm_metrics = aggregator.get_event_platform_events("dbm-metrics")
+    assert len(dbm_metrics) == 1, "should have collected exactly one dbm-metrics payload"
+    payload = dbm_metrics[0]
+    # metrics rows
+    sqlserver_rows = payload.get('sqlserver_rows', [])
+    assert sqlserver_rows, "should have collected some sqlserver query metrics rows"
+    if filter_to_configured_database:
+        assert all(
+            row['database_name'] == configured_database for row in sqlserver_rows
+        ), "should have only collected metrics for configured database"
+    else:
+        database_names = {row['database_name'] for row in sqlserver_rows}
+        assert 'datadog_test' in database_names, "should have collected metrics for datadog_test databases"
+        assert 'master' in database_names, "should have collected metrics for master databases"