Parameterize SQL Server agent history query (#24203)

* Parameterize SQL Server agent history query

* Add SQL Server changelog entry

Author: lu-zhengda

sqlserver/changelog.d/24203.fixed

@@ -0,0 +1 @@
+Parameterize SQL Server Agent job history collection to avoid plan cache churn.

sqlserver/datadog_checks/sqlserver/agent_history.py

@@ -94,7 +94,7 @@
 	message
 FROM HISTORY_ENTRIES
 WHERE
-    completion_epoch_time > {last_collection_time_filter};
+    completion_epoch_time > ?;
 """
 
 
@@ -136,14 +136,12 @@ def run_job(self):
 
     @tracked_method(agent_check_getter=agent_check_getter)
     def _get_new_agent_job_history(self, cursor):
-        last_collection_time_filter = "{last_collection_time}".format(last_collection_time=self._last_collection_time)
         history_row_limit_filter = "TOP {history_row_limit}".format(history_row_limit=self.history_row_limit)
-        query = AGENT_HISTORY_QUERY.format(
-            history_row_limit_filter=history_row_limit_filter, last_collection_time_filter=last_collection_time_filter
-        )
+        query = AGENT_HISTORY_QUERY.format(history_row_limit_filter=history_row_limit_filter)
+        params = (self._last_collection_time,)
         self.log.debug("collecting sql server agent jobs history")
-        self.log.debug("Running query [%s]", query)
-        cursor.execute(query)
+        self.log.debug("Running query [%s] %s", query, params)
+        cursor.execute(query, params)
         columns = [i[0] for i in cursor.description]
         # construct row dicts manually as there's no DictCursor for pyodbc
         rows = [dict(zip(columns, row)) for row in cursor.fetchall()]

sqlserver/tests/test_agent_jobs.py

@@ -5,10 +5,12 @@
 import logging
 import time
 from copy import copy
+from unittest.mock import Mock
 
 import pytest
 
 from datadog_checks.sqlserver import SQLServer
+from datadog_checks.sqlserver.agent_history import SqlserverAgentHistory
 
 from .common import (
     CHECK_NAME,
@@ -96,7 +98,7 @@
 	message
 FROM HISTORY_ENTRIES
 WHERE
-    completion_epoch_time > {last_collection_time_filter};
+    completion_epoch_time > ?;
 """
 
 
@@ -176,7 +178,7 @@
 	message
 FROM HISTORY_ENTRIES
 WHERE
-    completion_epoch_time > 10000;
+    completion_epoch_time > ?;
 """
 
 AGENT_ACTIVITY_DURATION_QUERY = """\
@@ -404,16 +406,52 @@ def test_connection_with_agent_history(instance_docker):
 
     with check.connection.open_managed_default_connection(KEY_PREFIX):
         with check.connection.get_managed_cursor(KEY_PREFIX) as cursor:
-            last_collection_time_filter = "{last_collection_time}".format(last_collection_time=10000)
             history_row_limit_filter = "TOP {history_row_limit}".format(history_row_limit=10000)
-            query = AGENT_HISTORY_QUERY.format(
-                history_row_limit_filter=history_row_limit_filter,
-                last_collection_time_filter=last_collection_time_filter,
-            )
-            cursor.execute(query)
+            query = AGENT_HISTORY_QUERY.format(history_row_limit_filter=history_row_limit_filter)
+            cursor.execute(query, (10000,))
             assert query == FORMATTED_HISTORY_QUERY
 
 
+class AgentHistoryCursor:
+    description = [('run_epoch_time',), ('run_duration_seconds',)]
+
+    def __init__(self) -> None:
+        self.executions: list[tuple[str, tuple[int, ...]]] = []
+
+    def execute(self, query: str, params: tuple[int, ...]) -> None:
+        self.executions.append((query, params))
+
+    def fetchall(self) -> list[tuple[int, int]]:
+        return []
+
+
+class AgentHistoryCheck:
+    name = CHECK_NAME
+
+    def __init__(self) -> None:
+        self.log = Mock()
+        self.count = Mock()
+        self.gauge = Mock()
+        self.histogram = Mock()
+
+
+def test_agent_history_query_parameterizes_last_collection_time() -> None:
+    check = AgentHistoryCheck()
+    agent_history = object.__new__(SqlserverAgentHistory)
+    agent_history._check = check
+    agent_history.log = check.log
+    agent_history.history_row_limit = 10000
+    agent_history._last_collection_time = 10000
+    cursor = AgentHistoryCursor()
+
+    agent_history._get_new_agent_job_history(cursor)
+
+    query, params = cursor.executions[0]
+    assert "completion_epoch_time > ?;" in query
+    assert "completion_epoch_time > 10000;" not in query
+    assert params == (10000,)
+
+
 @pytest.mark.usefixtures('dd_environment')
 def test_connection_with_agent_activity_duration(instance_docker):
     check = SQLServer(CHECK_NAME, {}, [instance_docker])
@@ -466,23 +504,15 @@ def test_history_output(instance_docker, sa_conn):
     check.initialize_connection()
     with check.connection.open_managed_default_connection(KEY_PREFIX):
         with check.connection.get_managed_cursor(KEY_PREFIX) as cursor:
-            last_collection_time_filter = "{last_collection_time}".format(last_collection_time=now - 1)
             history_row_limit_filter = "TOP {history_row_limit}".format(history_row_limit=10000)
-            query = AGENT_HISTORY_QUERY.format(
-                history_row_limit_filter=history_row_limit_filter,
-                last_collection_time_filter=last_collection_time_filter,
-            )
-            cursor.execute(query)
+            query = AGENT_HISTORY_QUERY.format(history_row_limit_filter=history_row_limit_filter)
+            cursor.execute(query, (now - 1,))
             results = cursor.fetchall()
             assert len(results) == 7, "should have 7 steps associated with completed jobs"
             assert len(results[0]) == 10, "should have 10 columns per step"
-            last_collection_time_filter = "{last_collection_time}".format(last_collection_time=now + 1)
             history_row_limit_filter = "TOP {history_row_limit}".format(history_row_limit=10000)
-            query = AGENT_HISTORY_QUERY.format(
-                history_row_limit_filter=history_row_limit_filter,
-                last_collection_time_filter=last_collection_time_filter,
-            )
-            cursor.execute(query)
+            query = AGENT_HISTORY_QUERY.format(history_row_limit_filter=history_row_limit_filter)
+            cursor.execute(query, (now + 1,))
             results = cursor.fetchall()
             assert len(results) == 4, (
                 "should only have 4 steps associated with completed jobs when filtering with last collection time"