migrate index usage and index fragmentation metrics to database metrics (#17374)

* migrate index usage and index fragmentation metrics to database metrics

* add changelog

* migrate database backup metrics

* fix lint

* fix lint

Author: lu-zhengda

sqlserver/changelog.d/17374.added

@@ -0,0 +1,2 @@
+Migrate `SQLServer index usage metrics`,  `SQLServer database index fragmentation metrics` and `SQLServer database backup metrics` to `database_metrics`.
+Increase `SQLServer database index fragmentation metrics` and `SQLServer database backup metrics` default collection interval to 5 mins.

sqlserver/datadog_checks/sqlserver/connection.py

@@ -10,6 +10,7 @@
 from datadog_checks.base import AgentCheck, ConfigurationError
 from datadog_checks.base.log import get_check_logger
 from datadog_checks.sqlserver.cursor import CommenterCursorWrapper
+from datadog_checks.sqlserver.utils import construct_use_statement
 
 try:
     import adodbapi
@@ -689,3 +690,29 @@ def test_network_connectivity(self):
                 return "ERROR: {}".format(e.strerror if hasattr(e, 'strerror') else repr(e))
 
         return None
+
+    def _get_current_database_context(self):
+        """
+        Get the current database name.
+        """
+        with self.get_managed_cursor() as cursor:
+            cursor.execute('select DB_NAME()')
+            data = cursor.fetchall()
+            return data[0][0]
+
+    @contextmanager
+    def restore_current_database_context(self):
+        """
+        Restores the default database after executing use statements.
+        """
+        current_db = self._get_current_database_context()
+        try:
+            yield
+        finally:
+            if current_db:
+                try:
+                    self.log.debug("Restoring the original database context %s", current_db)
+                    with self.get_managed_cursor() as cursor:
+                        cursor.execute(construct_use_statement(current_db))
+                except Exception as e:
+                    self.log.error("Failed to switch back to the original database context %s: %s", current_db, e)

sqlserver/datadog_checks/sqlserver/const.py

@@ -21,8 +21,6 @@
 ENGINE_EDITION_AZURE_SQL_EDGE = 9
 ENGINE_EDITION_AZURE_SYNAPSE_SERVERLESS_POOL = 11
 
-DEFAULT_INDEX_USAGE_STATS_INTERVAL = 5 * 60  # 5 minutes
-
 # Keys of the static info cache, used to cache server info which does not change
 STATIC_INFO_VERSION = 'version'
 STATIC_INFO_MAJOR_VERSION = 'major_version'

sqlserver/datadog_checks/sqlserver/database_metrics/__init__.py

@@ -0,0 +1,6 @@
+# (C) Datadog, Inc. 2024-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+from .database_backup_metrics import SqlserverDatabaseBackupMetrics
+from .db_fragmentation_metrics import SqlserverDBFragmentationMetrics
+from .index_usage_metrics import SqlserverIndexUsageMetrics

sqlserver/datadog_checks/sqlserver/database_metrics/base.py

@@ -0,0 +1,115 @@
+# (C) Datadog, Inc. 2024-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+
+
+from typing import Callable, List, Optional
+
+from datadog_checks.base.log import get_check_logger
+from datadog_checks.base.utils.db.core import QueryExecutor
+from datadog_checks.sqlserver.const import STATIC_INFO_ENGINE_EDITION, STATIC_INFO_MAJOR_VERSION
+
+
+class SqlserverDatabaseMetricsBase:
+    def __init__(
+        self,
+        instance_config,
+        new_query_executor,
+        server_static_info,
+        execute_query_handler,
+        track_operation_time=False,
+        databases=None,
+    ):
+        self.instance_config: dict = instance_config
+        self.server_static_info: dict = server_static_info
+        self.new_query_executor: Callable[
+            [List[dict], Callable, Optional[List[str]], Optional[bool]], QueryExecutor
+        ] = new_query_executor
+        self.execute_query_handler: Callable[[str, Optional[str]], List[tuple]] = execute_query_handler
+        self.track_operation_time: bool = track_operation_time
+        self._databases: Optional[List[str]] = databases
+        self._query_executors: List[QueryExecutor] = []
+        self.log = get_check_logger()
+
+    @property
+    def major_version(self) -> Optional[int]:
+        return self.server_static_info.get(STATIC_INFO_MAJOR_VERSION)
+
+    @property
+    def engine_edition(self) -> Optional[int]:
+        return self.server_static_info.get(STATIC_INFO_ENGINE_EDITION)
+
+    @property
+    def enabled(self) -> bool:
+        raise NotImplementedError
+
+    @property
+    def queries(self) -> List[dict]:
+        raise NotImplementedError
+
+    @property
+    def databases(self) -> Optional[List[str]]:
+        return self._databases
+
+    @property
+    def query_executors(self) -> List[QueryExecutor]:
+        '''
+        Returns a list of QueryExecutor objects for the database metrics.
+        '''
+        if not self._query_executors:
+            self._query_executors = self._build_query_executors()
+        return self._query_executors
+
+    def _build_query_executors(self) -> List[QueryExecutor]:
+        '''
+        Builds a list of QueryExecutor objects for the database metrics.
+        '''
+        executor = self.new_query_executor(
+            self.queries, executor=self.execute_query_handler, track_operation_time=self.track_operation_time
+        )
+        executor.compile_queries()
+        return [executor]
+
+    def __repr__(self) -> str:
+        return (
+            f"{self.__class__.__name__}("
+            f"enabled={self.enabled}, "
+            f"major_version={self.major_version}, "
+            f"engine_edition={self.engine_edition})"
+        )
+
+    def metric_names(self) -> List[str]:
+        '''
+        Returns a list of metric names for the queries in the database metrics.
+        Note: This method is used for testing purposes in order to verify that the correct metrics are being collected.
+        '''
+        metric_names = []
+        for query in self.queries:
+            names = [
+                "sqlserver." + c["name"]
+                for c in query["columns"]
+                if not c["type"].startswith("tag") and c["type"] != "source"
+            ]
+            if query.get("extras"):
+                names.extend(
+                    ["sqlserver." + e["name"] for e in query["extras"] if e.get("submit_type") or e.get("type")]
+                )
+            metric_names.append(names)
+        return metric_names
+
+    def tag_names(self) -> List[str]:
+        '''
+        Returns a list of tag names for the queries in the database metrics.
+        Note: This method is used for testing purposes in order to verify that the correct tags are being collected.
+        '''
+        tag_names = []
+        for query in self.queries:
+            tag_names.append([c["name"] for c in query["columns"] if c["type"].startswith("tag")])
+        return tag_names
+
+    def execute(self) -> None:
+        if not self.enabled:
+            self.log.debug("%s: not enabled, skipping execution", str(self))
+            return
+        for query_executor in self.query_executors:
+            query_executor.execute()

sqlserver/datadog_checks/sqlserver/database_metrics/database_backup_metrics.py

@@ -0,0 +1,68 @@
+# (C) Datadog, Inc. 2024-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+
+
+from datadog_checks.sqlserver.utils import is_azure_sql_database
+
+from .base import SqlserverDatabaseMetricsBase
+
+DATABASE_BACKUP_METRICS_QUERY = {
+    "name": "msdb.dbo.backupset",
+    "query": """SELECT
+        sys.databases.name as db,
+        sys.databases.name as database_name,
+        count(backup_set_id) as backup_set_id_count
+        from msdb.dbo.backupset right outer join sys.databases
+        on sys.databases.name = msdb.dbo.backupset.database_name
+        group by sys.databases.name
+    """,
+    "columns": [
+        {"name": "db", "type": "tag"},
+        {"name": "database", "type": "tag"},
+        {"name": "database.backup_count", "type": "gauge"},
+    ],
+}
+
+
+class SqlserverDatabaseBackupMetrics(SqlserverDatabaseMetricsBase):
+    # msdb.dbo.backupset
+    # Contains a row for each backup set. A backup set
+    # contains the backup from a single, successful backup operation.
+    # https://docs.microsoft.com/en-us/sql/relational-databases/system-tables/backupset-transact-sql?view=sql-server-ver15
+    @property
+    def enabled(self):
+        if is_azure_sql_database(self.engine_edition):
+            return False
+        return True
+
+    @property
+    def _default_collection_interval(self) -> int:
+        '''
+        Returns the default interval in seconds at which to collect database backup metrics.
+        '''
+        return 5 * 60  # 5 minutes
+
+    @property
+    def collection_interval(self) -> int:
+        '''
+        Returns the interval in seconds at which to collect database backup metrics.
+        Note: The database backup metrics query can be expensive, so it is recommended to set a higher interval.
+        '''
+        return int(self.instance_config.get('database_backup_metrics_interval', self._default_collection_interval))
+
+    @property
+    def queries(self):
+        # make a copy of the query to avoid modifying the original
+        # in case different instances have different collection intervals
+        query = DATABASE_BACKUP_METRICS_QUERY.copy()
+        query['collection_interval'] = self.collection_interval
+        return [query]
+
+    def __repr__(self) -> str:
+        return (
+            f"{self.__class__.__name__}("
+            f"enabled={self.enabled}, "
+            f"engine_edition={self.engine_edition}, "
+            f"collection_interval={self.collection_interval})"
+        )

sqlserver/datadog_checks/sqlserver/database_metrics/db_fragmentation_metrics.py

@@ -0,0 +1,137 @@
+# (C) Datadog, Inc. 2024-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+
+import copy
+import functools
+
+from datadog_checks.base.config import is_affirmative
+from datadog_checks.base.errors import ConfigurationError
+
+from .base import SqlserverDatabaseMetricsBase
+
+DB_FRAGMENTATION_QUERY = {
+    "name": "sys.dm_db_index_physical_stats",
+    "query": """SELECT
+        DB_NAME(DDIPS.database_id) as database_name,
+        OBJECT_NAME(DDIPS.object_id, DDIPS.database_id) as object_name,
+        DDIPS.index_id as index_id,
+        I.name as index_name,
+        DDIPS.fragment_count as fragment_count,
+        DDIPS.avg_fragment_size_in_pages as avg_fragment_size_in_pages,
+        DDIPS.page_count as page_count,
+        DDIPS.avg_fragmentation_in_percent as avg_fragmentation_in_percent
+        FROM sys.dm_db_index_physical_stats (DB_ID('{db}'),null,null,null,null) as DDIPS
+        INNER JOIN sys.indexes as I WITH (NOLOCK) ON I.object_id = DDIPS.object_id
+        AND DDIPS.index_id = I.index_id
+        WHERE DDIPS.fragment_count is not null
+    """,
+    "columns": [
+        {"name": "database_name", "type": "tag"},
+        {"name": "object_name", "type": "tag"},
+        {"name": "index_id", "type": "tag"},
+        {"name": "index_name", "type": "tag"},
+        {"name": "database.fragment_count", "type": "gauge"},
+        {"name": "database.avg_fragment_size_in_pages", "type": "gauge"},
+        {"name": "database.index_page_count", "type": "gauge"},
+        {"name": "database.avg_fragmentation_in_percent", "type": "gauge"},
+    ],
+}
+
+
+class SqlserverDBFragmentationMetrics(SqlserverDatabaseMetricsBase):
+    # sys.dm_db_index_physical_stats
+    #
+    # Returns size and fragmentation information for the data and
+    # indexes of the specified table or view in SQL Server.
+    #
+    # There are reports of this query being very slow for large datasets,
+    # so debug query timing are included to help monitor it.
+    # https://dba.stackexchange.com/q/76374
+    #
+    # https://docs.microsoft.com/en-us/sql/relational-databases/system-dynamic-management-views/sys-dm-db-index-physical-stats-transact-sql?view=sql-server-ver15
+    @property
+    def include_db_fragmentation_metrics(self):
+        return is_affirmative(self.instance_config.get('include_db_fragmentation_metrics', False))
+
+    @property
+    def include_db_fragmentation_metrics_tempdb(self):
+        return is_affirmative(self.instance_config.get('include_db_fragmentation_metrics_tempdb', False))
+
+    @property
+    def db_fragmentation_object_names(self):
+        return self.instance_config.get('db_fragmentation_object_names', [])
+
+    @property
+    def enabled(self):
+        if not self.include_db_fragmentation_metrics:
+            return False
+        return True
+
+    @property
+    def _default_collection_interval(self) -> int:
+        '''
+        Returns the default interval in seconds at which to collect database index fragmentation metrics.
+        '''
+        return 5 * 60  # 5 minutes
+
+    @property
+    def collection_interval(self) -> int:
+        '''
+        Returns the interval in seconds at which to collect database index fragmentation metrics.
+        Note: The index fragmentation metrics query can be expensive, so it is recommended to set a higher interval.
+        '''
+        return int(self.instance_config.get('db_fragmentation_metrics_interval', self._default_collection_interval))
+
+    @property
+    def databases(self):
+        '''
+        Returns a list of databases to collect index fragmentation metrics for.
+        By default, tempdb is excluded.
+        '''
+        if not self._databases:
+            raise ConfigurationError("No databases configured for index usage metrics")
+        if not self.include_db_fragmentation_metrics_tempdb:
+            try:
+                self._databases.remove('tempdb')
+            except ValueError:
+                pass
+        return self._databases
+
+    @property
+    def queries(self):
+        # make a copy of the query to avoid modifying the original
+        # in case different instances have different collection intervals
+        query = DB_FRAGMENTATION_QUERY.copy()
+        query['collection_interval'] = self.collection_interval
+        return [query]
+
+    def __repr__(self) -> str:
+        return (
+            f"{self.__class__.__name__}("
+            f"enabled={self.enabled}, "
+            f"include_db_fragmentation_metrics={self.include_db_fragmentation_metrics}, "
+            f"include_db_fragmentation_metrics_tempdb={self.include_db_fragmentation_metrics_tempdb}, "
+            f"db_fragmentation_object_names={self.db_fragmentation_object_names}, "
+            f"collection_interval={self.collection_interval})"
+        )
+
+    def _build_query_executors(self):
+        executors = []
+        for database in self.databases:
+            queries = copy.deepcopy(self.queries)
+            for query in queries:
+                query['query'] = query['query'].format(db=database)
+                if self.db_fragmentation_object_names:
+                    query['query'] += " AND OBJECT_NAME(DDIPS.object_id, DDIPS.database_id) IN ({})".format(
+                        ','.join(["'{}'".format(name) for name in self.db_fragmentation_object_names])
+                    )
+            executor = self.new_query_executor(
+                queries,
+                executor=functools.partial(self.execute_query_handler, db=database),
+                extra_tags=['db:{}'.format(database)],
+                track_operation_time=self.track_operation_time,
+            )
+            executor.compile_queries()
+            executors.append(executor)
+        return executors