sap_hana: address DO review feedback (cancel, query key, query_timeout)

Key the Data Observability scheduler off the monitor_id when the RC payload carries
a real (non-zero) one — the forward-looking contract shared with Postgres — and fall
back to a hash of the query text otherwise. This matters because the payload delivered
to agents today carries no per-query monitor_id (it decodes to 0/None); keying purely
off monitor_id would collapse every query onto 0. The query-text fallback is safe
because the SQL embeds the monitor id(s) it serves (in a trailing Datadog comment, or
the column alias for custom SQL), so distinct monitors always produce distinct query
text. The key upgrades to monitor_id automatically once the backend starts emitting it.

Rename the per-query timeout_seconds option to query_timeout in milliseconds to align
with the Postgres DO config and the agent RC handler, which delivers query_timeout as
timeout_seconds * 1000. Add SapHanaCheck.cancel() so the DBMAsyncJob thread pool is
released when the check is unscheduled (cluster-agent flavor / one-off runs). Apply
reviewer doc suggestions in the README and spec.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

Author: pawel-big-lebowski

sap_hana/README.md

@@ -153,7 +153,7 @@ See the [sample sap_hana.d/conf.yaml][4] for all available schema collection opt
 
 The Datadog backend can deliver monitoring queries to the SAP HANA check via Remote Configuration. When enabled, the Agent executes these queries against HANA on a schedule and forwards the results as Data Observability events.
 
-To allow Remote Configuration to push query configs to the `sap_hana` check, add it to the allow list in `datadog.yaml`:
+To allow Remote Configuration to push query configs to the `sap_hana` check, add `sap_hana` to the allowlist in `datadog.yaml`:
 
 ```yaml
 remote_configuration:
@@ -162,9 +162,9 @@ remote_configuration:
       - sap_hana
 ```
 
-Without this entry, the Agent silently drops any query dispatched by the backend with no error surfaced. After updating `datadog.yaml`, [restart the Agent][5].
+Without this entry, the Agent silently drops any query delivered by the backend without surfacing an error. After updating `datadog.yaml`, [restart the Agent][5].
 
