[DBMON-6432] ClickHouse Parts & Merges support (DataDog#23361)

* Adding parts and merges support

* Cleanup comments

* Address codex comments

* Fixing CI

* Make stalled-merge and stuck-replication thresholds configurable

Expose STALLED_MERGE_ELAPSED_THRESHOLD_SECONDS and
STUCK_REPLICATION_NUM_TRIES as parts_and_merges config options so
operators can tune them to match their cluster behaviour without
rebuilding the integration.

  parts_and_merges:
    stalled_merge_elapsed_threshold_seconds: 3600  # default: 1 h
    stuck_replication_num_tries: 3                 # default: 3

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

* Address comment around payload schema

* Adding two new thresholds for merge times

---------

Co-authored-by: Claude Sonnet 4.6 <noreply@anthropic.com>

Author: sangeetashivaji

clickhouse/assets/configuration/spec.yaml

@@ -296,6 +296,99 @@ files:
           value:
             type: boolean
             example: false
+    - name: parts_and_merges
+      description: Configure parts and merges monitoring
+      options:
+        - name: enabled
+          description: |
+            Enable collection of parts and merges monitoring. Requires `dbm: true`.
+            Emits per-table gauge metrics (clickhouse.table.parts.*, clickhouse.merges.*, clickhouse.mutations.*,
+            clickhouse.replication.*) for trend dashboards and alerting, plus a row-level event payload per
+            collection cycle consumed by the DBM Storage Health timeline view.
+          value:
+            type: boolean
+            example: true
+        - name: collection_interval
+          description: |
+            Set the parts and merges collection interval (in seconds). All four system tables
+            (parts, merges, mutations, replication_queue) are queried on each run.
+          value:
+            type: number
+            example: 60
+        - name: max_parts_rows
+          description: |
+            Maximum number of rows to include in the per-cycle event payload from system.parts,
+            ordered by active_part_count descending. With `table_metrics_include_partition_tag`
+            disabled (default) rows are aggregated per-table server-side, so this caps tables.
+            When partition tagging is enabled, rows are per-partition and this caps partitions.
+          value:
+            type: integer
+            example: 500
+        - name: max_mutations_rows
+          description: |
+            Maximum number of rows to include in the per-cycle event payload from system.mutations,
+            ordered by create_time ascending. Caps payload size when many mutations are pending.
+          value:
+            type: integer
+            example: 200
+        - name: max_detached_parts_rows
+          description: |
+            Maximum number of rows returned from system.detached_parts per cycle.
+            Applied as a raw SQL LIMIT; increase only if your cluster has a pathological number
+            of (database, table, reason) combinations in detached storage.
+          value:
+            type: integer
+            example: 1000
+        - name: max_replication_queue_rows
+          description: |
+            Maximum number of rows returned from system.replication_queue per cycle,
+            ordered by position ascending. Busy replicated clusters can exceed the default
+            during incidents; raise this if replication queue backlogs are being truncated.
+          value:
+            type: integer
+            example: 1000
+        - name: table_metrics_include_partition_tag
+          description: |
+            Add a partition tag to parts gauges (clickhouse.table.parts.*).
+            Disabled by default to avoid high cardinality: a table partitioned by day over two years
+            yields ~730 partitions and hundreds of additional metric series per table.
+            Enable only when per-partition alerting is explicitly needed.
+          value:
+            type: boolean
+            example: false
+        - name: table_metrics_max_tables
+          description: |
+            Cap per-table gauge emission to the top N tables by active part count.
+            Guards against pathological schemas with very large numbers of tables or partitions.
+          value:
+            type: integer
+            example: 200
+        - name: stalled_merge_elapsed_threshold_seconds
+          description: |
+            Number of seconds a merge must be running before it is counted as stalled
+            in the `clickhouse.merges.stalled` gauge. Raise this threshold if long-running
+            merges are expected in your environment (e.g. large Wide-part merges on spinning
+            disks) to avoid false-positive stall alerts.
+          value:
+            type: integer
+            example: 3600
+        - name: stuck_replication_num_tries
+          description: |
+            Minimum number of failed attempts (num_tries) before a replication queue entry
+            is counted as stuck in the `clickhouse.replication.stuck` gauge. Entries below
+            this threshold are still counted in `clickhouse.replication.queue.depth`.
+            Increase this value if transient failures routinely exceed the default before
+            self-healing.
+          value:
+            type: integer
+            example: 3
+        - name: run_sync
+          hidden: true
+          description: |
+            Run the parts and merges collection synchronously. For testing only.
+          value:
+            type: boolean
+            example: false
     - template: instances/db
       overrides:
         custom_queries.value.example:

clickhouse/changelog.d/23361.added

@@ -0,0 +1 @@
+Add parts and merges monitoring (DBM): per-table gauges for parts, merges, mutations, and replication queue health, plus a per-cycle row-level event payload for the DBM Storage Health timeline view.

clickhouse/datadog_checks/clickhouse/clickhouse.py

@@ -17,6 +17,7 @@
 from .__about__ import __version__
 from .config import build_config, sanitize
 from .health import ClickhouseHealth, HealthEvent, HealthStatus
+from .parts_and_merges import ClickhousePartsAndMerges
 from .query_completions import ClickhouseQueryCompletions
 from .query_errors import ClickhouseQueryErrors
 from .statement_samples import ClickhouseStatementSamples
@@ -127,6 +128,12 @@ def _init_dbm_components(self):
         else:
             self.query_errors = None
 
+        # Initialize parts and merges monitoring (from system.parts, merges, mutations, replication_queue)
+        if self._config.dbm and self._config.parts_and_merges.enabled:
+            self.parts_and_merges = ClickhousePartsAndMerges(self, self._config.parts_and_merges)
+        else:
+            self.parts_and_merges = None
+
     @property
     def tags(self) -> list[str]:
         """Return the current list of tags from the TagManager."""
@@ -255,6 +262,10 @@ def check(self, _):
         if self.query_errors:
             self.query_errors.run_job_loop(self.tags)
 
+        # Run parts and merges monitoring if enabled
+        if self.parts_and_merges:
+            self.parts_and_merges.run_job_loop(self.tags)
+
     @AgentCheck.metadata_entrypoint
     def collect_version(self):
         version = list(self.execute_query_raw('SELECT version()'))[0][0]
@@ -474,6 +485,8 @@ def cancel(self):
             self.query_completions.cancel()
         if self.query_errors:
             self.query_errors.cancel()
+        if self.parts_and_merges:
+            self.parts_and_merges.cancel()
 
         # Wait for job loops to finish
         if self.statement_metrics and self.statement_metrics._job_loop_future:
@@ -484,6 +497,8 @@ def cancel(self):
             self.query_completions._job_loop_future.result()
         if self.query_errors and self.query_errors._job_loop_future:
             self.query_errors._job_loop_future.result()
+        if self.parts_and_merges and self.parts_and_merges._job_loop_future:
+            self.parts_and_merges._job_loop_future.result()
 
         # Close main client
         if self._client:

clickhouse/datadog_checks/clickhouse/config.py

@@ -124,6 +124,10 @@ def build_config(check: ClickhouseCheck) -> Tuple[InstanceConfig, ValidationResu
                 **dict_defaults.instance_query_errors().model_dump(),
                 **(instance.get('query_errors', {})),
             },
+            "parts_and_merges": {
+                **dict_defaults.instance_parts_and_merges().model_dump(),
+                **(instance.get('parts_and_merges', {})),
+            },
             # Tags - ensure we have a list, not None
             "tags": list(instance.get('tags', [])),
             # Other settings
@@ -213,6 +217,30 @@ def _apply_validated_defaults(args: dict, instance: dict, validation_result: Val
             f"query_errors.samples_per_hour_per_query must be greater than 0, defaulting to {default_value}."
         )
 
+    if _safefloat(args.get('parts_and_merges', {}).get('collection_interval')) <= 0:
+        default_value = dict_defaults.instance_parts_and_merges().collection_interval
+        args['parts_and_merges']['collection_interval'] = default_value
+        validation_result.add_warning(
+            f"parts_and_merges.collection_interval must be greater than 0, defaulting to {default_value} seconds."
+        )
+
+    _pm_defaults = dict_defaults.instance_parts_and_merges()
+    for _field in (
+        'max_parts_rows',
+        'max_mutations_rows',
+        'max_detached_parts_rows',
+        'max_replication_queue_rows',
+        'table_metrics_max_tables',
+        'stalled_merge_elapsed_threshold_seconds',
+        'stuck_replication_num_tries',
+    ):
+        if _safefloat(args.get('parts_and_merges', {}).get(_field)) <= 0:
+            default_value = getattr(_pm_defaults, _field)
+            args['parts_and_merges'][_field] = default_value
+            validation_result.add_warning(
+                f"parts_and_merges.{_field} must be greater than 0, defaulting to {default_value}."
+            )
+
 
 def _validate_config(config: InstanceConfig, instance: dict, validation_result: ValidationResult):
     """Validate the configuration and add warnings/errors."""
@@ -229,6 +257,7 @@ def _validate_config(config: InstanceConfig, instance: dict, validation_result:
             config.query_completions.enabled if config.query_completions else False,
         ),
         ('query_errors', config.query_errors.enabled if config.query_errors else False),
+        ('parts_and_merges', config.parts_and_merges.enabled if config.parts_and_merges else False),
     ]
     for feature_name, _is_enabled in dbm_features:
         if instance.get(feature_name, {}).get('enabled') and not config.dbm:
@@ -270,6 +299,11 @@ def _apply_features(config: InstanceConfig, validation_result: ValidationResult)
         config.query_errors.enabled and config.dbm,
         None if config.dbm else "Requires `dbm: true`",
     )
+    validation_result.add_feature(
+        FeatureKey.PARTS_AND_MERGES,
+        config.parts_and_merges.enabled and config.dbm,
+        None if config.dbm else "Requires `dbm: true`",
+    )
     validation_result.add_feature(FeatureKey.SINGLE_ENDPOINT_MODE, config.single_endpoint_mode)
 
 

clickhouse/datadog_checks/clickhouse/config_models/dict_defaults.py

@@ -55,3 +55,19 @@ def instance_query_errors():
         max_samples_per_collection=1000,
         run_sync=False,
     )
+
+
+def instance_parts_and_merges():
+    return instance.PartsAndMerges(
+        enabled=True,
+        collection_interval=60,
+        max_parts_rows=500,
+        max_mutations_rows=200,
+        max_detached_parts_rows=1000,
+        max_replication_queue_rows=1000,
+        run_sync=False,
+        table_metrics_include_partition_tag=False,
+        table_metrics_max_tables=200,
+        stalled_merge_elapsed_threshold_seconds=3600,
+        stuck_replication_num_tries=3,
+    )

