sap_hana: run schema collection in a DBMAsyncJob background thread (#24128)

* sap_hana: run schema collection in a DBMAsyncJob background thread

Schema collection previously blocked the main check loop on every run.
Wrap HanaSchemaCollector in HanaSchemaCollectionJob (DBMAsyncJob) so it
runs in a background thread at the configured collection_interval, using
its own dedicated hdbcli connection — the same pattern used by the DO job.

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

* sap_hana: add changelog entry for async schema collection PR #24128

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

* sap_hana: address review feedback on schema collection job

- cancel() now stops the schema collection job too, so its background
  thread and dedicated HANA connection are released promptly on teardown
  instead of lingering until the inactivity timeout.
- Reset the job connection when a HANA error is swallowed inside
  collect_schemas(): the base SchemaCollector catches per-database errors
  and returns, so a transient disconnect never reached run_job's handler
  and the dead connection was reused every cycle. The collector now drops
  its connection reference on HanaError and the job reconnects next cycle.

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

* sap_hana: correct misleading cancel() comment

cancel() only sets each job's cancel event; it does not itself close the
schema job's dedicated HANA connection. Reword the comment to reflect what
cancel() actually does, per DBM review feedback.

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

* sap_hana: harden schema job connection and version handling

Address review feedback on the async schema collection job:

- Clamp a fractional collection_interval (< 1) so it no longer truncates
  to 0 and raises ZeroDivisionError while building rate_limit, which took
  down the whole check on construction.
- Replace the overloaded _conn sentinel with an explicit set_connection()
  + connection_lost contract, and drop the dead check._conn fallback so
  the catalog query can never silently run on the main check connection.
- Resolve dbms_version on the schema job's dedicated connection via
  _resolve_dbms_version(conn) instead of querying the main check
  connection off-thread, removing the race and the 'unknown' poisoning.

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

---------

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

Author: pawel-big-lebowski

sap_hana/assets/configuration/spec.yaml

@@ -178,6 +178,13 @@ files:
                 type: string
               example:
                 - "TMP_SCHEMA"
+          - name: run_sync
+            hidden: true
+            description: Run the schema collection job synchronously.
+            value:
+              type: boolean
+              example: false
+              display_default: false
       - name: data_observability
         hidden: true
         display_priority: 0

sap_hana/changelog.d/24128.added

@@ -0,0 +1 @@
+Run SAP HANA schema collection in a background thread to avoid blocking the main check loop.

sap_hana/datadog_checks/sap_hana/config_models/instance.py

@@ -35,6 +35,7 @@ class CollectSchemas(BaseModel):
     max_columns: Optional[int] = None
     max_tables: Optional[int] = None
     max_views: Optional[int] = None
+    run_sync: Optional[bool] = None
 
 
 class CustomQuery(BaseModel):

sap_hana/datadog_checks/sap_hana/sap_hana.py

@@ -4,7 +4,6 @@
 from __future__ import division
 
 import functools
-import time
 from collections import defaultdict
 from contextlib import closing
 from datetime import datetime
@@ -32,7 +31,7 @@
 from .data_observability import SapHanaDataObservability
 from .diagnose import run_diagnostics
 from .exceptions import OperationalError, QueryExecutionError
-from .schemas import HanaSchemaCollector
+from .schemas import HanaSchemaCollectionJob
 from .utils import compute_percent, positive
 
 
@@ -93,11 +92,9 @@ def __init__(self, name, init_config, instances):
         # Save master database hostname to act as the default if `use_hana_hostnames` is true
         self._master_hostname = None
 
-        # Schema collection (DBM)
+        # Schema collection async job (DBM)
         collect_schemas = CollectSchemas(**(self.instance.get('collect_schemas') or {}))
-        self._schema_collector = HanaSchemaCollector(self, collect_schemas) if collect_schemas.enabled else None
-        self._schema_collection_interval = int(collect_schemas.collection_interval or 600)
-        self._last_schema_collection_time = 0
+        self._schema_collection_job = HanaSchemaCollectionJob(self, collect_schemas)
         self._dbms_version = None
 
         # Data Observability async job (RC-delivered queries)
@@ -131,7 +128,7 @@ def check(self, _):
                 except Exception as e:
                     self.log.exception('Unexpected error running `%s`: %s', query_method.__name__, str(e))
                     continue
-            self._maybe_collect_schemas()
+            self._schema_collection_job.run_job_loop(self._tags)
             if self._do_config.enabled:
                 self.data_observability.run_job_loop(self._tags)
         finally:
@@ -162,9 +159,11 @@ 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.
+        # Signal both async jobs to stop so their executor threads are released when the
+        # check is unscheduled (e.g. cluster-agent flavor or one-off check invocations),
+        # instead of leaking the shared DBMAsyncJob thread pool. This only sets each job's
+        # cancel event; it does not itself close the schema job's dedicated HANA connection.
+        self._schema_collection_job.cancel()
         self.data_observability.cancel()
 
     def set_default_methods(self):
@@ -220,16 +219,30 @@ def dbms(self):
 
     @property
     def dbms_version(self):
-        if self._dbms_version is None and self._conn is not None:
-            try:
-                with closing(self._conn.cursor()) as cursor:
-                    cursor.execute("SELECT VERSION FROM SYS.M_DATABASE")
-                    row = cursor.fetchone()
-                    self._dbms_version = str(row[0]).split()[0] if row else 'unknown'
-            except Exception:
-                self._dbms_version = 'unknown'
+        # Only returns the cached value; resolution happens via _resolve_dbms_version on the
+        # schema job's dedicated connection. The version is read from base_event on the job
+        # thread, and querying self._conn here would race with the main check loop's concurrent
+        # use of that same connection (there is no thread-safe pool, unlike DBM integrations).
         return self._dbms_version or 'unknown'
 
+    def _resolve_dbms_version(self, conn):
+        """Resolve and cache the HANA version using the given connection.
+
+        Called from the schema-collection job thread on its dedicated connection so the main
+        check connection is never touched off-thread. Caches on first success; a transient
+        failure leaves the value unresolved so the next cycle retries instead of caching
+        'unknown' permanently.
+        """
+        if self._dbms_version is not None:
+            return
+        try:
+            with closing(conn.cursor()) as cursor:
+                cursor.execute("SELECT VERSION FROM SYS.M_DATABASE")
+                row = cursor.fetchone()
+                self._dbms_version = str(row[0]).split()[0] if row else 'unknown'
+        except Exception:
+            pass
+
     @property
     def tags(self):
         return self._tags
@@ -238,17 +251,6 @@ def tags(self):
     def cloud_metadata(self):
         return {}
 
-    def _maybe_collect_schemas(self):
-        if not self._schema_collector or not self._conn:
-            return
-        if time.time() - self._last_schema_collection_time < self._schema_collection_interval:
-            return
-        try:
-            self._schema_collector.collect_schemas()
-            self._last_schema_collection_time = time.time()
-        except Exception as e:
-            self.log.error('Error collecting HANA schemas: %s', e)
-
     def query_master_database(self):
         # https://help.sap.com/viewer/4fe29514fd584807ac9f2a04f6754767/2.0.02/en-US/20ae63aa7519101496f6b832ec86afbd.html
         # Only 1 database

sap_hana/datadog_checks/sap_hana/schemas.py

@@ -40,14 +40,20 @@
 
 import contextlib
 from contextlib import closing
-from typing import TYPE_CHECKING
+from typing import TYPE_CHECKING, Any
 
 from datadog_checks.base.utils.db.schemas import DatabaseInfo, SchemaCollector, SchemaCollectorConfig
+from datadog_checks.base.utils.db.utils import DBMAsyncJob
 
 if TYPE_CHECKING:
     from .config_models.instance import CollectSchemas
     from .sap_hana import SapHanaCheck
 
+try:
+    from hdbcli.dbapi import Error as HanaError
+except ImportError:
+    HanaError = Exception  # type: ignore[misc,assignment]
+
 # Flush a schema payload once this many columns have accumulated. The base SchemaCollector
 # chunks payloads by table count (`payload_chunk_size`, default 10,000 tables), but a HANA
 # tenant is a single "database", so that threshold rarely trips: every table for the tenant
@@ -153,7 +159,10 @@
 class HanaSchemaCollectorConfig(SchemaCollectorConfig):
     def __init__(self, config: CollectSchemas):
         super().__init__()
-        self.collection_interval = int(config.collection_interval or 600)
+        # Guard against a non-positive interval (a sub-1 value truncates to 0), which would
+        # otherwise ship as collection_interval: 0 in every schema payload.
+        collection_interval = int(config.collection_interval or 600)
+        self.collection_interval = collection_interval if collection_interval > 0 else 600
         self.max_tables = int(config.max_tables or 2000)
         self.max_views = int(config.max_views or 2000)
         self.max_columns = int(config.max_columns or 500)
@@ -233,6 +242,16 @@ def __init__(self, check: SapHanaCheck, config: CollectSchemas):
         super().__init__(check, HanaSchemaCollectorConfig(config))
         self._query_builder = HanaSchemaQueryBuilder(self._config, self._log)
         self._pending_row = None
+        # The owning HanaSchemaCollectionJob injects its dedicated connection via
+        # set_connection() before each cycle; connection_lost is the collector's signal
+        # back to the job that the connection died mid-cycle and must be rebuilt.
+        self._conn: Any = None
+        self.connection_lost = False
+
+    def set_connection(self, conn: Any) -> None:
+        """Inject the job's dedicated connection for the upcoming collection cycle."""
+        self._conn = conn
+        self.connection_lost = False
 
     def _reset(self) -> None:
         super()._reset()
@@ -254,7 +273,7 @@ def kind(self) -> str:
 
     def _get_databases(self) -> list[DatabaseInfo]:
         try:
-            with closing(self._check._conn.cursor()) as cursor:
+            with closing(self._conn.cursor()) as cursor:
                 cursor.execute(CURRENT_DATABASE_QUERY)
                 row = cursor.fetchone()
                 if not row:
@@ -271,21 +290,31 @@ def _get_databases(self) -> list[DatabaseInfo]:
                     pass
                 return [{'name': db_name, 'description': description}]
         except Exception as e:
+            # A dead HANA connection surfaces here rather than propagating, so flag it for
+            # the owning job to reconnect on the next cycle.
+            if isinstance(e, HanaError):
+                self.connection_lost = True
             self._log.warning("Could not determine current HANA database; skipping schema collection: %s", e)
             return []
 
     @contextlib.contextmanager
     def _get_cursor(self, _database_name):
-        conn = self._check._conn
-        self._query_builder.ensure_stats_permission(conn)
-        query, params = self._query_builder.build()
-        with closing(conn.cursor()) as cursor:
-            cursor.execute(query, params)
-            self._pending_row = cursor.fetchone()
-            try:
-                yield cursor
-            finally:
-                self._pending_row = None
+        conn = self._conn
+        try:
+            self._query_builder.ensure_stats_permission(conn)
+            query, params = self._query_builder.build()
+            with closing(conn.cursor()) as cursor:
+                cursor.execute(query, params)
+                self._pending_row = cursor.fetchone()
+                try:
+                    yield cursor
+                finally:
+                    self._pending_row = None
+        except HanaError:
+            # collect_schemas() swallows this per-database error, so flag the dead
+            # connection to the owning job; it reconnects next cycle.
+            self.connection_lost = True
+            raise
 
     def _get_next(self, cursor):
         """Assemble one table/view from consecutive cursor rows sharing the same (schema, table) key."""
@@ -343,3 +372,68 @@ def _map_row(self, database: DatabaseInfo, table_row) -> dict:
                 }
             ],
         }
