ClickHouse Read-Only Replica Restore Guard

id: clickhouse-readonly-replica-restore-guard
namespace: company.team

inputs:
  - id: clickhouse_url
    type: STRING
    defaults: jdbc:clickhouse://localhost:8123/default
    description: ClickHouse JDBC URL for the server that holds the replicated
      tables. Example jdbc:clickhouse://clickhouse:8123/default.

  - id: use_secret_auth
    type: BOOL
    defaults: true
    description: Leave true in production so the connection reads
      CLICKHOUSE_USERNAME and CLICKHOUSE_PASSWORD from secrets. Set false only
      to pass credentials inline in a disposable test environment.

  - id: clickhouse_username
    type: STRING
    defaults: default
    description: Username used only when use_secret_auth is false. Ignored in production.

  - id: clickhouse_password
    type: STRING
    defaults: ""
    description: Password used only when use_secret_auth is false. Ignored in production.

  - id: database
    type: STRING
    defaults: default
    description: Database whose ReplicatedMergeTree tables are checked.

  - id: table
    type: STRING
    defaults: ""
    description: Optional single table to restrict the check to. Leave empty to
      check every replicated table in the database.

  - id: query_timeout_seconds
    type: INT
    defaults: 30
    description: Ceiling in seconds on the two reads that probe Keeper. A read of
      system.zookeeper blocks rather than erroring while Keeper is unreachable,
      so this turns a Keeper outage into a failed task and a Slack alert instead
      of a run that hangs.

  - id: dry_run
    type: BOOL
    defaults: true
    description: When true, report which replicas lost their Keeper metadata without
      running SYSTEM RESTORE REPLICA.

  - id: auto_remediate
    type: BOOL
    defaults: false
    description: When true and dry_run is false, run SYSTEM RESTORE REPLICA on the
      targeted tables behind an approval gate, then re-read system.replicas.

variables:
  survey_sql: |
    SELECT
      r.database AS database,
      r.table AS table,
      toString(r.is_session_expired) AS session_expired,
      toString(coalesce(z.znode_children, 0)) AS znode_count,
      toString(r.total_replicas) AS replicas_known,
      toString(r.active_replicas) AS replicas_active,
      multiIf(
        r.is_session_expired = 1, 'RESTART case, the Keeper session expired so the metadata is still in Keeper',
        coalesce(z.znode_children, 0) = 0, 'RESTORE case, the Keeper metadata for this table is gone',
        'REVIEW case, read-only with its Keeper metadata present and its session alive'
      ) AS recovery_case,
      replaceAll(replaceAll(replaceAll(substringUTF8(r.zookeeper_exception, 1, 160), '\\', '/'), '"', ''), '\n', ' ') AS zookeeper_exception_short
    FROM system.replicas AS r
    LEFT JOIN (
      SELECT path, count() AS znode_children
      FROM system.zookeeper
      WHERE path IN (
        SELECT zookeeper_path FROM system.replicas
        WHERE is_readonly = 1 AND is_session_expired = 0
      )
      GROUP BY path
    ) AS z ON r.zookeeper_path = z.path
    WHERE r.is_readonly = 1
      AND r.database = '{{ inputs.database }}'
      AND ('{{ inputs.table }}' = '' OR r.table = '{{ inputs.table }}')
    ORDER BY r.database, r.table
    SETTINGS max_execution_time = {{ inputs.query_timeout_seconds }}

  detect_sql: |
    SELECT
      r.database AS database,
      r.table AS table,
      r.replica_name AS replica_name,
      r.zookeeper_path AS zookeeper_path,
      r.replica_path AS replica_path,
      toString(coalesce(p.local_active_parts, 0)) AS local_active_parts,
      toString(coalesce(p.local_rows, 0)) AS local_rows,
      replaceAll(replaceAll(replaceAll(substringUTF8(r.zookeeper_exception, 1, 160), '\\', '/'), '"', ''), '\n', ' ') AS zookeeper_exception_short
    FROM system.replicas AS r
    LEFT JOIN (
      SELECT path, count() AS znode_children
      FROM system.zookeeper
      WHERE path IN (
        SELECT zookeeper_path FROM system.replicas
        WHERE is_readonly = 1 AND is_session_expired = 0
      )
      GROUP BY path
    ) AS z ON r.zookeeper_path = z.path
    LEFT JOIN (
      SELECT database, table, count() AS local_active_parts, sum(rows) AS local_rows
      FROM system.parts
      WHERE active
      GROUP BY database, table
    ) AS p ON r.database = p.database AND r.table = p.table
    WHERE r.is_readonly = 1
      AND r.is_session_expired = 0
      AND coalesce(z.znode_children, 0) = 0
      AND r.database = '{{ inputs.database }}'
      AND ('{{ inputs.table }}' = '' OR r.table = '{{ inputs.table }}')
    ORDER BY r.database, r.table
    SETTINGS max_execution_time = {{ inputs.query_timeout_seconds }}

triggers:
  - id: hourly_readonly_replica_check
    type: io.kestra.plugin.core.trigger.Schedule
    disabled: true
    cron: "25 * * * *"
    description: Hourly sweep for replicas stuck read-only. Disabled by default and
      pinned to safe mode so it reports the lost metadata without restoring
      anything until an operator enables it.
    inputs:
      dry_run: true
      auto_remediate: false

tasks:
  - id: survey_readonly
    type: io.kestra.plugin.jdbc.clickhouse.Query
    description: >-
      Read every read-only replica in the database and label each one with the
      recovery statement it actually needs. A replica whose Keeper session
      expired keeps its metadata in Keeper and comes back on its own, so it is a
      RESTART case. A replica whose znode is gone cannot come back without
      rebuilding the metadata, so it is a RESTORE case. The split is read from
      is_session_expired plus a direct read of the table's znode path through
      system.zookeeper.
    url: "{{ inputs.clickhouse_url }}"
    username: "{{ inputs.use_secret_auth ? secret('CLICKHOUSE_USERNAME') :
      inputs.clickhouse_username }}"
    password: "{{ inputs.use_secret_auth ? secret('CLICKHOUSE_PASSWORD') :
      inputs.clickhouse_password }}"
    fetchType: FETCH
    sql: "{{ render(vars.survey_sql) }}"

  - id: detect
    type: io.kestra.plugin.jdbc.clickhouse.Query
    description: >-
      Keep only the RESTORE cases, the replicas that are read-only with a live
      Keeper session and no znode at their zookeeper_path. This is the single
      list the remediation loop iterates, so a RESTART case can never reach
      SYSTEM RESTORE REPLICA. It also reports the local active part count and
      row count behind each target, because RESTORE rebuilds the Keeper metadata
      from the parts on this host and nothing else.
    url: "{{ inputs.clickhouse_url }}"
    username: "{{ inputs.use_secret_auth ? secret('CLICKHOUSE_USERNAME') :
      inputs.clickhouse_username }}"
    password: "{{ inputs.use_secret_auth ? secret('CLICKHOUSE_PASSWORD') :
      inputs.clickhouse_password }}"
    fetchType: FETCH
    sql: "{{ render(vars.detect_sql) }}"

  - id: summarize
    type: io.kestra.plugin.core.output.OutputValues
    description: Materialize the targeted table list, the comma separated names used
      by the re-read, the per target evidence plus the full read-only survey
      once, so the log, the Slack alert and the failure message all read the
      same value.
    values:
      target_list: >-
        {{ outputs.detect.rows | jq('map(.database + "." + .table) | join(",
        ")') | first }}
      table_csv: >-
        {{ outputs.detect.rows | jq('map(.table) | join(",")') | first }}
      target_evidence: >-
        {{ outputs.detect.rows | jq('map(.database + "." + .table + " [replica "
        + .replica_name + ", znode " + .zookeeper_path + " missing, " +
        .local_active_parts + " local active parts holding " + .local_rows + "
        rows] " + (if .zookeeper_exception_short == "" then "no Keeper exception
        recorded" else .zookeeper_exception_short end)) | join(" || ")') | first
        }}
      survey_list: >-
        {{ outputs.survey_readonly.rows | jq('map(.database + "." + .table + "
        is " + .recovery_case) | join(" || ")') | first }}

  - id: log_detect
    type: io.kestra.plugin.core.log.Log
    message: |
      ClickHouse {{ inputs.database }}: {{ outputs.survey_readonly.size }} read-only replica(s) found, {{ outputs.detect.size }} of them need SYSTEM RESTORE REPLICA.
      Every read-only replica and its case: {{ outputs.summarize.values.survey_list }}
      Restore targets: {{ outputs.summarize.values.target_list }}
      Evidence per target: {{ outputs.summarize.values.target_evidence }}
      Mode: dry_run={{ inputs.dry_run }} auto_remediate={{ inputs.auto_remediate }}
      SYSTEM RESTART REPLICA is the wrong tool for a RESTORE case. It only reinitialises the session against Keeper, so with the metadata genuinely gone it restarts straight back into read-only. SYSTEM RESTORE REPLICA re-creates the metadata from the parts on local disk, which also means any part that lived only on another replica and never replicated here is not recovered by it.

  - id: guard
    type: io.kestra.plugin.core.flow.If
    condition: "{{ outputs.detect.size > 0 }}"
    then:
      - id: findings_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: Alert on every detection, in dry run and in remediate mode, so a
          human sees which replicas lost their Keeper metadata even when nothing
          is restored.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {"text": "ClickHouse read-only replica guard: {{ outputs.detect.size }} replica(s) in {{ inputs.database }} are read-only because their Keeper metadata is gone, so SYSTEM RESTART REPLICA cannot fix them. {{ outputs.summarize.values.target_evidence }} Full read-only survey: {{ outputs.summarize.values.survey_list }} Mode dry_run={{ inputs.dry_run }} auto_remediate={{ inputs.auto_remediate }}."}

      - id: mode
        type: io.kestra.plugin.core.flow.If
        condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
        then:
          - id: approval_gate
            type: io.kestra.plugin.core.flow.Pause
            description: >-
              Hold the run until a human approves the restore. Approving means
              two things have been checked. The metadata loss is real rather
              than a Keeper outage in progress. This host also holds the parts
              worth keeping, because SYSTEM RESTORE REPLICA rebuilds the Keeper
              metadata from local parts only.
            pauseDuration: PT2H
            behavior: FAIL
            onResume:
              - id: approved
                type: BOOL
                defaults: false
                description: Set true to run SYSTEM RESTORE REPLICA on the listed tables.
                  Anything else leaves them alone.

          - id: apply
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.approval_gate.onResume.approved ?? false }}"
            then:
              - id: remediation_restore_loop
                type: io.kestra.plugin.core.flow.Loop
                description: Remediation loop. One SYSTEM RESTORE REPLICA per targeted table,
                  concurrencyLimit 1 so the restores rebuild one set of Keeper
                  znodes at a time instead of racing each other.
                concurrencyLimit: 1
                values: "{{ outputs.detect.rows | jq('map({database: .database, table:
                  .table})') | first | toJson }}"
                tasks:
                  - id: restore_one
                    type: io.kestra.plugin.jdbc.clickhouse.Query
                    description: Rebuild one table's Keeper metadata from its local parts.
                      allowFailure so a restore the server refuses does not stop
                      the pass, which leaves the re-read as the real gate.
                    allowFailure: true
                    url: "{{ inputs.clickhouse_url }}"
                    username: "{{ inputs.use_secret_auth ? secret('CLICKHOUSE_USERNAME') :
                      inputs.clickhouse_username }}"
                    password: "{{ inputs.use_secret_auth ? secret('CLICKHOUSE_PASSWORD') :
                      inputs.clickhouse_password }}"
                    fetchType: NONE
                    sql: "SYSTEM RESTORE REPLICA `{{ fromJson(item.value).database }}`.`{{
                      fromJson(item.value).table }}`"

                  - id: restore_log
                    type: io.kestra.plugin.core.log.Log
                    message: "Issued SYSTEM RESTORE REPLICA for {{ fromJson(item.value).database
                      }}.{{ fromJson(item.value).table }}."

              - id: reverify
                type: io.kestra.plugin.jdbc.clickhouse.Query
                description: >-
                  Re-read system.replicas for exactly the tables that were
                  restored and ask one question only, is this replica still
                  read-only. The check deliberately drops the znode and session
                  predicates the fix moves, so a restore that re-creates some
                  znodes while the table stays read-only fails the run instead
                  of reading clean.
                url: "{{ inputs.clickhouse_url }}"
                username: "{{ inputs.use_secret_auth ? secret('CLICKHOUSE_USERNAME') :
                  inputs.clickhouse_username }}"
                password: "{{ inputs.use_secret_auth ? secret('CLICKHOUSE_PASSWORD') :
                  inputs.clickhouse_password }}"
                fetchType: FETCH
                sql: |
                  SELECT
                    database AS database,
                    table AS table,
                    toString(is_readonly) AS readonly_flag,
                    toString(is_session_expired) AS session_expired,
                    toString(total_replicas) AS replicas_known,
                    toString(active_replicas) AS replicas_active,
                    replaceAll(replaceAll(replaceAll(substringUTF8(zookeeper_exception, 1, 160), '\\', '/'), '"', ''), '\n', ' ') AS zookeeper_exception_short
                  FROM system.replicas
                  WHERE database = '{{ inputs.database }}'
                    AND has(splitByChar(',', '{{ outputs.summarize.values.table_csv }}'), table)
                    AND is_readonly = 1
                  ORDER BY table

              - id: verify_result
                type: io.kestra.plugin.core.flow.If
                condition: "{{ outputs.reverify.size > 0 }}"
                then:
                  - id: remediation_failed
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: >-
                      {{ outputs.reverify.size }} of {{ outputs.detect.size }}
                      restored replica(s) in {{ inputs.database }} are still
                      read-only: {{ outputs.reverify.rows | jq('map(.database +
                      "." + .table + " [session_expired " + .session_expired +
                      ", total_replicas " + .replicas_known + "] " + (if
                      .zookeeper_exception_short == "" then "no Keeper exception
                      recorded" else .zookeeper_exception_short end)) | join("
                      || ")') | first }}. The restore did not put the table back
                      in a writable state, so writes are still refused. Check
                      the Keeper cluster is healthy, that the user holds the
                      SYSTEM RESTORE REPLICA privilege, then run again.
                else:
                  - id: remediated
                    type: io.kestra.plugin.core.log.Log
                    message: "Restored and verified. {{ outputs.detect.size }} replica(s) were
                      read-only with their Keeper metadata gone, {{
                      outputs.reverify.size }} still are. Every targeted table
                      accepts writes again, rebuilt from the parts already on
                      local disk."
            else:
              - id: not_approved
                type: io.kestra.plugin.core.log.Log
                message: "Approval not granted, so nothing was restored. {{ outputs.detect.size
                  }} replica(s) are still read-only in {{ inputs.database }}: {{
                  outputs.summarize.values.target_list }}."
        else:
          - id: report_only
            type: io.kestra.plugin.core.log.Log
            message: "Report only (dry_run={{ inputs.dry_run }}, auto_remediate={{
              inputs.auto_remediate }}). {{ outputs.detect.size }} replica(s)
              would get SYSTEM RESTORE REPLICA: {{
              outputs.summarize.values.target_list }}. No change made."
    else:
      - id: clean
        type: io.kestra.plugin.core.log.Log
        message: "No replica in {{ inputs.database }} needs SYSTEM RESTORE REPLICA. {{
          outputs.survey_readonly.size }} read-only replica(s) were seen in
          total: {{ outputs.summarize.values.survey_list }}. A read-only replica
          whose Keeper session expired recovers on its own once Keeper is
          reachable, so it is left alone here."

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Post to Slack on any failure, including an unreachable server, a
      refused restore or replicas still read-only after a restore pass.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "ClickHouse read-only replica restore guard FAILED for {{ inputs.database }} (execution {{ execution.id }}). Either the server was unreachable, the restore was refused or a targeted replica is still read-only after the restore pass. Check the run."}