ClickHouse Replica Lag and Readonly Health Gate with Kestra

id: clickhouse-replica-lag-health-gate
namespace: company.team
description: |
  Query ClickHouse's own system.replicas table for a ReplicatedMergeTree
  table, classify it HEALTHY, AT_RISK (lagging or missing replicas), or
  UNREACHABLE (readonly - lost its ClickHouse Keeper session, or no row
  returned at all), alert Slack, and optionally run SYSTEM RESTART REPLICA
  to resync it. auto_remediate and dry_run are two independent gates: no
  table data is ever read, inserted, or deleted - only the replica's
  Keeper session state is touched.

triggers:
  - id: periodic_replica_check
    type: io.kestra.plugin.core.trigger.Schedule
    description: Check every 15 minutes - replication lag can accumulate quickly
      under heavy write load, so this needs a tighter cadence than a
      schema-drift check. Shipped disabled; enable it once
      clickhouse_url/target_table point at a real cluster and a manual run has
      been verified. Always fires in safe mode (auto_remediate false, dry_run
      true) regardless of this flow's own defaults.
    cron: "*/15 * * * *"
    disabled: true
    inputs:
      auto_remediate: "false"
      dry_run: "true"

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

  - id: target_table
    type: STRING
    displayName: "Target Table (db.table)"
    defaults: "default.events"
    description: Fully qualified ReplicatedMergeTree table audited in
      system.replicas, and restarted on remediation. Must use a
      ReplicatedMergeTree-family engine - system.replicas has no row for a plain
      MergeTree table.

  - id: max_absolute_delay_seconds
    type: INT
    displayName: "Max Absolute Delay (seconds)"
    defaults: 300
    description: Above this, system.replicas.absolute_delay for target_table counts
      as AT_RISK lag.

  - id: auto_remediate
    type: SELECT
    displayName: "Auto Remediate"
    defaults: "false"
    values:
      - "false"
      - "true"
    description: If "true" and the replica is AT_RISK or UNREACHABLE (readonly), run
      SYSTEM RESTART REPLICA on target_table (still gated by dry_run below). If
      "false", the gap is only reported.

  - id: dry_run
    type: SELECT
    displayName: "Dry Run"
    defaults: "true"
    values:
      - "true"
      - "false"
    description: If "true", remediation is only logged, SYSTEM RESTART REPLICA never
      runs. Must be explicitly "false", together with auto_remediate "true", for
      the replica to actually be restarted.

