Collect sqlserver deadlock data

sqlserver/assets/configuration/spec.yaml

@@ -760,6 +760,29 @@ files:
           value:
             type: number
             example: 10
+    - name: deadlocks_collection
+      hidden: True
+      description: |
+        Configure the collection of deadlock data. The feature is supported for odbc connector only.
+      options:
+        - name: enabled
+          description: |
+            Enable the collection of deadlock data. Requires `dbm: true`. Disabled by default.
+          value:
+            type: boolean
+            example: false
+        - name: collection_interval
+          description: |
+            Set the interval for collecting deadlock data, in seconds. Defaults to 600 seconds.
+          value:
+            type: number
+            example: 600
+        - name: max_deadlocks
+          description: |
+            Set the maximum number of deadlocks to retrieve per collection.
+          value:
+            type: number
+            example: 100
     - template: instances/default
   - template: logs
     example:

sqlserver/changelog.d/18108.added

@@ -0,0 +1 @@
+Added deadlock collection feature to the SQL Server integration.

sqlserver/datadog_checks/sqlserver/config.py

@@ -50,6 +50,7 @@ def __init__(self, init_config, instance, log):
         self.settings_config: dict = instance.get('collect_settings', {}) or {}
         self.activity_config: dict = instance.get('query_activity', {}) or {}
         self.schema_config: dict = instance.get('schemas_collection', {}) or {}
+        self.deadlocks_config: dict = instance.get('deadlocks_collection', {}) or {}
         self.cloud_metadata: dict = {}
         aws: dict = instance.get('aws', {}) or {}
         gcp: dict = instance.get('gcp', {}) or {}

sqlserver/datadog_checks/sqlserver/config_models/instance.py

@@ -68,6 +68,16 @@ class CustomQuery(BaseModel):
     tags: Optional[tuple[str, ...]] = None
 
 