+
+
+class HanaSchemaCollectionJob(DBMAsyncJob):
+    """Background job that runs HanaSchemaCollector on its own dedicated connection."""
+
+    def __init__(self, check: SapHanaCheck, config: CollectSchemas) -> None:
+        self._check = check
+        self._schema_collector = HanaSchemaCollector(check, config)
+        self._job_conn: Any = None
+        # Cast to float (not int) so a sub-1 interval doesn't truncate to 0 and make the
+        # rate_limit below divide by zero, which would crash the whole check on construction.
+        # A non-positive value is meaningless as an interval, so clamp it back to the default.
+        collection_interval = float(config.collection_interval or 600)
+        if collection_interval <= 0:
+            collection_interval = 600
+        super().__init__(
+            check,
+            config_host=check._server,
+            rate_limit=1.0 / collection_interval,
+            run_sync=config.run_sync or False,
+            enabled=config.enabled or False,
+            dbms="saphana",
+            min_collection_interval=check.instance.get('min_collection_interval', 15),
+            expected_db_exceptions=(HanaError,),
+            shutdown_callback=self._shutdown,
+            job_name="schema-collection",
+        )
+
+    def _shutdown(self) -> None:
+        if self._job_conn is not None:
+            try:
+                self._job_conn.close()
+            except Exception:
+                pass
+            self._job_conn = None
+
+    def _get_conn(self) -> Any:
+        if self._job_conn is not None:
+            return self._job_conn
+        from hdbcli.dbapi import connect as hana_connect  # noqa: PLC0415
+
+        props = self._check._get_connection_properties()
+        self._job_conn = hana_connect(**props)
+        return self._job_conn
+
+    def run_job(self) -> None:
+        try:
+            conn = self._get_conn()
+            # Resolve the HANA version on this dedicated connection so base_event never
+            # reads it off the main check's connection from this thread.
+            self._check._resolve_dbms_version(conn)
+            self._schema_collector.set_connection(conn)
+            self._schema_collector.collect_schemas()
+        except HanaError:
+            self._reset_conn()
+            raise
+        # collect_schemas() swallows per-database HANA errors internally, so a transient
+        # disconnect never reaches the except above. The collector flags connection_lost
+        # in that case; close ours too so the next cycle reconnects instead of reusing a
+        # dead handle forever.
+        if self._schema_collector.connection_lost:
+            self._reset_conn()
+
+    def _reset_conn(self) -> None:
+        self._shutdown()