[sqlserver] Share one performance counter query and index rows by instance

sqlserver/changelog.d/24908.fixed

@@ -0,0 +1 @@
+Further speed up performance counter metric collection on instances with many autodiscovered databases, and skip fraction metrics whose base counter is missing for an instance instead of ending the collection early.

sqlserver/datadog_checks/sqlserver/metrics.py

@@ -9,10 +9,11 @@
 Collection of metric classes for specific SQL Server tables.
 """
 
-from __future__ import division
+from __future__ import annotations, division
 
 from collections import defaultdict
 from functools import partial
+from typing import Any
 
 # Queries
 ALL_INSTANCES = 'ALL'
@@ -86,120 +87,128 @@ def fetch_metric(self, rows, columns, values_cache=None):
         raise NotImplementedError
 
 
+# A snapshot of sys.dm_os_performance_counters, indexed for lookup by the two columns a metric knows about
+# itself: counter name, then instance name, then the (object name, counter value) of each row underneath.
+PerfCounterIndex = dict[str, dict[str, list[tuple[str, int]]]]
+
+
 # https://docs.microsoft.com/en-us/sql/relational-databases/system-dynamic-management-views/sys-dm-os-performance-counters-transact-sql
-class SqlSimpleMetric(BaseSqlServerMetric):
+class SqlPerfCounterMetric(BaseSqlServerMetric):
+    """Base class for the metrics read from sys.dm_os_performance_counters.
+
+    Subclasses differ only in how they turn counter values into metrics, so they share a single query and a
+    single snapshot of the table: scanning it costs the same whether two rows come back or thousands, so a
+    query per subclass is pure overhead.
+    """
+
     TABLE = 'sys.dm_os_performance_counters'
-    DEFAULT_METRIC_TYPE = None  # can be either rate or gauge
     QUERY_BASE = """select counter_name, instance_name, object_name, cntr_value
                     from {table} where counter_name in ({{placeholders}})""".format(table=TABLE)
-    OPERATION_NAME = 'simple_metrics'
+    OPERATION_NAME = 'perf_counter_metrics'
 
     @classmethod
-    def fetch_all_values(cls, cursor, counters_list, logger, databases=None, engine_edition=None):
+    def fetch_all_values(
+        cls,
+        cursor: Any,
+        counters_list: list[str],
+        logger: Any,
+        databases: list[str] | None = None,
+        engine_edition: str | None = None,
+    ) -> tuple[PerfCounterIndex, None]:
         rows, _ = cls._fetch_generic_values(cursor, counters_list, logger)
-        # The name columns are nchar(128), so every value arrives blank-padded. Strip once here and group by
-        # counter name so each metric only walks its own rows: with autodiscovery both the number of metrics
-        # and the number of rows grow with the database count, making a per-metric scan quadratic.
-        results = defaultdict(list)
+        # The name columns are nchar(128), so every value arrives blank-padded. Strip once here and index by
+        # counter name and instance name so each metric can look its own rows up directly: with autodiscovery
+        # both the number of metrics and the number of rows grow with the database count, so scanning rows per
+        # metric is quadratic.
+        results = defaultdict(lambda: defaultdict(list))
         for counter_name, instance_name, object_name, cntr_value in rows:
-            results[counter_name.strip()].append((instance_name.strip(), object_name.strip(), cntr_value))
+            results[counter_name.strip()][instance_name.strip()].append((object_name.strip(), cntr_value))
         return results, None
 
+    def _configured_instance_names(self) -> tuple[str, ...]:
+        """The instance names this metric collects from, in precedence order.
+
+        The physical database name is a fallback for the logical one and differs from it only on Azure SQL
+        Database; skip it when the two match so the metric is not reported twice.
+        """
+        if self.physical_db_name and self.physical_db_name != self.instance:
+            return (self.instance, self.physical_db_name)
+        return (self.instance,)
+
+
+class SqlSimpleMetric(SqlPerfCounterMetric):
+    DEFAULT_METRIC_TYPE = None  # can be either rate or gauge
+
     def fetch_metric(self, results, columns, values_cache=None):
-        for instance_name, object_name, cntr_value in results.get(self.sql_name, ()):
-            if (self.instance == ALL_INSTANCES and instance_name != "_Total") or (
-                (instance_name == self.instance or instance_name == self.physical_db_name)
-                and (not self.object_name or object_name == self.object_name)
-            ):
-                metric_tags = list(self.tags)
-                if self.instance == ALL_INSTANCES:
-                    metric_tags.append('{}:{}'.format(self.tag_by, instance_name))
-                self.report_function(self.metric_name, cntr_value, tags=metric_tags)
-                if self.instance != ALL_INSTANCES:
-                    break
+        counters_by_instance = results.get(self.sql_name)
+        if not counters_by_instance:
+            return
 
+        if self.instance == ALL_INSTANCES:
+            for instance_name, counters in counters_by_instance.items():
+                if instance_name == "_Total":
+                    continue
+                metric_tags = [*self.tags, '{}:{}'.format(self.tag_by, instance_name)]
+                for _object_name, cntr_value in counters:
+                    self.report_function(self.metric_name, cntr_value, tags=metric_tags)
+            return
 
-class SqlFractionMetric(BaseSqlServerMetric):
-    TABLE = 'sys.dm_os_performance_counters'
-    DEFAULT_METRIC_TYPE = 'gauge'
-    QUERY_BASE = """select counter_name, cntr_type, cntr_value, instance_name, object_name
-                    from {table}
-                    where counter_name in ({{placeholders}})
-                    order by cntr_type;""".format(table=TABLE)
-    OPERATION_NAME = 'fraction_metrics'
+        for instance_name in self._configured_instance_names():
+            for object_name, cntr_value in counters_by_instance.get(instance_name, ()):
+                if not self.object_name or object_name == self.object_name:
+                    self.report_function(self.metric_name, cntr_value, tags=list(self.tags))
+                    return
 
-    @classmethod
-    def fetch_all_values(cls, cursor, counters_list, logger, databases=None, engine_edition=None):
-        placeholders = ', '.join('?' for _ in counters_list)
-        query = cls.QUERY_BASE.format(placeholders=placeholders)
 
-        logger.debug("%s: fetch_all executing query: %s, %s", cls.__name__, query, str(counters_list))
-        cursor.execute(query, counters_list)
-        rows = cursor.fetchall()
-        results = defaultdict(list)
-
-        for counter_name, cntr_type, cntr_value, instance_name, object_name in rows:
-            counter_result = {
-                'cntr_type': cntr_type,
-                'cntr_value': cntr_value,
-                'instance_name': instance_name.strip(),
-                'object_name': object_name.strip(),
-            }
-            logger.debug("Adding new counter_result %s", str(counter_result))
-            results[counter_name.strip()].append(counter_result)
-        return results, None
+class SqlFractionMetric(SqlPerfCounterMetric):
+    DEFAULT_METRIC_TYPE = 'gauge'
 
     def fetch_metric(self, results, columns, values_cache=None):
         if not self.base_name:
             self.log.error('Skipping counter. Missing base counter name')
             return
-        num_counters = results.get(self.sql_name.strip())
-        base_counters = results.get(self.base_name.strip())
-        if not num_counters or not base_counters:
+        numerators = results.get(self.sql_name.strip())
+        bases = results.get(self.base_name.strip())
+        if not numerators or not bases:
             self.log.error(
-                'Skipping counter. Missing numerator and/or base counters \nsql_name=%s \nbase_name=%s \nresults=%s',
+                'Skipping counter. Missing numerator and/or base counters \nsql_name=%s \nbase_name=%s',
                 self.sql_name,
                 self.base_name,
-                str(results),
             )
             return
 
-        base_by_key = {}
-
-        # let's organize each base counter by key
-        for base in base_counters:
-            key = '{}::{}'.format(base['instance_name'], base['object_name'])
-            if base_by_key.get(key):
-                self.log.warning('Found duplicate base counters for key:%s', key)
-            base_by_key[key] = base
-
-        for numerator in num_counters:
-            instance_name = numerator['instance_name']
-            object_name = numerator['object_name']
-            if (
-                self.instance != ALL_INSTANCES
-                and instance_name != self.instance
-                and instance_name != self.physical_db_name
-            ):
-                continue
-            if self.object_name and self.object_name != object_name:
-                continue
-            key = '{}::{}'.format(numerator['instance_name'], numerator['object_name'])
-            corresponding_base = base_by_key.get(key)
-
-            if not corresponding_base:
-                self.log.warning(
-                    'Could not find corresponding base counter for sql_name: %s base_name: %s',
-                    self.sql_name,
-                    self.base_name,
-                )
+        if self.instance == ALL_INSTANCES:
+            instance_names = list(numerators)
+        else:
+            instance_names = self._configured_instance_names()
 
-            metric_tags = list(self.tags)
+        for instance_name in instance_names:
             if self.instance == ALL_INSTANCES:
-                metric_tags.append('{}:{}'.format(self.tag_by, instance_name))
-            self.report_fraction(
-                numerator['cntr_value'], corresponding_base['cntr_value'], metric_tags, previous_values=values_cache
-            )
+                metric_tags = [*self.tags, '{}:{}'.format(self.tag_by, instance_name)]
+            else:
+                metric_tags = list(self.tags)
+
+            for object_name, cntr_value in numerators.get(instance_name, ()):
+                if self.object_name and self.object_name != object_name:
+                    continue
+                base_value = self._base_value(bases.get(instance_name), object_name)
+                if base_value is None:
+                    self.log.warning(
+                        'Could not find corresponding base counter for sql_name: %s base_name: %s',
+                        self.sql_name,
+                        self.base_name,
+                    )
+                    continue
+
+                self.report_fraction(cntr_value, base_value, metric_tags, previous_values=values_cache)
+
+    @staticmethod
+    def _base_value(base_counters: list[tuple[str, int]] | None, object_name: str) -> int | None:
+        """The value of the base counter recorded under the same object as the numerator, if there is one."""
+        for base_object_name, base_value in base_counters or ():
+            if base_object_name == object_name:
+                return base_value
+        return None
 
     def report_fraction(self, value, base, metric_tags, previous_values):
         try:
@@ -220,8 +229,6 @@ class SqlIncrFractionMetric(SqlFractionMetric):
     the current value and the base value (denominator) between two collection points that are one second apart.
     """
 
-    OPERATION_NAME = 'incr_fraction_metrics'
-
     def report_fraction(self, value, base, metric_tags, previous_values):
         # return if nil is passed as the values cache, as this should be instantiated
         # at check instantiation

sqlserver/datadog_checks/sqlserver/sqlserver.py

@@ -2,11 +2,13 @@
 # All rights reserved
 # Licensed under a 3-clause BSD style license (see LICENSE)
 
-from __future__ import division
+from __future__ import annotations, division
 
 import functools
 import time
 from collections import defaultdict
+from collections.abc import Iterable
+from typing import Any
 
 from cachetools import TTLCache
 
@@ -128,6 +130,10 @@
 
 KEY_PREFIX = "dbm-sqlserver-"
 
+# What a metric class's `fetch_all_values` hands back: the rows it fetched, in whatever shape that class
+# dispatches from, and their column names. Classes that index their own rows return None for the columns.
+MetricFetchResult = tuple[Any, list[str] | None]
+
 
 class SQLServer(DatabaseCheck):
     DBMS = "sqlserver"
@@ -997,31 +1003,7 @@ def load_basic_metrics(self, cursor):
         if self.autodiscover_databases(cursor) or not self.instance_metrics:
             self._make_metric_list_to_collect(self._config.custom_metrics)
 