clickhouse/datadog_checks/clickhouse/config_models/instance.py

@@ -52,6 +52,24 @@ class MetricPatterns(BaseModel):
     include: Optional[tuple[str, ...]] = None
 
 
+class PartsAndMerges(BaseModel):
+    model_config = ConfigDict(
+        arbitrary_types_allowed=True,
+        frozen=True,
+    )
+    collection_interval: Optional[float] = None
+    enabled: Optional[bool] = None
+    max_detached_parts_rows: Optional[int] = None
+    max_mutations_rows: Optional[int] = None
+    max_parts_rows: Optional[int] = None
+    max_replication_queue_rows: Optional[int] = None
+    run_sync: Optional[bool] = None
+    stalled_merge_elapsed_threshold_seconds: Optional[int] = None
+    stuck_replication_num_tries: Optional[int] = None
+    table_metrics_include_partition_tag: Optional[bool] = None
+    table_metrics_max_tables: Optional[int] = None
+
+
 class QueryCompletions(BaseModel):
     model_config = ConfigDict(
         arbitrary_types_allowed=True,
@@ -121,6 +139,7 @@ class InstanceConfig(BaseModel):
     metric_patterns: Optional[MetricPatterns] = None
     min_collection_interval: Optional[float] = None
     only_custom_queries: Optional[bool] = None
+    parts_and_merges: Optional[PartsAndMerges] = None
     password: Optional[str] = None
     port: Optional[int] = None
     query_completions: Optional[QueryCompletions] = None

clickhouse/datadog_checks/clickhouse/data/conf.yaml.example

@@ -197,6 +197,83 @@ instances:
         #
         # samples_per_hour_per_query: 60
 
+    ## Configure parts and merges monitoring
+    #
+    # parts_and_merges:
+
+        ## @param enabled - boolean - optional - default: true
+        ## Enable collection of parts and merges monitoring. Requires `dbm: true`.
+        ## Emits per-table gauge metrics (clickhouse.table.parts.*, clickhouse.merges.*, clickhouse.mutations.*,
+        ## clickhouse.replication.*) for trend dashboards and alerting, plus a row-level event payload per
+        ## collection cycle consumed by the DBM Storage Health timeline view.
+        #
+        # enabled: true
+
+        ## @param collection_interval - number - optional - default: 60
+        ## Set the parts and merges collection interval (in seconds). All four system tables
+        ## (parts, merges, mutations, replication_queue) are queried on each run.
+        #
+        # collection_interval: 60
+
+        ## @param max_parts_rows - integer - optional - default: 500
+        ## Maximum number of rows to include in the per-cycle event payload from system.parts,
+        ## ordered by active_part_count descending. With `table_metrics_include_partition_tag`
+        ## disabled (default) rows are aggregated per-table server-side, so this caps tables.
+        ## When partition tagging is enabled, rows are per-partition and this caps partitions.
+        #
+        # max_parts_rows: 500
+
+        ## @param max_mutations_rows - integer - optional - default: 200
+        ## Maximum number of rows to include in the per-cycle event payload from system.mutations,
+        ## ordered by create_time ascending. Caps payload size when many mutations are pending.
+        #
+        # max_mutations_rows: 200
+
+        ## @param max_detached_parts_rows - integer - optional - default: 1000
+        ## Maximum number of rows returned from system.detached_parts per cycle.
+        ## Applied as a raw SQL LIMIT; increase only if your cluster has a pathological number
+        ## of (database, table, reason) combinations in detached storage.
+        #
+        # max_detached_parts_rows: 1000
+
+        ## @param max_replication_queue_rows - integer - optional - default: 1000
+        ## Maximum number of rows returned from system.replication_queue per cycle,
+        ## ordered by position ascending. Busy replicated clusters can exceed the default
+        ## during incidents; raise this if replication queue backlogs are being truncated.
+        #
+        # max_replication_queue_rows: 1000
+
+        ## @param table_metrics_include_partition_tag - boolean - optional - default: false
+        ## Add a partition tag to parts gauges (clickhouse.table.parts.*).
+        ## Disabled by default to avoid high cardinality: a table partitioned by day over two years
+        ## yields ~730 partitions and hundreds of additional metric series per table.
+        ## Enable only when per-partition alerting is explicitly needed.
+        #
+        # table_metrics_include_partition_tag: false
+
+        ## @param table_metrics_max_tables - integer - optional - default: 200
+        ## Cap per-table gauge emission to the top N tables by active part count.
+        ## Guards against pathological schemas with very large numbers of tables or partitions.
+        #
+        # table_metrics_max_tables: 200
+
+        ## @param stalled_merge_elapsed_threshold_seconds - integer - optional - default: 3600
+        ## Number of seconds a merge must be running before it is counted as stalled
+        ## in the `clickhouse.merges.stalled` gauge. Raise this threshold if long-running
+        ## merges are expected in your environment (e.g. large Wide-part merges on spinning
+        ## disks) to avoid false-positive stall alerts.
+        #
+        # stalled_merge_elapsed_threshold_seconds: 3600
+
+        ## @param stuck_replication_num_tries - integer - optional - default: 3
+        ## Minimum number of failed attempts (num_tries) before a replication queue entry
+        ## is counted as stuck in the `clickhouse.replication.stuck` gauge. Entries below
+        ## this threshold are still counted in `clickhouse.replication.queue.depth`.
+        ## Increase this value if transient failures routinely exceed the default before
+        ## self-healing.
+        #
+        # stuck_replication_num_tries: 3
+
     ## @param only_custom_queries - boolean - optional - default: false
     ## Set this parameter to `true` if you want to skip the integration's default metrics collection.
     ## Only metrics specified in `custom_queries` will be collected.

clickhouse/datadog_checks/clickhouse/features.py

@@ -23,6 +23,7 @@ class FeatureKey(Enum):
     QUERY_COMPLETIONS = "query_completions"
     EXPLAIN_PLANS = "explain_plans"
     QUERY_ERRORS = "query_errors"
+    PARTS_AND_MERGES = "parts_and_merges"
     SINGLE_ENDPOINT_MODE = "single_endpoint_mode"
 
 
@@ -33,6 +34,7 @@ class FeatureKey(Enum):
     FeatureKey.QUERY_COMPLETIONS: 'Query Completions',
     FeatureKey.QUERY_ERRORS: 'Query Errors',
     FeatureKey.EXPLAIN_PLANS: 'Explain Plans',
+    FeatureKey.PARTS_AND_MERGES: 'Parts and Merges',
     FeatureKey.SINGLE_ENDPOINT_MODE: 'Single Endpoint Mode',
 }