tasks:
  - id: check_replica_status
    type: io.kestra.plugin.jdbc.clickhouse.Query
    description: system.replicas is ClickHouse's own system table (columns database,
      table, is_readonly, is_session_expired, absolute_delay, queue_size,
      total_replicas, active_replicas, zookeeper_exception - verified from the
      official ClickHouse system-tables documentation). FETCH_ONE returns at
      most the single row for target_table; zero rows means the table is not a
      Replicated table, or does not exist.
    url: "{{ inputs.clickhouse_url }}"
    username: "{{ secret('CLICKHOUSE_USERNAME') }}"
    password: "{{ secret('CLICKHOUSE_PASSWORD') }}"
    fetchType: FETCH_ONE
    sql: |
      SELECT
        database,
        table,
        is_readonly,
        is_session_expired,
        absolute_delay,
        queue_size,
        total_replicas,
        active_replicas,
        zookeeper_exception
      FROM system.replicas
      WHERE concat(database, '.', table) = '{{ inputs.target_table }}'

  - id: evaluate_replica_presence
    type: io.kestra.plugin.core.flow.If
    description: UNREACHABLE when system.replicas returns no row for target_table at
      all - it either does not exist or is not a ReplicatedMergeTree-family
      table, distinct from a row that exists but reports readonly.
    condition: "{{ outputs.check_replica_status.row.database is defined }}"
    then:
      - id: evaluate_replica_health
        type: io.kestra.plugin.core.flow.If
        description: Also UNREACHABLE when is_readonly is 1 - the replica has lost its
          ClickHouse Keeper session and cannot serve writes, a more severe state
          than lag alone.
        condition: "{{ outputs.check_replica_status.row.is_readonly == 0 }}"
        then:
          - id: evaluate_replica_lag
            type: io.kestra.plugin.core.flow.If
            description: AT_RISK when the replica is writable but lagging past
              max_absolute_delay_seconds or has fewer active_replicas than
              total_replicas.
            condition: "{{ outputs.check_replica_status.row.absolute_delay >
              inputs.max_absolute_delay_seconds or
              outputs.check_replica_status.row.active_replicas <
              outputs.check_replica_status.row.total_replicas }}"
            then:
              - id: route_remediation
                type: io.kestra.plugin.core.flow.Switch
                description: auto_remediate only decides whether remediation is considered at
                  all; dry_run, checked separately below, decides whether SYSTEM
                  RESTART REPLICA actually runs.
                value: "{{ inputs.auto_remediate }}"
                cases:
                  "true":
                    - id: check_dry_run
                      type: io.kestra.plugin.core.flow.If
                      condition: "{{ inputs.dry_run == \"false\" }}"
                      then:
                        - id: restart_replica
                          type: io.kestra.plugin.jdbc.clickhouse.Query
                          description: "SYSTEM RESTART REPLICA [db.]table is ClickHouse's own statement
                            (verified from the official SYSTEM statements
                            documentation): it reinitializes the table's
                            ZooKeeper/Keeper session state, comparing current
                            state against Keeper and re-queuing tasks as needed.
                            The table is temporarily unavailable while this
                            runs."
                          url: "{{ inputs.clickhouse_url }}"
                          username: "{{ secret('CLICKHOUSE_USERNAME') }}"
                          password: "{{ secret('CLICKHOUSE_PASSWORD') }}"
                          fetchType: NONE
                          sql: "SYSTEM RESTART REPLICA {{ inputs.target_table }}"
                        - id: log_remediation_applied
                          type: io.kestra.plugin.core.log.Log
                          message: "Ran SYSTEM RESTART REPLICA on {{ inputs.target_table }}
                            (absolute_delay was {{
                            outputs.check_replica_status.row.absolute_delay }}s,
                            active_replicas {{
                            outputs.check_replica_status.row.active_replicas
                            }}/{{
                            outputs.check_replica_status.row.total_replicas
                            }})."
                      else:
                        - id: log_dry_run_remediation
                          type: io.kestra.plugin.core.log.Log
                          message: "DRY RUN: would run SYSTEM RESTART REPLICA on {{ inputs.target_table }}
                            (dry_run is true). No replica was restarted."
                  "false":
                    - id: log_remediation_disabled
                      type: io.kestra.plugin.core.log.Log
                      message: "auto_remediate is false: {{ inputs.target_table }} is AT_RISK
                        (absolute_delay={{
                        outputs.check_replica_status.row.absolute_delay }}s,
                        active_replicas={{
                        outputs.check_replica_status.row.active_replicas }}/{{
                        outputs.check_replica_status.row.total_replicas }}), but
                        no remediation was attempted."
              - id: alert_replica_at_risk
                type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
                description: Always alert on AT_RISK, independent of whether remediation ran.
                url: "{{ secret('SLACK_WEBHOOK_URL') }}"
                payload: |
                  {
                    "text": "ClickHouse replica health gate: AT_RISK. {{ inputs.target_table }} absolute_delay={{ outputs.check_replica_status.row.absolute_delay }}s (threshold {{ inputs.max_absolute_delay_seconds }}s), active_replicas={{ outputs.check_replica_status.row.active_replicas }}/{{ outputs.check_replica_status.row.total_replicas }}. auto_remediate: {{ inputs.auto_remediate }}. dry_run: {{ inputs.dry_run }}. Remediated: {{ outputs.restart_replica is defined }}. Execution {{ execution.id }}."
                  }
            else:
              - id: log_replica_healthy
                type: io.kestra.plugin.core.log.Log
                message: "HEALTHY: {{ inputs.target_table }} absolute_delay={{
                  outputs.check_replica_status.row.absolute_delay }}s (threshold
                  {{ inputs.max_absolute_delay_seconds }}s), active_replicas={{
                  outputs.check_replica_status.row.active_replicas }}/{{
                  outputs.check_replica_status.row.total_replicas }}."
        else:
          - id: alert_replica_readonly
            type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
            description: UNREACHABLE means is_readonly=1 - the replica lost its Keeper
              session and cannot accept writes, reported separately from AT_RISK
              lag since it is a harder failure.
            url: "{{ secret('SLACK_WEBHOOK_URL') }}"
            payload: |
              {
                "text": "ClickHouse replica health gate: UNREACHABLE. {{ inputs.target_table }} is readonly (is_readonly=1, is_session_expired={{ outputs.check_replica_status.row.is_session_expired }}), zookeeper_exception: {{ outputs.check_replica_status.row.zookeeper_exception }}. Execution {{ execution.id }}."
              }
    else:
      - id: alert_table_not_replicated
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: UNREACHABLE means system.replicas returned zero rows for
          target_table - it may not exist, or it may use a non-Replicated
          engine, in which case there is nothing for this gate to audit.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {
            "text": "ClickHouse replica health gate: UNREACHABLE. {{ inputs.target_table }} has no row in system.replicas - it may not exist, or may not use a ReplicatedMergeTree-family engine. Execution {{ execution.id }}."
          }

  - id: log_audit_complete
    type: io.kestra.plugin.core.log.Log
    description: Final status line for every run, independent of which branch fired,
      so the execution history always shows the headline numbers.
    message: "Replica health audit complete for {{ inputs.target_table }}.
      row_found={{ outputs.check_replica_status.row.database is defined }}.
      auto_remediate={{ inputs.auto_remediate }}. dry_run={{ inputs.dry_run }}."

errors:
  - id: alert_execution_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Alert when the flow itself fails outside the handled branches, for
      example an unreachable ClickHouse instance or invalid credentials.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {
        "text": "ClickHouse replica health gate FAILED in execution {{ execution.id }} for {{ inputs.target_table }}. The system.replicas audit did not complete; investigate before trusting the last known HEALTHY/AT_RISK/UNREACHABLE status."
      }

A ClickHouse replica that falls behind on its replication queue degrades reads silently - queries still return results, just increasingly stale ones. A replica that loses its ClickHouse Keeper session degrades harder: it flips to readonly and stops accepting writes outright, often without an obvious client-side error until something tries to insert into it. This blueprint reads ClickHouse's own system.replicas table to tell these two failure modes apart, classifies the result into a three-state risk level, alerts on anything but a clean bill of health, and can resync a lagging or readonly replica with ClickHouse's own SYSTEM RESTART REPLICA statement, behind two independent gates.

How it works

periodic_replica_check (io.kestra.plugin.core.trigger.Schedule) runs every 15 minutes, always forcing auto_remediate: "false" and dry_run: "true" through its own inputs: override regardless of this flow's defaults. Shipped disabled so you can validate a manual run first.
check_replica_status (io.kestra.plugin.jdbc.clickhouse.Query, fetchType: FETCH_ONE) reads system.replicas filtered to target_table, populating row.is_readonly, row.is_session_expired, row.absolute_delay, row.queue_size, row.active_replicas, row.total_replicas, row.zookeeper_exception - or no row at all.
evaluate_replica_presence (io.kestra.plugin.core.flow.If) branches on whether a row was returned.
- else (UNREACHABLE): alert_table_not_replicated reports that target_table has no row in system.replicas at all.
- then (row found): nested evaluate_replica_health (io.kestra.plugin.core.flow.If) branches on is_readonly == 0.
  - else (UNREACHABLE): alert_replica_readonly reports the readonly/lost-session state, with zookeeper_exception if ClickHouse recorded one.
  - then (writable): nested evaluate_replica_lag (io.kestra.plugin.core.flow.If) branches on absolute_delay > max_absolute_delay_seconds or active_replicas < total_replicas.
    - else (HEALTHY): log_replica_healthy records a clean check.
    - then (AT_RISK): route_remediation (io.kestra.plugin.core.flow.Switch on auto_remediate) either considers remediation ("true", gated again by dry_run through check_dry_run) or only logs that remediation is disabled ("false"). When both gates allow it, restart_replica runs SYSTEM RESTART REPLICA {{ target_table }}. alert_replica_at_risk always fires to Slack regardless of which remediation path ran.
log_audit_complete always runs last, printing the headline numbers regardless of which branch fired.
The errors block alerts Slack separately if the flow itself fails outright - an unreachable cluster or bad credentials must not read as "no risk" or "healthy".

What you get

A three-state answer (HEALTHY / AT_RISK / UNREACHABLE) sourced directly from ClickHouse's own replication state, not inferred from query latency or an external agent.
Lag and lost-session treated as two distinct severities, so a replica that is merely behind is never confused with one that has stopped accepting writes.
A separate UNREACHABLE path for a table that is not even a Replicated table, so a configuration mistake is never reported as a healthy replica.
Remediation scoped to exactly SYSTEM RESTART REPLICA on the one table you named - it cannot touch any other table or any row of data.
Two independent gates (auto_remediate, dry_run) between "risk detected" and "replica actually restarted".

Who it's for

Platform and database reliability teams running ClickHouse with ReplicatedMergeTree-family tables who need an early signal when a replica falls behind or loses its Keeper session.
On-call engineers who want one Slack alert with the exact lag and session numbers instead of running SELECT * FROM system.replicas by hand during an incident.
Anyone evaluating the ClickHouse plugin who wants a worked example of Query driving both system-table introspection (system.replicas) and a SYSTEM administrative statement (RESTART REPLICA), not just table reads and inserts.

Why orchestrate this with Kestra

system.replicas run by hand answers one moment in time; it does not decide a cadence, classify the result into a three-state risk level, gate a restart behind two independent confirmations, or notify anyone. Kestra supplies the schedule, the HEALTHY/AT_RISK/UNREACHABLE classification as a first-class branch, two independent authorization gates before any replica is restarted, and an execution history that shows exactly when a replica degraded and whether anything was done about it.

Prerequisites

A ClickHouse cluster with at least one table using a ReplicatedMergeTree-family engine and a working ClickHouse Keeper (or ZooKeeper) ensemble - system.replicas has no row at all for a plain MergeTree table.
A database user with SELECT on system.replicas and, if auto_remediate will ever be "true", permission to run SYSTEM RESTART REPLICA.
A Slack incoming webhook for alerts.

Local testing: docker run -d --name clickhouse-replica-gate -p 8124:8123 -p 9001:9000 clickhouse/clickhouse-server:latest docker network connect YOUR_KESTRA_NETWORK clickhouse-replica-gate (only needed if the Kestra Worker runs in a separate Docker network than this container; set clickhouse_url to jdbc:clickhouse://clickhouse-replica-gate:8123/default from inside that network, or jdbc:clickhouse://localhost:8124/default from the host). Host port 8124 is used here specifically so it never collides with Kestra's own UI on 8080. A single-node container has no real Keeper ensemble to replicate against, so a genuinely meaningful ReplicatedMergeTree test needs either ClickHouse Keeper configured on the same node or a second container - a true replica-lag scenario cannot be fully exercised on one bare container.

Secrets

CLICKHOUSE_USERNAME / CLICKHOUSE_PASSWORD: credentials used by every io.kestra.plugin.jdbc.clickhouse.Query task in this flow.
SLACK_WEBHOOK_URL: Slack incoming webhook used by alert_replica_at_risk, alert_replica_readonly, alert_table_not_replicated, and the errors block.
In Kestra OSS (no Enterprise secrets backend), secrets are supplied as environment variables prefixed SECRET_, base64-encoded, and read back in flows with {{ secret('NAME') }} - for example SECRET_CLICKHOUSE_PASSWORD=$(echo -n '<your-local-password>' | base64). This keeps credentials out of the flow YAML but, per Kestra's own documentation, offers no encryption at rest or access control beyond the host environment; use the Enterprise secrets backend for stronger guarantees.

Inputs

clickhouse_url (STRING, default jdbc:clickhouse://localhost:8124/default): JDBC URL.
target_table (STRING, default default.events): fully qualified db.table audited and, if remediated, restarted.
max_absolute_delay_seconds (INT, default 300): lag ceiling before a writable replica counts as AT_RISK.
auto_remediate (SELECT: "false", "true"; default "false"): must be "true" for remediation to even be considered.
dry_run (SELECT: "true", "false"; default "true"): must be explicitly "false", together with auto_remediate: "true", for SYSTEM RESTART REPLICA to actually run.

Outputs

outputs.check_replica_status.row.*: the full system.replicas row for target_table every run (is_readonly, is_session_expired, absolute_delay, queue_size, active_replicas, total_replicas, zookeeper_exception), or no row at all if the table is not Replicated.
outputs.restart_replica: present only when remediation actually ran; SYSTEM RESTART REPLICA returns no result set, so this output carries no row data of its own, only its existence matters.

Quick start

Start ClickHouse locally with the command above and create a ReplicatedMergeTree table (requires Keeper/ZooKeeper configured - see Prerequisites) under target_table.
Add the CLICKHOUSE_USERNAME, CLICKHOUSE_PASSWORD, and SLACK_WEBHOOK_URL secrets.
Run the flow manually with the defaults and confirm log_replica_healthy fires on a freshly created, idle replica.
Pause replication or disconnect Keeper briefly to push absolute_delay past max_absolute_delay_seconds, then re-run and confirm alert_replica_at_risk fires.
Run once with auto_remediate: "true" and dry_run: "true" and confirm log_dry_run_remediation describes the restart without running it.
Run once with auto_remediate: "true" and dry_run: "false" and confirm restart_replica runs, then verify with SELECT * FROM system.replicas that absolute_delay has recovered.
Enable periodic_replica_check once you trust the check; it always runs in the safe auto_remediate: "false" / dry_run: "true" mode regardless of what you leave the flow's own defaults set to.

How to extend

Loop over several target_table values with io.kestra.plugin.core.flow.Loop to run the same gate across every Replicated table in a cluster in one flow.
Add a check on queue_size or future_parts alongside absolute_delay to catch a replica that is building a backlog before its lag clock actually starts moving.
On the UNREACHABLE (readonly) branch, use SYSTEM RESTORE REPLICA instead of SYSTEM RESTART REPLICA when local data exists but Keeper metadata was lost - ClickHouse's own documentation describes RESTORE REPLICA as the correct recovery for that specific case, which this flow does not attempt to distinguish from a routine session expiry.
Route the UNREACHABLE (readonly) branch to a different Slack channel than AT_RISK lag, since a lost Keeper session is a harder failure than lag alone.

Pitfalls

system.replicas has no row at all for a non-Replicated table. A plain MergeTree table pointed at by target_table makes evaluate_replica_presence report UNREACHABLE on every run, which is correct behavior, not a bug - this gate only has something to audit for ReplicatedMergeTree-family engines.
is_readonly and lag are different failure modes, not degrees of the same one. A replica can have absolute_delay = 0 and still be is_readonly = 1 immediately after losing its Keeper session, before any lag has had time to accumulate - that is why evaluate_replica_health is checked before evaluate_replica_lag, not folded into the same condition.
There is no structured "replica status" or "restart replica" task in this plugin. Both system.replicas introspection and the SYSTEM RESTART REPLICA statement go through the same generic io.kestra.plugin.jdbc.clickhouse.Query task used elsewhere in this repo's own ClickHouse flows (system.mutations, system.parts).
SYSTEM RESTART REPLICA makes the table temporarily unavailable. Per ClickHouse's own documentation, this reinitializes the Keeper session and the table cannot serve queries normally while that happens - it is a real, if usually brief, availability cost, which is exactly why it sits behind both auto_remediate and dry_run.
auto_remediate and dry_run are independent gates, not a single boolean. Both must be "true"/"false" respectively for SYSTEM RESTART REPLICA to run; setting only one leaves the other still blocking.
Switch case keys "true"/"false" are quoted strings. auto_remediate renders as the literal string "true" or "false"; unquoted true:/false: YAML map keys would parse as booleans instead and would not match.
This blueprint is UNTESTED against a live ClickHouse cluster. The system.replicas column names are taken verbatim from ClickHouse's own system-tables documentation, and SYSTEM RESTART REPLICA syntax from ClickHouse's own SYSTEM statements documentation - but no Docker container or Kestra engine was run to execute this flow end to end, and a true multi-replica lag scenario was not exercised at all.

ClickHouse Replica Lag and Readonly Health Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Outputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

New to Kestra?

ClickHouse Replica Lag and Readonly Health Gate

How it works

What you get

Who it's for

Why orchestrate this with Kestra

Prerequisites

Secrets

Inputs

Outputs

Quick start

How to extend

Pitfalls

Links

Tasks

Orchestrate with Kestra

Share this Blueprint

Create a Blueprint

More Data Blueprints

Chroma Vector Store Dangling Document and Quarantine Eviction Guard

Citus Inactive Worker Reactivate Guard

Citus Colocation Violation Guard

ChromaDB Vector Collection Snapshot and AWS S3 Disaster Recovery Pipeline

Citus Reference Table Replica Guard

Score change requests on risk and build the CAB agenda, pre-approving the low-risk ones

More Infrastructure Blueprints

Kafka Client Quota Guard

Kafka ACL Baseline Restore Guard

Archive ownerless failed Pod diagnostics before cleanup

Kafka Connect Sink Poison Pill Offset Rewind and Skip Gate

Self-heal Kafka Connect connectors with a restart budget and escalation

Kafka ACL Overly Broad Grant Audit

New to Kestra?