diff --git a/clickhouse/changelog.d/24920.fixed b/clickhouse/changelog.d/24920.fixed new file mode 100644 index 0000000000000..9e7d20e19b2a4 --- /dev/null +++ b/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. diff --git a/clickhouse/datadog_checks/clickhouse/clickhouse.py b/clickhouse/datadog_checks/clickhouse/clickhouse.py index f8d603a9a05a8..92a9ece0dbe14 100644 --- a/clickhouse/datadog_checks/clickhouse/clickhouse.py +++ b/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.) + For single endpoint mode: Returns clusterAllReplicas(, system.
) For direct connection: Returns system.
+ 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() diff --git a/clickhouse/datadog_checks/clickhouse/utils.py b/clickhouse/datadog_checks/clickhouse/utils.py index a41ae4515d870..823707ac4d7f0 100644 --- a/clickhouse/datadog_checks/clickhouse/utils.py +++ b/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 FROM system.
[ ]``. @@ -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'}], } diff --git a/clickhouse/tests/docker/volumes/clickhouse.xml b/clickhouse/tests/docker/volumes/clickhouse.xml index 47c7dd0dac352..5b2d936340001 100644 --- a/clickhouse/tests/docker/volumes/clickhouse.xml +++ b/clickhouse/tests/docker/volumes/clickhouse.xml @@ -29,5 +29,19 @@ clickhouse + dd_test_cluster + + + + + + clickhouse + 9000 + + + + diff --git a/clickhouse/tests/test_cluster_fanout_integration.py b/clickhouse/tests/test_cluster_fanout_integration.py new file mode 100644 index 0000000000000..f16be3db3f54e --- /dev/null +++ b/clickhouse/tests/test_cluster_fanout_integration.py @@ -0,0 +1,63 @@ +# (C) Datadog, Inc. 2026-present +# All rights reserved +# Licensed under a 3-clause BSD style license (see LICENSE) +import pytest + +from datadog_checks.clickhouse import ClickhouseCheck +from datadog_checks.clickhouse.utils import CLUSTER_NODE_TAG + +from .common import CLICKHOUSE_VERSION, is_legacy + +# Defined alongside the `cluster` macro and `remote_servers` block in +# tests/docker/volumes/clickhouse.xml: a self-hosted style cluster deliberately not named +# 'default', so these tests fail if the fan-out logic ever goes back to assuming that name +# (see PR #24920). +TEST_CLUSTER_NAME = 'dd_test_cluster' + +pytestmark = [ + pytest.mark.integration, + pytest.mark.usefixtures('dd_environment'), + pytest.mark.skipif( + is_legacy(CLICKHOUSE_VERSION), + reason='dd_test_cluster is only defined in the non-legacy docker compose config', + ), +] + + +def test_fanout_cluster_name_resolves_the_real_cluster(instance, dd_run_check): + """A self-hosted deployment with a real cluster must fan out over it, not a hardcoded 'default'.""" + instance = {**instance, 'single_endpoint_mode': True} + check = ClickhouseCheck('clickhouse', {}, [instance]) + dd_run_check(check) + + assert check.cluster_name == TEST_CLUSTER_NAME + assert check.fanout_cluster_name == TEST_CLUSTER_NAME + + +def test_get_system_table_fans_out_over_the_resolved_cluster(instance, dd_run_check): + instance = {**instance, 'single_endpoint_mode': True} + check = ClickhouseCheck('clickhouse', {}, [instance]) + dd_run_check(check) + + table_ref = check.get_system_table('one') + assert table_ref == f"clusterAllReplicas('{TEST_CLUSTER_NAME}', system.one)" + + # Executed for real: a hardcoded 'default' would either raise UNKNOWN_CLUSTER (no such + # cluster on a self-hosted deployment) or silently read the stock local-only cluster instead. + rows = check.execute_query_raw(f'SELECT count() FROM {table_ref}') + assert rows[0][0] == 1 + + +def test_single_endpoint_mode_metrics_carry_the_cluster_node_tag(aggregator, instance, dd_run_check): + """The QueryManager queries built by get_queries() must actually execute against the resolved cluster.""" + instance = {**instance, 'single_endpoint_mode': True} + check = ClickhouseCheck('clickhouse', {}, [instance]) + dd_run_check(check) + + node_tagged_metrics = [ + name + for name in aggregator.metric_names + for stub in aggregator.metrics(name) + if any(tag.startswith(f'{CLUSTER_NODE_TAG}:') for tag in stub.tags) + ] + assert node_tagged_metrics, 'Expected at least one metric fanned out via clusterAllReplicas to carry a node tag' diff --git a/clickhouse/tests/test_table_metrics.py b/clickhouse/tests/test_table_metrics.py index 99db116dcef29..18999abeb184c 100644 --- a/clickhouse/tests/test_table_metrics.py +++ b/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) diff --git a/clickhouse/tests/test_unit.py b/clickhouse/tests/test_unit.py index 8037f2c816f4c..7a18eb0df91cf 100644 --- a/clickhouse/tests/test_unit.py +++ b/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: