feat: enhance ClickHouse Cloud log collection with internal user filtering and additional metrics

Author: pythonicrahul

.editorconfig

@@ -0,0 +1,18 @@
+root = true
+
+[*]
+indent_style = space
+indent_size = 4
+end_of_line = lf
+charset = utf-8
+trim_trailing_whitespace = true
+insert_final_newline = true
+
+[*.{yml,yaml}]
+indent_size = 2
+
+[*.md]
+trim_trailing_whitespace = false
+
+[Makefile]
+indent_style = tab

.gitignore

@@ -37,4 +37,5 @@ conf.d/openmetrics_clickhouse.yaml
 clickhouse-data/
 
 PRD.md
-*.yaml
+*.yaml
+*.csv

CHANGELOG.md

@@ -4,20 +4,3 @@ All notable changes to this project will be documented in this file.
 
 The format is based on [Keep a Changelog](https://keepachangelog.com/en/1.1.0/),
 and this project adheres to [Semantic Versioning](https://semver.org/spec/v2.0.0.html).
-
-## [1.0.0] - 2026-04-03
-
-### Added
-- Custom Datadog Agent check for ClickHouse Cloud log collection
-- Query log collection from `system.query_log` (completed queries and exceptions)
-- Text log collection from `system.text_log` (Error, Warning, Fatal levels)
-- OpenMetrics configuration for ClickHouse Cloud Prometheus endpoint
-- Cursor-based pagination with duplicate-delivery-over-loss semantics
-- Configurable batch size, slow-query threshold, backfill window, and query timeout
-- Input validation with bounds checking on all numeric config parameters
-- HTTP retry logic (2 retries with backoff on 502/503/504)
-- Configurable `cluster_name` for the Datadog `service` field
-- `ddsource: clickhouse` on all log entries for Datadog pipeline compatibility
-- Comprehensive test suite (53 tests)
-- CI workflow with ruff lint and pytest across Python 3.10/3.11/3.12
-- Dependabot configuration for pip and GitHub Actions

checks/clickhouse_cloud.py

@@ -35,30 +35,57 @@
     read_bytes,
     result_rows,
     written_rows,
+    written_bytes,
     exception,
     exception_code,
     query,
     type,
+    query_kind,
+    current_database,
     arrayStringConcat(tables, ', ') AS tables,
     client_name
 FROM system.query_log
-WHERE type IN ('QueryFinish', 'ExceptionWhileProcessing')
+WHERE type IN ('QueryFinish', 'ExceptionWhileProcessing', 'ExceptionBeforeStart')
+  AND is_initial_query = 1
   AND event_time_microseconds > fromUnixTimestamp64Micro({last_cursor})
+  AND query NOT LIKE '%system.query_log%'
+  AND query NOT LIKE '%system.text_log%'
+  {internal_user_filter}
 ORDER BY event_time_microseconds ASC
 LIMIT {batch_size}
 """
 
+# Filter clause injected into QUERY_LOG_SQL when exclude_internal_users is True.
+# ClickHouse Cloud runs health checks, backups, and observability queries under
+# service accounts whose usernames follow several patterns:
+#   1. "*-internal" suffix (monitoring-internal, operator-internal, backups-internal, etc.)
+#   2. "clickhouse-cloud-*" prefix (clickhouse-cloud-monitor — the primary metrics scraper)
+#   3. "prometheus-exporter" — Prometheus metrics collection
+#   4. Empty string "" — internal system-level metrics queries (SELECT from
+#      system.dimensional_metrics, system.histogram_metrics, etc.) that run
+#      under a blank user and generate heavy query_log volume.
+# Together these generate ~99% of query_log volume on an idle cluster with zero
+# operational value for application teams.
+INTERNAL_USER_FILTER = (
+    "AND user NOT LIKE '%-internal'"
+    " AND user NOT LIKE 'clickhouse-cloud-%'"
+    " AND user != 'prometheus-exporter'"
+    " AND user != ''"
+)
+
 TEXT_LOG_SQL = """
 SELECT
     event_time,
     toUnixTimestamp64Micro(event_time_microseconds) AS cursor_us,
     level,
     logger_name,
     message,
-    thread_id
+    thread_id,
+    query_id
 FROM system.text_log
-WHERE level IN ('Error', 'Warning', 'Fatal')
+WHERE level IN ('Fatal', 'Critical', 'Error', 'Warning')
   AND event_time_microseconds > fromUnixTimestamp64Micro({last_cursor})
+  AND logger_name NOT IN ('QueryProfiler', 'GlobalProfiler')
 ORDER BY event_time_microseconds ASC
 LIMIT {batch_size}
 """
@@ -76,6 +103,7 @@
 
 TYPE_QUERY_FINISH = "QueryFinish"
 TYPE_QUERY_EXCEPTION = "ExceptionWhileProcessing"
+TYPE_QUERY_EXCEPTION_BEFORE_START = "ExceptionBeforeStart"
 
 # ---------------------------------------------------------------------------
 # Datadog metric / service-check names
@@ -92,6 +120,7 @@
 
 TEXT_LOG_LEVEL_MAP: dict[str, str] = {
     "Fatal": "critical",
+    "Critical": "critical",
     "Error": "error",
     "Warning": "warning",
 }
@@ -128,6 +157,7 @@ def __init__(
         # Feature toggles
         self.collect_query_logs: bool = inst.get("collect_query_logs", True)
         self.collect_text_logs: bool = inst.get("collect_text_logs", True)
+        self.exclude_internal_users: bool = inst.get("exclude_internal_users", True)
 
         # Tuning — validate all numeric config to prevent SQL injection and
         # catch misconfiguration early (e.g. log_batch_size: "all").
@@ -371,8 +401,10 @@ def _collect_logs(
 
     def _collect_query_logs(self) -> None:
         """Fetch new rows from system.query_log and send as Datadog logs."""
+        user_filter = INTERNAL_USER_FILTER if self.exclude_internal_users else ""
+        sql = QUERY_LOG_SQL.replace("{internal_user_filter}", user_filter)
         self._collect_logs(
-            sql_template=QUERY_LOG_SQL,
+            sql_template=sql,
             cursor_key=CURSOR_QUERY_LOG,
             check_name=SC_QUERY_LOG_CONNECT,
             gauge_name=GAUGE_QUERY_LOG_ROWS,
@@ -384,7 +416,7 @@ def _build_query_log_payload(self, row: dict[str, Any]) -> dict[str, Any]:
         query_type = row.get("type", "")
 
         # Determine log level
-        if query_type == TYPE_QUERY_EXCEPTION:
+        if query_type in (TYPE_QUERY_EXCEPTION, TYPE_QUERY_EXCEPTION_BEFORE_START):
             level = "error"
             type_label = "exception"
         else:
@@ -407,9 +439,12 @@ def _build_query_log_payload(self, row: dict[str, Any]) -> dict[str, Any]:
             "clickhouse.read_bytes": int(row.get("read_bytes", 0)),
             "clickhouse.result_rows": int(row.get("result_rows", 0)),
             "clickhouse.written_rows": int(row.get("written_rows", 0)),
+            "clickhouse.written_bytes": int(row.get("written_bytes", 0)),
             "clickhouse.exception": row.get("exception", ""),
             "clickhouse.exception_code": int(row.get("exception_code", 0)),
             "clickhouse.query_type": type_label,
+            "clickhouse.query_kind": row.get("query_kind", ""),
+            "clickhouse.database": row.get("current_database", ""),
             "clickhouse.tables": row.get("tables", ""),
             "clickhouse.client": row.get("client_name", ""),
         }
@@ -441,6 +476,7 @@ def _build_text_log_payload(self, row: dict[str, Any]) -> dict[str, Any]:
             "status": level,
             "clickhouse.logger": row.get("logger_name", ""),
             "clickhouse.thread_id": str(row.get("thread_id", "")),
+            "clickhouse.query_id": row.get("query_id", ""),
         }
 
     # ------------------------------------------------------------------

conf.d/clickhouse_cloud.d/conf.yaml.example

@@ -40,6 +40,16 @@ instances:
     #
     # collect_text_logs: true
 
+    ## @param exclude_internal_users - boolean - optional - default: true
+    ## Exclude queries from ClickHouse Cloud internal service accounts.
+    ## ClickHouse Cloud runs health checks, backups, and observability queries
+    ## under system accounts (e.g. clickhouse-cloud-monitor, *-internal,
+    ## prometheus-exporter) that generate ~99% of query_log volume on idle
+    ## clusters with no operational value for application teams.
+    ## Set to false if you need visibility into internal platform queries.
+    #
+    # exclude_internal_users: true
+
     ## @param log_batch_size - integer - optional - default: 1000
     ## Maximum number of log rows fetched per check run, per table.
     ## Range: 1–10000. Higher values reduce API calls but increase memory usage.

tests/fixtures/query_log_rows.json

@@ -10,10 +10,13 @@
     "read_bytes": 1048576,
     "result_rows": 50,
     "written_rows": 0,
+    "written_bytes": 0,
     "exception": "",
     "exception_code": 0,
     "query": "SELECT count() FROM events WHERE date > '2026-03-01'",
     "type": "QueryFinish",
+    "query_kind": "Select",
+    "current_database": "default",
     "tables": "default.events",
     "client_name": "python-driver"
   },
@@ -28,10 +31,13 @@
     "read_bytes": 524288000,
     "result_rows": 1,
     "written_rows": 0,
+    "written_bytes": 0,
     "exception": "",
     "exception_code": 0,
     "query": "SELECT uniq(user_id) FROM events",
     "type": "QueryFinish",
+    "query_kind": "Select",
+    "current_database": "default",
     "tables": "default.events",
     "client_name": "clickhouse-client"
   },
@@ -46,10 +52,13 @@
     "read_bytes": 0,
     "result_rows": 0,
     "written_rows": 0,
+    "written_bytes": 0,
     "exception": "Code: 60. DB::Exception: Table default.nonexistent doesn't exist.",
     "exception_code": 60,
     "query": "SELECT * FROM nonexistent",
     "type": "ExceptionWhileProcessing",
+    "query_kind": "Select",
+    "current_database": "default",
     "tables": "",
     "client_name": "python-driver"
   }

tests/fixtures/text_log_rows.json

@@ -5,22 +5,25 @@
     "level": "Error",
     "logger_name": "MergeTreeBackgroundExecutor",
     "message": "Exception while processing merge: Code: 241. DB::Exception: Memory limit exceeded",
-    "thread_id": 12345
+    "thread_id": 12345,
+    "query_id": "abc-123-def-456"
   },
   {
     "event_time": "2026-03-29 10:01:05",
     "cursor_us": 1743246065000000,
     "level": "Warning",
     "logger_name": "ReplicatedMergeTreeQueue",
     "message": "Part all_0_0_0 is not available on any replica",
-    "thread_id": 12346
+    "thread_id": 12346,
+    "query_id": ""
   },
   {
     "event_time": "2026-03-29 10:01:10",
     "cursor_us": 1743246070000000,
     "level": "Fatal",
     "logger_name": "Application",
     "message": "Caught signal 11 (SIGSEGV). Stack trace: ...",
-    "thread_id": 1
+    "thread_id": 1,
+    "query_id": ""
   }
 ]