ClickHouse Failed Dictionary Reload Guard

id: clickhouse-failed-dictionary-reload-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 dictionaries.
      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 dictionaries are checked.

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

  - id: include_not_loaded
    type: BOOL
    defaults: false
    description: When false, only dictionaries carrying a last_exception are
      targeted. When true, also target dictionaries whose status is not LOADED.
      Keep it false on a server with lazy loading, where NOT_LOADED just means
      nothing has queried the dictionary yet.

  - id: dry_run
    type: BOOL
    defaults: true
    description: When true, report the failing dictionaries with their exceptions
      without reloading anything.

  - id: auto_remediate
    type: BOOL
    defaults: false
    description: When true and dry_run is false, reload the targeted dictionaries
      behind an approval gate, then re-read system.dictionaries.

variables:
  detect_sql: |
    SELECT
      database AS database,
      name AS name,
      toString(status) AS status,
      element_count AS element_count,
      toString(last_successful_update_time) AS last_good_at,
      if(element_count > 0 AND last_successful_update_time > toDateTime(0), 'previously loaded so dictGet still answers with stale values', 'never loaded so dictGet fails') AS serving,
      replaceAll(replaceAll(replaceAll(substringUTF8(last_exception, 1, 180), '\\', '/'), '"', ''), '\n', ' ') AS last_exception_short
    FROM system.dictionaries
    WHERE database = '{{ inputs.database }}'
      AND ('{{ inputs.dictionary }}' = '' OR name = '{{ inputs.dictionary }}')
      AND (
        last_exception != ''
        OR ({{ inputs.include_not_loaded ? 1 : 0 }} = 1 AND toString(status) NOT IN ('LOADED', 'LOADED_AND_RELOADING'))
      )
    ORDER BY name

triggers:
  - id: hourly_dictionary_check
    type: io.kestra.plugin.core.trigger.Schedule
    disabled: true
    cron: "15 * * * *"
    description: Hourly dictionary load sweep. Disabled by default and pinned to
      safe mode so it reports the exceptions without reloading until an operator
      enables it.
    inputs:
      dry_run: true
      auto_remediate: false

tasks:
  - id: detect
    type: io.kestra.plugin.jdbc.clickhouse.Query
    description: Read system.dictionaries for the database and keep every dictionary
      carrying a last_exception, plus, when opted in, any dictionary whose
      status is not LOADED. A dictionary that loaded once then failed to refresh
      keeps status LOADED and records the exception here, so last_exception is
      the signal that catches the stale case.
    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 dictionary list, the comma separated names
      used by the re-read, plus the per dictionary exception once, so the log,
      the Slack alert and the failure message all read the same value.
    values:
      dict_list: >-
        {{ outputs.detect.rows | jq('map(.name) | join(", ")') | first }}
      name_csv: >-
        {{ outputs.detect.rows | jq('map(.name) | join(",")') | first }}
      reason_list: >-
        {{ outputs.detect.rows | jq('map(.name + " [status " + .status + ", " +
        .serving + ", last good " + .last_good_at + "] " + (if
        .last_exception_short == "" then "no exception recorded" else
        .last_exception_short end)) | join(" || ")') | first }}

  - id: log_detect
    type: io.kestra.plugin.core.log.Log
    message: |
      ClickHouse {{ inputs.database }}: {{ outputs.detect.size }} dictionary(s) failing to load.
      Targeted: {{ outputs.summarize.values.dict_list }}
      Why each one failed: {{ outputs.summarize.values.reason_list }}
      Mode: dry_run={{ inputs.dry_run }} auto_remediate={{ inputs.auto_remediate }} include_not_loaded={{ inputs.include_not_loaded }}
      Read the exceptions before approving a reload. A reload re-runs the dictionary source, so it only helps once the cause is cleared.

  - 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 the failing dictionaries with their exceptions even when
          nothing is reloaded.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {"text": "ClickHouse failed dictionary guard: {{ outputs.detect.size }} dictionary(s) in {{ inputs.database }} failing to load. {{ outputs.summarize.values.reason_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 reload. Approving means the
              root cause in the exception above has been cleared, because a
              reload re-runs the same source query or fetch and fails again if
              it has not.
            pauseDuration: PT2H
            behavior: FAIL
            onResume:
              - id: approved
                type: BOOL
                defaults: false
                description: Set true to reload the listed dictionaries. Anything else leaves
                  them alone.

          - id: apply
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.approval_gate.onResume.approved ?? false }}"
            then:
              - id: remediation_reload_loop
                type: io.kestra.plugin.core.flow.Loop
                description: Remediation loop. One SYSTEM RELOAD DICTIONARY per targeted
                  dictionary, concurrencyLimit 1 so the reloads run in turn
                  instead of competing for the same source.
                concurrencyLimit: 1
                values: "{{ outputs.detect.rows | jq('map({name: .name})') | first | toJson }}"
                tasks:
                  - id: reload_one
                    type: io.kestra.plugin.jdbc.clickhouse.Query
                    description: Reload one targeted dictionary. allowFailure so a reload that fails
                      again on an uncleared cause 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 RELOAD DICTIONARY `{{ inputs.database }}`.`{{
                      fromJson(item.value).name }}`"

                  - id: reload_log
                    type: io.kestra.plugin.core.log.Log
                    message: "Issued SYSTEM RELOAD DICTIONARY for {{ inputs.database }}.{{
                      fromJson(item.value).name }}."

              - id: reverify
                type: io.kestra.plugin.jdbc.clickhouse.Query
                description: Re-read system.dictionaries for exactly the dictionaries that were
                  reloaded. A successful reload clears last_exception and sets
                  status LOADED, so any row here means that dictionary is still
                  broken.
                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
                    name AS name,
                    toString(status) AS status,
                    element_count AS element_count,
                    replaceAll(replaceAll(replaceAll(substringUTF8(last_exception, 1, 180), '\\', '/'), '"', ''), '\n', ' ') AS last_exception_short
                  FROM system.dictionaries
                  WHERE database = '{{ inputs.database }}'
                    AND has(splitByChar(',', '{{ outputs.summarize.values.name_csv }}'), name)
                    AND (last_exception != '' OR toString(status) NOT IN ('LOADED', 'LOADED_AND_RELOADING'))
                  ORDER BY name

              - 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 }}
                      reloaded dictionary(s) in {{ inputs.database }} are still
                      not loading: {{ outputs.reverify.rows | jq('map(.name + "
                      [status " + .status + "] " + .last_exception_short) |
                      join(" || ")') | first }}. The reload re-ran the source
                      and hit the same problem, so the cause is still there. Fix
                      the source then run again.
                else:
                  - id: remediated
                    type: io.kestra.plugin.core.log.Log
                    message: "Reloaded and verified. {{ outputs.detect.size }} dictionary(s) were
                      failing before, {{ outputs.reverify.size }} after. Every
                      targeted dictionary is LOADED with no exception recorded."
            else:
              - id: not_approved
                type: io.kestra.plugin.core.log.Log
                message: "Approval not granted, so nothing was reloaded. {{ outputs.detect.size
                  }} dictionary(s) still failing in {{ inputs.database }}: {{
                  outputs.summarize.values.dict_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 }} dictionary(s)
              would be reloaded: {{ outputs.summarize.values.dict_list }}. No
              change made."
    else:
      - id: clean
        type: io.kestra.plugin.core.log.Log
        message: "Every dictionary in {{ inputs.database }} is loading cleanly (no
          last_exception recorded). Nothing to reload."

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Post to Slack on any failure, including a connection error or
      dictionaries still failing to load after a reload pass.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "ClickHouse failed dictionary guard FAILED for {{ inputs.database }} (execution {{ execution.id }}). A connection error, a reload error or dictionaries still carrying an exception after a reload pass. Check the run."}