[DBMON-6892] Fan out ClickHouse cluster queries over the resolved cluster name

clickhouse/changelog.d/24920.fixed

@@ -0,0 +1 @@
+Fan out over the cluster the instance actually belongs to in single endpoint mode, instead of assuming it is named ``default``, and read local system tables when no cluster can be resolved.

clickhouse/datadog_checks/clickhouse/clickhouse.py

@@ -32,6 +32,7 @@
     SHARED_MERGE_TREE_QUERY,
     ErrorSanitizer,
     HostingType,
+    cluster_all_replicas,
     cluster_aware_query,
 )
 
@@ -305,11 +306,11 @@ def check(self, _):
 
     def get_queries(self) -> list[dict]:
         query_list = []
-        single = self._config.single_endpoint_mode
+        cluster = self.fanout_cluster_name if self._config.single_endpoint_mode else None
 
         def pick(query: dict) -> dict:
             """In single endpoint mode, read all replicas and tag each row per node."""
-            return cluster_aware_query(query) if single else query
+            return cluster_aware_query(query, cluster) if cluster else query
 
         if self._config.use_legacy_queries:
             query_list.extend(
@@ -407,6 +408,18 @@ def _resolve_cluster_name(self) -> str | None:
         self.log.debug('No ClickHouse cluster name found; %s tag will not be emitted', CLUSTER_TAG)
         return None
 
+    @property
+    def fanout_cluster_name(self) -> str | None:
+        """The cluster to fan out over with clusterAllReplicas, or None when there is none to use.
+
+        'default' is only guessed when the deployment is not known to be self-hosted, since Cloud
+        always uses that name. A self-hosted cluster is named arbitrarily, so None is returned
+        instead and callers fall back to the local system table.
+        """
+        if self.cluster_name:
+            return self.cluster_name
+        return None if self.hosting_type == HostingType.SELF_HOSTED else 'default'
+
     @property
     def hosting_type(self) -> str:
         """Whether this instance is ClickHouse Cloud or self-hosted, cached after the first check run."""
@@ -486,9 +499,12 @@ def get_system_table(self, table_name):
         """
         Get the appropriate system table reference based on deployment type.
 
-        For single endpoint mode: Returns clusterAllReplicas('default', system.<table>)
+        For single endpoint mode: Returns clusterAllReplicas(<cluster>, system.<table>)
         For direct connection: Returns system.<table>
 
+        A single endpoint mode instance whose cluster cannot be determined also reads the local
+        table, since there is no cluster name to fan out over.
+
         Args:
             table_name: The system table name (e.g., 'query_log', 'processes')
 
@@ -502,12 +518,10 @@ def get_system_table(self, table_name):
             "system.query_log"  # Direct connection
         """
         if self._config.single_endpoint_mode:
-            # Single endpoint mode: Use clusterAllReplicas to query all nodes
-            # The cluster name is 'default' for ClickHouse Cloud and most setups
-            return f"clusterAllReplicas('default', system.{table_name})"
-        else:
-            # Direct connection: Query the local system table directly
-            return f"system.{table_name}"
+            cluster = self.fanout_cluster_name
+            if cluster:
+                return cluster_all_replicas(cluster, table_name)
+        return f"system.{table_name}"
 
     def ping_clickhouse(self):
         return self._client.ping()

clickhouse/datadog_checks/clickhouse/utils.py

@@ -74,6 +74,21 @@ def compact_query(query):
 )
 
 
+def quote_string(value: str) -> str:
+    """Render a SQL string literal, escaping a cluster name that arrives as server-supplied data."""
+    escaped = value.replace('\\', '\\\\').replace("'", "\\'")
+    return f"'{escaped}'"
+
+
+def cluster_all_replicas(cluster: str, table: str) -> str:
+    """Reference a system table on every replica of a cluster.
+
+    Only Cloud names its cluster 'default'; a literal 'default' either raises UNKNOWN_CLUSTER or,
+    against the stock localhost-only 'default' cluster, silently returns the local node alone.
+    """
+    return f"clusterAllReplicas({quote_string(cluster)}, system.{table})"
+
+
 HOSTING_TYPE_TAG = 'hosting_type'
 
 
@@ -90,8 +105,8 @@ class HostingType:
 SHARED_MERGE_TREE_QUERY = "SELECT count() FROM system.table_engines WHERE name = 'SharedMergeTree'"
 
 