-        instance_results = {}
-        engine_edition = self.static_info_cache.get(STATIC_INFO_ENGINE_EDITION, "")
-        # Execute the `fetch_all` operations first to minimize the database calls
-        for cls, metric_names in self.instance_per_type_metrics.items():
-            if not metric_names:
-                instance_results[cls] = None, None
-            else:
-                try:
-                    db_names = [d.name for d in self.databases] or [
-                        self.instance.get("database", self.connection.DEFAULT_DATABASE)
-                    ]
-                    metric_cls = getattr(metrics, cls)
-                    with tracked_query(self, operation=metric_cls.OPERATION_NAME):
-                        rows, cols = metric_cls.fetch_all_values(
-                            cursor,
-                            list(metric_names),
-                            self.log,
-                            databases=db_names,
-                            engine_edition=engine_edition,
-                        )
-                except Exception as e:
-                    self.log.error("Error running `fetch_all` for metrics %s - skipping.  Error: %s", cls, e)
-                    rows, cols = None, None
-
-                instance_results[cls] = rows, cols
+        instance_results = self._fetch_instance_results(cursor)
 
         for metric in self.instance_metrics:
             key = metric.__class__.__name__
@@ -1035,6 +1017,56 @@ def load_basic_metrics(self, cursor):
                     else:
                         metric.fetch_metric(rows, cols)
 
