ClickHouse Mutation and Partition Parts Watchdog Sentinel

id: clickhouse-mutation-and-parts-watchdog
namespace: company.team
description: |
  Audit ClickHouse system health by monitoring active table mutations (system.mutations),
  detecting partition parts explosion (system.parts), calculating merge pressure,
  and dispatching operational diagnostics to Slack before query performance degrades.

triggers:
  - id: scheduled_watchdog
    type: io.kestra.plugin.core.trigger.Schedule
    description: Periodic off-peak cluster health and partition watchdog check.
      Shipped disabled by default.
    cron: "*/30 * * * *"
    disabled: true

  - id: manual_webhook
    type: io.kestra.plugin.core.trigger.Webhook
    description: Authenticated webhook trigger for on-demand cluster audits from
      CI/CD, incident response drills, or ChatOps.
    key: "{{ secret('WEBHOOK_KEY') }}"

inputs:
  - id: clickhouse_url
    type: STRING
    displayName: ClickHouse JDBC URL
    description: JDBC connection URL for target ClickHouse database instance.
    defaults: "jdbc:clickhouse://clickhouse:8123/default"

  - id: max_mutation_age_minutes
    type: INT
    displayName: Max Mutation Age (Minutes)
    description: Age threshold in minutes for in-progress mutations before raising a
      stuck mutation alert.
    defaults: 60

  - id: max_active_parts_threshold
    type: INT
    displayName: Max Active Parts Threshold
    description: Maximum active parts per table partition before flagging excessive
      parts explosion and "too many parts" risk.
    defaults: 150

tasks:
  - id: check_stuck_mutations
    type: io.kestra.plugin.jdbc.clickhouse.Query
    description: Query system.mutations to detect in-progress table mutations
      exceeding the runtime age threshold.
    url: "{{ inputs.clickhouse_url }}"
    username: "{{ secret('CLICKHOUSE_USERNAME') }}"
    password: "{{ secret('CLICKHOUSE_PASSWORD') }}"
    fetchType: FETCH
    sql: |
      SELECT
        database,
        table,
        mutation_id,
        command,
        create_time,
        parts_to_do,
        is_done,
        latest_fail_reason,
        dateDiff('minute', create_time, now()) AS age_minutes
      FROM system.mutations
      WHERE is_done = 0
        AND dateDiff('minute', create_time, now()) >= {{ inputs.max_mutation_age_minutes }}
      ORDER BY create_time ASC;

  - id: check_partition_parts_explosion
    type: io.kestra.plugin.jdbc.clickhouse.Query
    description: Query system.parts to detect partition-level parts explosion
      exceeding operational merge limits.
    url: "{{ inputs.clickhouse_url }}"
    username: "{{ secret('CLICKHOUSE_USERNAME') }}"
    password: "{{ secret('CLICKHOUSE_PASSWORD') }}"
    fetchType: FETCH
    sql: |
      SELECT
        database,
        table,
        partition,
        count() AS active_parts_count,
        round(sum(bytes_on_disk) / (1024 * 1024), 2) AS total_size_mb,
        max(modification_time) AS latest_part_time
      FROM system.parts
      WHERE active = 1
        AND database NOT IN ('system', 'information_schema')
      GROUP BY database, table, partition
      HAVING count() >= {{ inputs.max_active_parts_threshold }}
      ORDER BY active_parts_count DESC
      LIMIT 20;

  - id: emit_cluster_health_metrics
    type: io.kestra.plugin.core.output.OutputValues
    description: Export calculated cluster health diagnostics into execution outputs.
    values:
      stuck_mutations_count: "{{ outputs.check_stuck_mutations.size }}"
      exploded_partitions_count: "{{ outputs.check_partition_parts_explosion.size }}"
      cluster_url: "{{ inputs.clickhouse_url }}"
      inspected_at: "{{ now() }}"

  - id: evaluate_cluster_health
    type: io.kestra.plugin.core.flow.If
    description: Check if any stuck mutations or partition parts count breached
      warning thresholds.
    condition: "{{ outputs.check_stuck_mutations.size > 0 or
      outputs.check_partition_parts_explosion.size > 0 }}"
    then:
      - id: notify_cluster_alert_slack
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Dispatch high-priority Slack diagnostic alert detailing flagged
          mutations and partition bloat.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        messageText: |
          :warning: *ClickHouse Cluster Health Watchdog Alert*
          *Target URL:* `{{ inputs.clickhouse_url }}`
          *Execution:* `{{ execution.id }}`
          *Stuck Mutations:* `{{ outputs.check_stuck_mutations.size }}` (age >= {{ inputs.max_mutation_age_minutes }}m)
          *Partitions with Parts Explosion:* `{{ outputs.check_partition_parts_explosion.size }}` (parts >= {{ inputs.max_active_parts_threshold }})

          {% if outputs.check_stuck_mutations.size > 0 %}*Top Stuck Mutations:*
          {% for m in outputs.check_stuck_mutations.rows %}• `{{ m.database }}.{{ m.table }}` | ID: `{{ m.mutation_id }}` | Age: {{ m.age_minutes }}m | Parts to do: {{ m.parts_to_do }} | Reason: {{ m.latest_fail_reason | default('In progress', true) }}
          {% endfor %}{% endif %}
          {% if outputs.check_partition_parts_explosion.size > 0 %}*Top High-Part Partitions:*
          {% for p in outputs.check_partition_parts_explosion.rows %}• `{{ p.database }}.{{ p.table }}` [Partition: `{{ p.partition }}`] | Active Parts: {{ p.active_parts_count }} | Size: {{ p.total_size_mb }} MB
          {% endfor %}{% endif %}
          *Recommended Action:* Check `system.merges` for disk/CPU saturation or run `OPTIMIZE TABLE ... [PARTITION ...]` during maintenance windows.
    else:
      - id: log_cluster_healthy
        type: io.kestra.plugin.core.log.Log
        description: Log healthy cluster verification status.
        message: "ClickHouse cluster health verified on {{ inputs.clickhouse_url }}: 0
          stuck mutations and 0 partitions exceeding active parts threshold ({{
          inputs.max_active_parts_threshold }} parts)."

errors:
  - id: alert_on_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert on-call data engineers if the health check query or watchdog
      flow encounters an unhandled failure.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    messageText: |
      :rotating_light: *ClickHouse Health Watchdog FAILED*
      Flow `{{ flow.id }}` in namespace `{{ flow.namespace }}` failed during execution `{{ execution.id }}`.
      Please check ClickHouse endpoint availability and credentials.