+class DeadlocksCollection(BaseModel):
+    model_config = ConfigDict(
+        arbitrary_types_allowed=True,
+        frozen=True,
+    )
+    collection_interval: Optional[float] = None
+    enabled: Optional[bool] = None
+    max_deadlocks: Optional[float] = None
+
+
 class Gcp(BaseModel):
     model_config = ConfigDict(
         arbitrary_types_allowed=True,
@@ -185,6 +195,7 @@ class InstanceConfig(BaseModel):
     database_instance_collection_interval: Optional[float] = None
     db_fragmentation_object_names: Optional[tuple[str, ...]] = None
     dbm: Optional[bool] = None
+    deadlocks_collection: Optional[DeadlocksCollection] = None
     disable_generic_tags: Optional[bool] = None
     driver: Optional[str] = None
     dsn: Optional[str] = None

sqlserver/datadog_checks/sqlserver/deadlocks.py

@@ -0,0 +1,193 @@
+# (C) Datadog, Inc. 2024-present
+# All rights reserved
+# Licensed under a 3-clause BSD style license (see LICENSE)
+
+import xml.etree.ElementTree as ET
+from time import time
+
+from datadog_checks.base.utils.db.sql import compute_sql_signature
+from datadog_checks.base.utils.db.utils import DBMAsyncJob, default_json_event_encoding, obfuscate_sql_with_metadata
+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.const import STATIC_INFO_ENGINE_EDITION, STATIC_INFO_VERSION
+from datadog_checks.sqlserver.queries import DEADLOCK_QUERY, DEADLOCK_TIMESTAMP_ALIAS, DEADLOCK_XML_ALIAS
+
+try:
+    import datadog_agent
+except ImportError:
+    from ..stubs import datadog_agent
+
+DEFAULT_COLLECTION_INTERVAL = 600
+MAX_DEADLOCKS = 100
+MAX_PAYLOAD_BYTES = 19e6
+
+PAYLOAD_TIMESTAMP = "deadlock_timestamp"
+PAYLOAD_QUERY_SIGNATURE = "query_signatures"
+PAYLOAD_XML = "xml"
+
+
+def agent_check_getter(self):
+    return self._check
+
+
+class Deadlocks(DBMAsyncJob):
+    def __init__(self, check, config: SQLServerConfig):
+        self.tags = [t for t in check.tags if not t.startswith('dd.internal')]
+        self._check = check
+        self._log = self._check.log
+        self._config = config
+        self._last_deadlock_timestamp = time()
+        self._max_deadlocks = config.deadlocks_config.get("max_deadlocks", MAX_DEADLOCKS)
+        self._deadlock_payload_max_bytes = MAX_PAYLOAD_BYTES
+        self.collection_interval = config.deadlocks_config.get("collection_interval", DEFAULT_COLLECTION_INTERVAL)
+        super(Deadlocks, self).__init__(
+            check,
+            run_sync=True,
+            enabled=self._config.deadlocks_config.get('enabled', False),
+            expected_db_exceptions=(),
+            min_collection_interval=self._config.min_collection_interval,
+            dbms="sqlserver",
+            rate_limit=1 / float(self.collection_interval),
+            job_name="deadlocks",
+            shutdown_callback=self._close_db_conn,
+        )
+        self._conn_key_prefix = "dbm-deadlocks-"
+
+    def _close_db_conn(self):
+        pass
+
+    def obfuscate_no_except_wrapper(self, sql_text):
+        try:
+            sql_text = obfuscate_sql_with_metadata(
+                sql_text, self._config.obfuscator_options, replace_null_character=True
+            )['query']
+        except Exception as e:
+            sql_text = "ERROR: failed to obfuscate"
+            error_text = "Failed to obfuscate sql text within a deadlock"
+            if self._config.log_unobfuscated_queries:
+                error_text += "=[%s]" % sql_text
+            error_text += " | err=[%s]"
+            self._log.error(error_text, e)
+        return sql_text
+
+    def _obfuscate_xml(self, root):
+        process_list = root.find(".//process-list")
+        if process_list is None:
+            raise Exception("process-list element not found. The deadlock XML is in an unexpected format.")
+        query_signatures = []
+        for process in process_list.findall('process'):
+            for inputbuf in process.findall('.//inputbuf'):
+                if inputbuf.text is not None:
+                    inputbuf.text = self.obfuscate_no_except_wrapper(inputbuf.text)
+                    spid = process.get('spid')
+                    if spid is not None:
+                        try:
+                            spid = int(spid)
+                        except ValueError:
+                            self._log.error("spid not an integer. Skipping query signature computation.")
+                            continue
+                        if spid in query_signatures:
+                            continue
+                        query_signatures.append({"spid": spid, "signature": compute_sql_signature(inputbuf.text)})
+                    else:
+                        self._log.error("spid not found in process element. Skipping query signature computation.")
+            for frame in process.findall('.//frame'):
+                if frame.text is not None:
+                    frame.text = self.obfuscate_no_except_wrapper(frame.text)
+        return query_signatures
+
+    def _get_lookback_seconds(self):
+        return min(-60, self._last_deadlock_timestamp - time())
+
+    def _query_deadlocks(self):
+        with self._check.connection.open_managed_default_connection(key_prefix=self._conn_key_prefix):
+            with self._check.connection.get_managed_cursor(key_prefix=self._conn_key_prefix) as cursor:
+                self._log.debug("collecting sql server deadlocks")
+                self._log.debug(
+                    "Running query [%s] with max deadlocks %s and timestamp %s",
+                    DEADLOCK_QUERY,
+                    self._max_deadlocks,
+                    self._last_deadlock_timestamp,
+                )
+                try:
+                    cursor.execute(DEADLOCK_QUERY, (self._max_deadlocks, self._get_lookback_seconds()))
+                except Exception as e:
+                    if "Data column of Unknown ADO type" in str(e):
+                        raise Exception(f"{str(e)} | cursor.description: {cursor.description}")
+                    raise e
+
+                columns = [column[0] for column in cursor.description]
+                return [dict(zip(columns, row)) for row in cursor.fetchall()]
+
+    def _create_deadlock_rows(self):
+        db_rows = self._query_deadlocks()
+        deadlock_events = []
+        total_number_of_characters = 0
+        for i, row in enumerate(db_rows):
+            try:
+                root = ET.fromstring(row[DEADLOCK_XML_ALIAS])
+            except Exception as e:
+                self._log.error(
+                    """An error occurred while collecting SQLServer deadlocks.
+                        One of the deadlock XMLs couldn't be parsed. The error: {}. XML: {}""".format(
+                        e, row
+                    )
+                )
+                continue
+            query_signatures = {}
+            try:
+                query_signatures = self._obfuscate_xml(root)
+            except Exception as e:
+                error = "An error occurred while obfuscating SQLServer deadlocks. The error: {}".format(e)
+                self._log.error(error)
+                continue
+
+            total_number_of_characters += len(row) + len(query_signatures)
+            if total_number_of_characters > self._deadlock_payload_max_bytes:
+                self._log.warning(
+                    """We've dropped {} deadlocks from a total of {} deadlocks as the
+                     max deadlock payload of {} bytes was exceeded.""".format(
+                        len(db_rows) - i, len(db_rows), self._deadlock_payload_max_bytes
+                    )
+                )
+                break
+
+            deadlock_events.append(
+                {
+                    PAYLOAD_TIMESTAMP: row[DEADLOCK_TIMESTAMP_ALIAS],
+                    PAYLOAD_XML: ET.tostring(root, encoding='unicode'),
+                    PAYLOAD_QUERY_SIGNATURE: query_signatures,
+                }
+            )
+        self._last_deadlock_timestamp = time()
+        return deadlock_events
+
+    @tracked_method(agent_check_getter=agent_check_getter)
+    def collect_deadlocks(self):
+        rows = self._create_deadlock_rows()
+        # Send payload only if deadlocks found
+        if rows:
+            deadlocks_event = self._create_deadlock_event(rows)
+            payload = json.dumps(deadlocks_event, default=default_json_event_encoding)
+            self._log.debug("Deadlocks payload: %s", str(payload))
+            self._check.database_monitoring_query_activity(payload)
+
+    def _create_deadlock_event(self, deadlock_rows):
+        event = {
+            "host": self._check.resolved_hostname,
+            "ddagentversion": datadog_agent.get_version(),
+            "ddsource": "sqlserver",
+            "dbm_type": "deadlocks",
+            "collection_interval": self.collection_interval,
+            "ddtags": self.tags,
+            "timestamp": time() * 1000,
+            'sqlserver_version': self._check.static_info_cache.get(STATIC_INFO_VERSION, ""),
+            'sqlserver_engine_edition': self._check.static_info_cache.get(STATIC_INFO_ENGINE_EDITION, ""),
+            "cloud_metadata": self._config.cloud_metadata,
+            "sqlserver_deadlocks": deadlock_rows,
+        }
+        return event
+
+    def run_job(self):
+        self.collect_deadlocks()

sqlserver/datadog_checks/sqlserver/queries.py

@@ -214,6 +214,23 @@
     FK.name, FK.parent_object_id, FK.referenced_object_id;
 """
 
+DEADLOCK_TIMESTAMP_ALIAS = "timestamp"
+DEADLOCK_XML_ALIAS = "event_xml"
+DEADLOCK_QUERY = """
+SELECT TOP(?) xdr.value('@timestamp', 'datetime') AS [{timestamp}],
+    xdr.query('.') AS [{xml}]
+FROM (SELECT CAST([target_data] AS XML) AS Target_Data
+            FROM sys.dm_xe_session_targets AS xt
+            INNER JOIN sys.dm_xe_sessions AS xs ON xs.address = xt.event_session_address
+            WHERE xs.name = N'system_health'
+              AND xt.target_name = N'ring_buffer'
+    ) AS XML_Data
+CROSS APPLY Target_Data.nodes('RingBufferTarget/event[@name="xml_deadlock_report"]') AS XEventData(xdr)
+WHERE xdr.value('@timestamp', 'datetime') >= DATEADD(SECOND, ?, GETDATE())
+;""".format(
+    **{"timestamp": DEADLOCK_TIMESTAMP_ALIAS, "xml": DEADLOCK_XML_ALIAS}
+)
+
 
 def get_query_ao_availability_groups(sqlserver_major_version):
     """

sqlserver/datadog_checks/sqlserver/schemas.py

@@ -32,7 +32,6 @@
 
 
 class SubmitData:
-
     def __init__(self, submit_data_function, base_event, logger):
         self._submit_to_agent_queue = submit_data_function
         self._base_event = base_event

sqlserver/datadog_checks/sqlserver/sqlserver.py

@@ -1,6 +1,7 @@
 # (C) Datadog, Inc. 2018-present
 # All rights reserved
 # Licensed under a 3-clause BSD style license (see LICENSE)
+
 from __future__ import division
 
 import copy
@@ -12,7 +13,11 @@
 from datadog_checks.base import AgentCheck
 from datadog_checks.base.config import is_affirmative
 from datadog_checks.base.utils.db import QueryExecutor, QueryManager
-from datadog_checks.base.utils.db.utils import default_json_event_encoding, resolve_db_host, tracked_query
+from datadog_checks.base.utils.db.utils import (
+    default_json_event_encoding,
+    resolve_db_host,
+    tracked_query,
+)
 from datadog_checks.base.utils.serialization import json
 from datadog_checks.sqlserver.activity import SqlserverActivity
 from datadog_checks.sqlserver.agent_history import SqlserverAgentHistory
@@ -23,6 +28,7 @@
     SqlserverDBFragmentationMetrics,
     SqlserverIndexUsageMetrics,
 )
+from datadog_checks.sqlserver.deadlocks import Deadlocks
 from datadog_checks.sqlserver.metadata import SqlserverMetadata
 from datadog_checks.sqlserver.schemas import Schemas
 from datadog_checks.sqlserver.statements import SqlserverStatementMetrics
@@ -135,6 +141,7 @@ def __init__(self, name, init_config, instances):
         self.sql_metadata = SqlserverMetadata(self, self._config)
         self.activity = SqlserverActivity(self, self._config)
         self.agent_history = SqlserverAgentHistory(self, self._config)
+        self.deadlocks = Deadlocks(self, self._config)
 
         self.static_info_cache = TTLCache(
             maxsize=100,
@@ -171,6 +178,7 @@ def cancel(self):
         self.activity.cancel()
         self.sql_metadata.cancel()
         self._schemas.cancel()
+        self.deadlocks.cancel()
 
     def config_checks(self):
         if self._config.autodiscovery and self.instance.get("database"):
@@ -785,6 +793,7 @@ def check(self, _):
                 self.activity.run_job_loop(self.tags)
                 self.sql_metadata.run_job_loop(self.tags)
                 self._schemas.run_job_loop(self.tags)
+                self.deadlocks.run_job_loop(self.tags)
         else:
             self.log.debug("Skipping check")
 

sqlserver/hatch.toml

@@ -14,7 +14,7 @@ setup = ["single", "ha"]
 [[envs.default.matrix]]
 python = ["3.12"]
 os = ["windows"]
-driver = ["SQLOLEDB", "SQLNCLI11", "MSOLEDBSQL", "odbc"]
+driver = ["SQLOLEDB", "MSOLEDBSQL", "odbc"]
 version = ["2019", "2022"]
 setup = ["single"]
 

sqlserver/tests/compose-ha/sql/aoag_primary.sql

@@ -121,6 +121,21 @@ CREATE USER bob FOR LOGIN bob;
 CREATE USER fred FOR LOGIN fred;
 GO
 
+-- Create a simple table for deadlocks
+CREATE TABLE [datadog_test-1].dbo.deadlocks (a int PRIMARY KEY not null ,b int null); 
+
+INSERT INTO [datadog_test-1].dbo.deadlocks VALUES (1,10),(2,20),(3,30) 
+
+-- Grant permissions to bob and fred to update the deadlocks table
+GRANT INSERT ON [datadog_test-1].dbo.deadlocks TO bob;
+GRANT UPDATE ON [datadog_test-1].dbo.deadlocks TO bob;
+GRANT DELETE ON [datadog_test-1].dbo.deadlocks TO bob;
+
+GRANT INSERT ON [datadog_test-1].dbo.deadlocks TO fred;
+GRANT UPDATE ON [datadog_test-1].dbo.deadlocks TO fred;
+GRANT DELETE ON [datadog_test-1].dbo.deadlocks TO fred;
+GO
+
 EXEC sp_addrolemember 'db_datareader', 'bob'
 EXEC sp_addrolemember 'db_datareader', 'fred'
 EXEC sp_addrolemember 'db_datawriter', 'bob'