+    def _fetch_instance_results(self, cursor: Any) -> dict[str, MetricFetchResult]:
+        """Run the `fetch_all` of every metric class in use, keyed by class name.
+
+        Executing them up front keeps the number of database calls down, and the performance counter classes
+        share one call between them because they all read the same snapshot of a table that is expensive to
+        scan however few rows are wanted from it.
+        """
+        instance_results = {}
+        engine_edition = self.static_info_cache.get(STATIC_INFO_ENGINE_EDITION, "")
+        perf_counter_classes = []
+        perf_counter_names = set()
+
+        for cls, metric_names in self.instance_per_type_metrics.items():
+            if not metric_names:
+                instance_results[cls] = None, None
+            elif issubclass(getattr(metrics, cls), metrics.SqlPerfCounterMetric):
+                perf_counter_classes.append(cls)
+                perf_counter_names.update(metric_names)
+            else:
+                instance_results[cls] = self._fetch_all_values(cursor, cls, metric_names, engine_edition)
+
+        if perf_counter_classes:
+            perf_counter_results = self._fetch_all_values(
+                cursor, metrics.SqlPerfCounterMetric.__name__, perf_counter_names, engine_edition
+            )
+            for cls in perf_counter_classes:
+                instance_results[cls] = perf_counter_results
+
+        return instance_results
+
+    def _fetch_all_values(
+        self, cursor: Any, cls: str, metric_names: Iterable[str], engine_edition: str
+    ) -> MetricFetchResult:
+        try:
+            db_names = [d.name for d in self.databases] or [
+                self.instance.get("database", self.connection.DEFAULT_DATABASE)
+            ]
+            metric_cls = getattr(metrics, cls)
+            with tracked_query(self, operation=metric_cls.OPERATION_NAME):
+                return metric_cls.fetch_all_values(
+                    cursor,
+                    list(metric_names),
+                    self.log,
+                    databases=db_names,
+                    engine_edition=engine_edition,
+                )
+        except Exception as e:
+            self.log.error("Error running `fetch_all` for metrics %s - skipping.  Error: %s", cls, e)
+            return None, None
+
     def collect_metrics(self):
         """Fetch the metrics from all the associated database tables."""
         with self.connection.open_managed_default_connection(KEY_PREFIX):

sqlserver/tests/common.py

@@ -254,9 +254,7 @@ def get_local_driver():
 }
 
 OPERATION_TIME_METRICS = [
-    'simple_metrics',
-    'fraction_metrics',
-    'incr_fraction_metrics',
+    'perf_counter_metrics',
 ]
 
 OPERATION_TIME_METRIC_NAME = 'dd.sqlserver.operation.time'