-def cluster_aware_query(base: dict) -> dict:
-    """Build a cluster-aware variant that reads all replicas and tags each row per node.
+def cluster_aware_query(base: dict, cluster: str) -> dict:
+    """Build a cluster-aware variant that reads all replicas of a cluster and tags each row per node.
 
     Derives the SELECT list and table from the base query, whose shape is always
     ``SELECT <cols> FROM system.<table>[ <trailing clause>]``.
@@ -101,8 +116,7 @@ def cluster_aware_query(base: dict) -> dict:
     return {
         'name': base['name'],
         'query': (
-            f"{select}, hostName() AS {CLUSTER_NODE_TAG} "
-            f"FROM clusterAllReplicas('default', system.{table}){sep}{trailing}"
+            f"{select}, hostName() AS {CLUSTER_NODE_TAG} FROM {cluster_all_replicas(cluster, table)}{sep}{trailing}"
         ),
         'columns': [*base['columns'], {'name': CLUSTER_NODE_TAG, 'type': 'tag'}],
     }

clickhouse/tests/test_table_metrics.py

@@ -172,7 +172,9 @@ def test_routes_through_cluster_all_replicas_in_single_endpoint_mode(schema_metr
     with (
         mock.patch.object(check.table_metrics, '_execute_query', side_effect=lambda q: dbm_queries.append(q) or []),
         mock.patch.object(check, 'execute_query_raw', side_effect=lambda q: raw_queries.append(q) or []),
+        mock.patch.object(ClickhouseCheck, 'fanout_cluster_name', new_callable=mock.PropertyMock) as fanout,
     ):
+        fanout.return_value = 'default'
         check.table_metrics.run_job()
 
     assert any("clusterAllReplicas('default', system.tables)" in q for q in dbm_queries)

clickhouse/tests/test_unit.py

@@ -411,7 +411,7 @@ def test_database_hostname_ignores_reported_hostname_override(reported_hostname,
 
 def test_cluster_aware_query_bulk_match_query():
     """The cluster-aware variant reads all replicas and tags system.events per node."""
-    variant = cluster_aware_query(advanced_queries.SystemEvents)
+    variant = cluster_aware_query(advanced_queries.SystemEvents, 'default')
 
     assert variant['query'] == (
         "SELECT value, event, hostName() AS clickhouse_node FROM clusterAllReplicas('default', system.events)"
@@ -424,7 +424,7 @@ def test_cluster_aware_query_bulk_match_query():
 
 def test_cluster_aware_query_preserves_where_clause():
     """system.errors carries a WHERE clause that must survive in the cluster-aware variant."""
-    variant = cluster_aware_query(advanced_queries.SystemErrors)
+    variant = cluster_aware_query(advanced_queries.SystemErrors, 'default')
 
     assert variant['query'] == (
         "SELECT value, name, code, remote, hostName() AS clickhouse_node "
@@ -435,7 +435,7 @@ def test_cluster_aware_query_preserves_where_clause():
 
 def test_cluster_aware_query_legacy_query():
     """The helper builds a cluster-aware variant for a legacy query too."""
-    variant = cluster_aware_query(queries.SystemMetrics)
+    variant = cluster_aware_query(queries.SystemMetrics, 'default')
 
     assert variant['query'] == (
         "SELECT value, metric, hostName() AS clickhouse_node FROM clusterAllReplicas('default', system.metrics)"
@@ -444,6 +444,19 @@ def test_cluster_aware_query_legacy_query():
     assert queries.SystemMetrics['query'] == 'SELECT value, metric FROM system.metrics'
 
 
+def test_cluster_aware_query_fans_out_over_the_named_cluster():
+    """The resolved name has to reach the SQL, since a self-hosted cluster is not 'default'."""
+    variant = cluster_aware_query(queries.SystemMetrics, 'prod_cluster')
+
+    assert "clusterAllReplicas('prod_cluster', system.metrics)" in variant['query']
+
+
+def test_cluster_aware_query_escapes_the_cluster_name():
+    variant = cluster_aware_query(queries.SystemMetrics, "o'brien")
+
+    assert "clusterAllReplicas('o\\'brien', system.metrics)" in variant['query']
+
+
 @pytest.mark.parametrize('use_advanced_queries', [True, False])
 def test_get_queries_tags_system_tables_per_node_in_single_endpoint_mode(instance, use_advanced_queries):
     instance = {
@@ -482,6 +495,49 @@ def test_get_queries_uses_base_queries_for_direct_connection(instance, use_advan
     assert all('clusterAllReplicas' not in q['query'] for q in check.get_queries())
 
 
+def test_get_queries_fans_out_over_the_resolved_cluster(instance):
+    instance = {**instance, 'single_endpoint_mode': True}
+    check = ClickhouseCheck('clickhouse', {}, [instance])
+    check._server_version = '24.8'
+    with mock.patch.object(ClickhouseCheck, 'fanout_cluster_name', new_callable=mock.PropertyMock) as fanout:
+        fanout.return_value = 'prod_cluster'
+        cluster_aware = [q for q in check.get_queries() if 'clusterAllReplicas' in q['query']]
+
+    assert cluster_aware
+    assert all("clusterAllReplicas('prod_cluster'," in q['query'] for q in cluster_aware)
+
+
+def test_get_queries_reads_local_tables_when_there_is_no_cluster_to_fan_out_over(instance):
+    """Without a cluster there is nothing to fan out over, so the local table is read."""
+    instance = {**instance, 'single_endpoint_mode': True}
+    check = ClickhouseCheck('clickhouse', {}, [instance])
+    check._server_version = '24.8'
+    with mock.patch.object(ClickhouseCheck, 'fanout_cluster_name', new_callable=mock.PropertyMock) as fanout:
+        fanout.return_value = None
+        query_list = check.get_queries()
+
+    assert query_list
+    assert all('clusterAllReplicas' not in q['query'] for q in query_list)
+
+
+@pytest.mark.parametrize(
+    ('single_endpoint_mode', 'fanout_cluster', 'expected'),
+    [
+        pytest.param(True, 'default', "clusterAllReplicas('default', system.query_log)", id='cloud'),
+        pytest.param(True, 'prod_cluster', "clusterAllReplicas('prod_cluster', system.query_log)", id='self-hosted'),
+        pytest.param(True, None, 'system.query_log', id='no-cluster'),
+        pytest.param(False, 'default', 'system.query_log', id='direct-connection'),
+    ],
+)
+def test_get_system_table(instance, single_endpoint_mode, fanout_cluster, expected):
+    instance = {**instance, 'single_endpoint_mode': single_endpoint_mode}
+    check = ClickhouseCheck('clickhouse', {}, [instance])
+    with mock.patch.object(ClickhouseCheck, 'fanout_cluster_name', new_callable=mock.PropertyMock) as fanout:
+        fanout.return_value = fanout_cluster
+
+        assert check.get_system_table('query_log') == expected
+
+
 def make_query_replaying_check(query_results):
     """Build a check whose execute_query_raw replays query_results keyed by SQL.
 
@@ -565,6 +621,26 @@ def test_cluster_name_is_cached_including_the_absent_case():
     assert check.execute_query_raw.call_count == 2  # one attempt per source, not per access
 
 
+@pytest.mark.parametrize(
+    ('cluster_name', 'hosting_type', 'expected'),
+    [
+        pytest.param('prod_cluster', HostingType.SELF_HOSTED, 'prod_cluster', id='self-hosted-named-cluster'),
+        pytest.param('default', HostingType.CLOUD, 'default', id='cloud'),
+        pytest.param(None, HostingType.CLOUD, 'default', id='cloud-unresolved-name'),
+        pytest.param(None, HostingType.UNKNOWN, 'default', id='unknown-hosting-unresolved-name'),
+        pytest.param(None, HostingType.SELF_HOSTED, None, id='self-hosted-without-a-cluster'),
+    ],
+)
+def test_fanout_cluster_name(cluster_name, hosting_type, expected):
+    check = make_query_replaying_check({})
+    with mock.patch.object(ClickhouseCheck, 'cluster_name', new_callable=mock.PropertyMock) as cluster_name_prop:
+        with mock.patch.object(ClickhouseCheck, 'hosting_type', new_callable=mock.PropertyMock) as hosting_type_prop:
+            cluster_name_prop.return_value = cluster_name
+            hosting_type_prop.return_value = hosting_type
+
+            assert check.fanout_cluster_name == expected
+
+
 def test_check_tags_with_cluster(instance):
     check = ClickhouseCheck('clickhouse', {}, [instance])
     with mock.patch.object(ClickhouseCheck, 'cluster_name', new_callable=mock.PropertyMock) as cluster_name: