[clickhouse] Skip emitting empty storage_health payloads (#23553)

* [clickhouse][dbm] Skip emitting empty storage_health payloads

Skip the database_monitoring_query_activity emission when every parts-and-merges
collection (parts, merges, mutations, replication queue, detached parts, thresholds)
is empty. This is the case for fresh ClickHouse Cloud instances with no user data
and restricted system tables, and for transient total-collection-failure scenarios.

Empty storage_health payloads previously wedged dbm-events-processor partitions on
the dd-go side; that processor is being fixed independently in
DataDog/dd-go#234748. This change also avoids emitting one useless Kafka message
per collection cycle for any genuinely-idle/empty instance.

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

* Add changelog entry for PR #23553

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

* Add end-to-end test for skipped emission when all collectors return empty

Pins the wiring between _collect_and_emit and _emit_events for the case where
every collection query fails or returns no rows, which is the scenario that
fires when all queries hit exceptions (auth/network) or on a fresh, restricted
ClickHouse Cloud instance.

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

---------

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

Author: sangeetashivaji

clickhouse/changelog.d/23553.fixed

@@ -0,0 +1 @@
+Skip emitting empty storage_health payloads when every parts-and-merges collection is empty.

clickhouse/datadog_checks/clickhouse/parts_and_merges.py

@@ -818,6 +818,8 @@ def _emit_events(
         thresholds: list[dict] | None = None,
     ) -> None:
         """Emit a per-cycle row-level payload consumed by dbm-events-processor."""
+        if not (parts or merges or mutations or replication_queue or detached_parts or thresholds):
+            return
         now_ms = int(time.time() * 1000)
         payload = {
             "host": self._check.reported_hostname,

clickhouse/tests/test_parts_and_merges.py

@@ -856,12 +856,27 @@ def test_emit_events_uses_query_activity_channel_not_metadata(check):
         mock.patch('datadog_checks.clickhouse.parts_and_merges.datadog_agent') as agent_mock,
     ):
         agent_mock.get_version.return_value = '7.64.0'
-        job._emit_events([], [], [], [], [])
+        job._emit_events(_collected_parts(), [], [], [], [])
 
     activity_mock.assert_called_once()
     metadata_mock.assert_not_called()
 
 
+def test_emit_events_skips_when_all_collections_empty(check):
+    job = check.parts_and_merges
+    job.tags = ['test:clickhouse']
+    job._tags_no_db = ['test:clickhouse']
+
+    with (
+        mock.patch.object(check, 'database_monitoring_query_activity') as activity_mock,
+        mock.patch('datadog_checks.clickhouse.parts_and_merges.datadog_agent') as agent_mock,
+    ):
+        agent_mock.get_version.return_value = '7.64.0'
+        job._emit_events([], [], [], [], [], [])
+
+    activity_mock.assert_not_called()
+
+
 # -----------------------------------------------------------------------------
 # Error handling
 # -----------------------------------------------------------------------------
@@ -907,6 +922,29 @@ def test_collect_and_emit_runs_with_partial_failures(check):
     assert payload['clickhouse']['active_merges'] == _collected_merges()
 
 
+def test_collect_and_emit_skips_when_all_collectors_empty(check):
+    job = check.parts_and_merges
+    job.tags = ['test:clickhouse']
+    job._tags_no_db = ['test:clickhouse']
+
+    with (
+        mock.patch.object(job, '_collect_parts', return_value=[]),
+        mock.patch.object(job, '_collect_merges', return_value=[]),
+        mock.patch.object(job, '_collect_mutations', return_value=[]),
+        mock.patch.object(job, '_collect_mutations_aggregated', return_value=[]),
+        mock.patch.object(job, '_collect_replication_queue', return_value=[]),
+        mock.patch.object(job, '_collect_replication_queue_aggregated', return_value=[]),
+        mock.patch.object(job, '_collect_detached_parts', return_value=[]),
+        mock.patch.object(job, '_collect_thresholds', return_value=[]),
+        mock.patch.object(check, 'database_monitoring_query_activity') as activity_mock,
+        mock.patch('datadog_checks.clickhouse.parts_and_merges.datadog_agent') as agent_mock,
+    ):
+        agent_mock.get_version.return_value = '7.64.0'
+        job._collect_and_emit()
+
+    activity_mock.assert_not_called()
+
+
 # -----------------------------------------------------------------------------
 # Cluster routing
 # -----------------------------------------------------------------------------