-Data Observability query actions require schema collection to be enabled. Ensure the `collect_schemas` block is present and `enabled: true` in your `sap_hana.d/conf.yaml` (see [Schema collection](#schema-collection)).
+Data Observability query actions require schema collection to be enabled. Verify that the `collect_schemas` block is present and `enabled: true` in your `sap_hana.d/conf.yaml` (see [Schema collection](#schema-collection)).
 
 ### Validation
 

sap_hana/assets/configuration/spec.yaml

@@ -161,7 +161,7 @@ files:
         fleet_configurable: true
         description: |
           Configure the Data Observability async job, which executes monitoring
-          queries delivered via Remote Configuration.
+          queries delivered through Remote Configuration.
         options:
           - name: enabled
             fleet_configurable: true
@@ -224,9 +224,9 @@ files:
                     schedule wins and interval_seconds is ignored. If neither is set, the
                     query is skipped at runtime with a warning.
                   type: string
-                - name: timeout_seconds
+                - name: query_timeout
                   description: |
-                    Statement timeout for this query in seconds. Applied as a connection-level
+                    Statement timeout for this query in milliseconds. Applied as a connection-level
                     statementTimeout when executing the query.
                   type: integer
                 - name: dbname

sap_hana/datadog_checks/sap_hana/config_models/instance.py

@@ -84,13 +84,13 @@ class Query(BaseModel):
     )
     monitor_id: Optional[int] = None
     query: str
-    schedule: Optional[str] = Field(
+    query_timeout: Optional[int] = Field(
         None,
-        description='A standard 5-field cron expression (minute hour dom month dow) specifying\nwhen to run this query. When both schedule and interval_seconds are set,\nschedule wins and interval_seconds is ignored. If neither is set, the\nquery is skipped at runtime with a warning.\n',
+        description='Statement timeout for this query in milliseconds. Applied as a connection-level\nstatementTimeout when executing the query.\n',
     )
-    timeout_seconds: Optional[int] = Field(
+    schedule: Optional[str] = Field(
         None,
-        description='Statement timeout for this query in seconds. Applied as a connection-level\nstatementTimeout when executing the query.\n',
+        description='A standard 5-field cron expression (minute hour dom month dow) specifying\nwhen to run this query. When both schedule and interval_seconds are set,\nschedule wins and interval_seconds is ignored. If neither is set, the\nquery is skipped at runtime with a warning.\n',
     )
     type: Optional[str] = None
 

sap_hana/datadog_checks/sap_hana/data_observability.py

@@ -29,13 +29,23 @@
 
 CRON_STARTUP_LOOKBACK_SECONDS = 300
 
-DEFAULT_DO_QUERY_TIMEOUT_S = 60
+DEFAULT_DO_QUERY_TIMEOUT_MS = 60_000
 
 Mode = Literal["cron", "interval"]
 
 
 def _query_key(q: Query) -> str:
-    """Stable per-query scheduling key derived from the query text."""
+    """Stable per-query scheduling key.
+
+    Prefer the monitor id when the RC payload carries a real one — this is the
+    forward-looking contract shared with Postgres. Today the payload delivers no
+    per-query monitor id (it decodes to 0/None), so fall back to a hash of the query
+    text. That is sufficient because the SQL embeds the monitor id(s) it serves (in a
+    trailing "-- Datadog {\"monitor_ids\":[...]}" comment, or the column alias for
+    custom SQL), so distinct monitors always produce distinct query text.
+    """
+    if q.monitor_id:  # real, non-zero monitor id
+        return f"monitor:{q.monitor_id}"
     return hashlib.sha256(q.query.encode()).hexdigest()[:16]
 
 
@@ -53,7 +63,7 @@ def __init__(self, check: SapHanaCheck, do_config: DataObservability) -> None:
         self._do_config = do_config
         self._last_execution: dict[str, float] = {}
         self._do_conn: Any = None
-        self._do_conn_timeout_s: int | None = None
+        self._do_conn_timeout_ms: int | None = None
 
         collection_interval = do_config.collection_interval or 10
         super().__init__(
@@ -140,15 +150,15 @@ def _build_base_tags(self) -> list[str]:
         tags.append('db_type:saphana')
         return tags
 
-    def _get_connection(self, timeout_seconds: int) -> Any:
+    def _get_connection(self, timeout_ms: int) -> Any:
         """Return the persistent DO connection, (re)creating it when the timeout changes."""
-        if self._do_conn is not None and self._do_conn_timeout_s == timeout_seconds:
+        if self._do_conn is not None and self._do_conn_timeout_ms == timeout_ms:
             return self._do_conn
         if self._do_conn is not None:
             self._log.debug(
-                "Data Observability: reopening DO connection (timeout changed from %ds to %ds).",
-                self._do_conn_timeout_s,
-                timeout_seconds,
+                "Data Observability: reopening DO connection (timeout changed from %dms to %dms).",
+                self._do_conn_timeout_ms,
+                timeout_ms,
             )
             try:
                 self._do_conn.close()
@@ -161,7 +171,7 @@ def _get_connection(self, timeout_seconds: int) -> Any:
         # its own connection (instead of reusing self._check._conn) because it
         # needs a per-query statementTimeout that must not affect main-check queries.
         conn_props = self._check._get_connection_properties()
-        conn_props['statementTimeout'] = timeout_seconds * 1000
+        conn_props['statementTimeout'] = timeout_ms
         try:
             self._do_conn = hana_connect(**conn_props)
         except HanaError as e:
@@ -174,14 +184,14 @@ def _get_connection(self, timeout_seconds: int) -> Any:
             raise
         self._log.debug(
             "Data Observability: DO connection opened (statementTimeout=%dms).",
-            timeout_seconds * 1000,
+            timeout_ms,
         )
-        self._do_conn_timeout_s = timeout_seconds
+        self._do_conn_timeout_ms = timeout_ms
         return self._do_conn
 
     def _execute_single_query(self, query_spec: Query) -> dict[str, Any]:
-        timeout_s = query_spec.timeout_seconds or DEFAULT_DO_QUERY_TIMEOUT_S
-        conn = self._get_connection(timeout_s)
+        timeout_ms = query_spec.query_timeout or DEFAULT_DO_QUERY_TIMEOUT_MS
+        conn = self._get_connection(timeout_ms)
         start = time.time()
         try:
             if self._cancel_event.is_set():
@@ -211,7 +221,7 @@ def _execute_single_query(self, query_spec: Query) -> dict[str, Any]:
             )
             self._log.warning("Data Observability: resetting DO connection after query failure.")
             self._do_conn = None
-            self._do_conn_timeout_s = None
+            self._do_conn_timeout_ms = None
             return {
                 'status': 'error',
                 'columns': [],

sap_hana/datadog_checks/sap_hana/sap_hana.py

@@ -159,6 +159,12 @@ def check(self, _):
                 )
                 self._connection_flaked = False
 
+    def cancel(self):
+        # Signal the Data Observability async job to stop so its executor thread is
+        # released when the check is unscheduled (e.g. cluster-agent flavor or one-off
+        # check invocations), instead of leaking the DBMAsyncJob thread pool.
+        self.data_observability.cancel()
+
     def set_default_methods(self):
         self._default_methods.extend(
             [

sap_hana/tests/test_data_observability.py

@@ -11,7 +11,7 @@
 
 from datadog_checks.sap_hana.config_models.instance import CustomSqlSelectFields, Entity, Query
 from datadog_checks.sap_hana.data_observability import (
-    DEFAULT_DO_QUERY_TIMEOUT_S,
+    DEFAULT_DO_QUERY_TIMEOUT_MS,
     SapHanaDataObservability,
     _query_key,
 )
@@ -39,7 +39,7 @@ def _make_do(queries=(), config_id=None, tags=None):
     do._do_config = do_config
     do._last_execution = {}
     do._do_conn = None
-    do._do_conn_timeout_s = None
+    do._do_conn_timeout_ms = None
     do._log = mock.MagicMock()
     do._cancel_event = threading.Event()
     do._tags = tags
@@ -52,15 +52,15 @@ def _q(
     interval_seconds=None,
     schedule=None,
     monitor_id=None,
-    timeout_seconds=None,
+    query_timeout=None,
     dbname=None,
 ):
     return Query(
         query=sql,
         interval_seconds=interval_seconds,
         schedule=schedule,
         monitor_id=monitor_id,
-        timeout_seconds=timeout_seconds,
+        query_timeout=query_timeout,
         dbname=dbname,
     )
 
@@ -79,17 +79,33 @@ def _mock_conn(columns, rows):
 
 
 class TestQueryKey:
-    def test_same_sql_same_key(self):
+    def test_same_query_same_key(self):
         assert _query_key(_q('SELECT 1')) == _query_key(_q('SELECT 1'))
 
     def test_different_sql_different_key(self):
+        # Distinct monitors always yield distinct SQL (monitor ids are embedded in the query).
         assert _query_key(_q('SELECT 1')) != _query_key(_q('SELECT 2'))
 
     def test_key_is_16_hex_chars(self):
         key = _query_key(_q('SELECT 1'))
         assert len(key) == 16
         assert all(c in '0123456789abcdef' for c in key)
 
+    def test_nonzero_monitor_id_used_as_key(self):
+        assert _query_key(_q('SELECT 1', monitor_id=42)) == 'monitor:42'
+
+    def test_different_monitor_ids_different_key(self):
+        # With a real monitor id, identity comes from the id, not the SQL.
+        assert _query_key(_q('SELECT 1', monitor_id=1)) != _query_key(_q('SELECT 1', monitor_id=2))
+
+    def test_same_monitor_id_same_key_regardless_of_sql(self):
+        assert _query_key(_q('SELECT 1', monitor_id=7)) == _query_key(_q('SELECT 2', monitor_id=7))
+
+    def test_zero_monitor_id_falls_back_to_query_hash(self):
+        # monitor_id 0 (the omitempty default) and None both fall back to the query hash.
+        assert _query_key(_q('SELECT 1', monitor_id=0)) == _query_key(_q('SELECT 1', monitor_id=None))
+        assert _query_key(_q('SELECT 1', monitor_id=0)) == _query_key(_q('SELECT 1'))
+
 
 # ── _filter_valid_queries ──────────────────────────────────────────────────────
 
@@ -121,11 +137,11 @@ def test_invalid_cron_rejected(self):
         assert do._queries == ()
 
     def test_mixed_valid_and_invalid(self):
-        queries = [_q(interval_seconds=30), _q(), _q(schedule='*/10 * * * *')]
+        queries = [_q('SELECT 1', interval_seconds=30), _q('SELECT 2'), _q('SELECT 3', schedule='*/10 * * * *')]
         do = _make_do(queries)
         assert len(do._queries) == 2
 
-    def test_cron_scheduler_keyed_by_query_hash(self):
+    def test_cron_scheduler_keyed_by_query_key(self):
         q = _q(schedule='0 * * * *')
         do = _make_do([q])
         assert _query_key(q) in do._schedulers
@@ -136,7 +152,7 @@ def test_interval_query_has_no_scheduler(self):
         assert _query_key(q) not in do._schedulers
 
     def test_skipped_queries_emit_warning(self):
-        do = _make_do([_q(), _q(interval_seconds=30)])
+        do = _make_do([_q('SELECT 1'), _q('SELECT 2', interval_seconds=30)])
         do._log.warning.assert_called()
 
 
@@ -237,7 +253,7 @@ def test_monitor_id_present_when_set(self):
         assert payload['monitor_id'] == 42
 
     def test_monitor_id_absent_when_none(self):
-        q = _q()
+        q = _q()  # monitor_id is None
         payload = _make_do()._build_event_payload(q, self._RESULT)
         assert 'monitor_id' not in payload
 
@@ -300,7 +316,7 @@ def test_success_returns_columns_and_rows(self):
         do = _make_do([q])
         conn, _ = _mock_conn(['id', 'name'], [[1, 'alice'], [2, 'bob']])
         do._do_conn = conn
-        do._do_conn_timeout_s = DEFAULT_DO_QUERY_TIMEOUT_S
+        do._do_conn_timeout_ms = DEFAULT_DO_QUERY_TIMEOUT_MS
 
         result = do._execute_single_query(q)
 
@@ -318,7 +334,7 @@ def test_hana_error_returns_error_status(self):
         conn, cursor = _mock_conn(['id'], [])
         cursor.execute.side_effect = HanaError('connection lost')
         do._do_conn = conn
-        do._do_conn_timeout_s = DEFAULT_DO_QUERY_TIMEOUT_S
+        do._do_conn_timeout_ms = DEFAULT_DO_QUERY_TIMEOUT_MS
 
         result = do._execute_single_query(q)
 
@@ -335,34 +351,34 @@ def test_hana_error_resets_connection(self):
         conn, cursor = _mock_conn([], [])
         cursor.execute.side_effect = HanaError('timeout')
         do._do_conn = conn
-        do._do_conn_timeout_s = DEFAULT_DO_QUERY_TIMEOUT_S
+        do._do_conn_timeout_ms = DEFAULT_DO_QUERY_TIMEOUT_MS
 
         do._execute_single_query(q)
 
         assert do._do_conn is None
-        assert do._do_conn_timeout_s is None
+        assert do._do_conn_timeout_ms is None
 
     def test_uses_default_timeout_when_none(self):
-        q = _q(interval_seconds=60)  # timeout_seconds is None
+        q = _q(interval_seconds=60)  # query_timeout is None
         do = _make_do([q])
         called_with = []
 
-        def fake_get_connection(timeout_s):
-            called_with.append(timeout_s)
+        def fake_get_connection(timeout_ms):
+            called_with.append(timeout_ms)
             conn, _ = _mock_conn(['x'], [])
             return conn
 
         do._get_connection = fake_get_connection
         do._execute_single_query(q)
-        assert called_with == [DEFAULT_DO_QUERY_TIMEOUT_S]
+        assert called_with == [DEFAULT_DO_QUERY_TIMEOUT_MS]
 
     def test_uses_query_timeout_when_set(self):
-        q = _q(interval_seconds=60, timeout_seconds=120)
+        q = _q(interval_seconds=60, query_timeout=120)
         do = _make_do([q])
         called_with = []
 
-        def fake_get_connection(timeout_s):
-            called_with.append(timeout_s)
+        def fake_get_connection(timeout_ms):
+            called_with.append(timeout_ms)
             conn, _ = _mock_conn(['x'], [])
             return conn