ClickHouse Detached Parts Cleanup Guard

id: clickhouse-detached-parts-cleanup-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 table. 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 detached parts are checked.

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

  - id: target_reasons
    type: STRING
    defaults: "broken,covered-by-broken,unexpected,ignored,clone"
    description: Comma separated system.detached_parts reasons that count as debris.
      These are parts a failed merge, a failed mutation or a bad startup left
      behind. They never rejoin a table on their own.

  - id: include_manual_detach
    type: BOOL
    defaults: false
    description: When false, a part with an empty reason (a plain ALTER TABLE DETACH
      PART) is never targeted, because an operator may be about to ATTACH it
      back. Set true to also target empty reason manual detaches.

  - id: staleness_minutes
    type: INT
    defaults: 60
    description: Grace window in minutes. A detached part is dropped only once this
      guard has seen it in system.detached_parts for at least this long. A part
      seen for the first time is reported but never dropped on the same run.

  - id: dry_run
    type: BOOL
    defaults: true
    description: When true, report the stale detached parts without dropping anything.

  - id: auto_remediate
    type: BOOL
    defaults: false
    description: When true and dry_run is false, drop the stale detached parts
      behind an approval gate, then re-read system.detached_parts.

variables:
  cursor_key: "detached_parts_firstseen_{{ inputs.database }}_{{ inputs.table ==
    '' ? 'all' : inputs.table }}"
  detect_sql: |
    SELECT
      database AS database,
      table AS table,
      name AS name,
      ifNull(reason, '') AS reason,
      disk AS disk,
      bytes_on_disk AS bytes_on_disk
    FROM system.detached_parts
    WHERE database = '{{ inputs.database }}'
      AND ('{{ inputs.table }}' = '' OR table = '{{ inputs.table }}')
      AND (
        has(splitByChar(',', '{{ inputs.target_reasons }}'), ifNull(reason, ''))
        OR ({{ inputs.include_manual_detach ? 1 : 0 }} = 1 AND ifNull(reason, '') = '')
      )
    ORDER BY name

triggers:
  - id: daily_detached_parts_check
    type: io.kestra.plugin.core.trigger.Schedule
    disabled: true
    cron: "0 3 * * *"
    description: Daily detached parts sweep. Disabled by default and pinned to safe
      mode so it reports without dropping until an operator enables it.
    inputs:
      dry_run: true
      auto_remediate: false

tasks:
  - id: now_probe
    type: io.kestra.plugin.jdbc.clickhouse.Query
    description: Read the server clock as epoch seconds so the first seen age is
      measured against the ClickHouse clock, not the Kestra host clock.
    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 toUnixTimestamp(now()) AS now_epoch"

  - id: detect
    type: io.kestra.plugin.jdbc.clickhouse.Query
    description: Read system.detached_parts for the database and keep the debris
      reasons. These parts sit in the detached directory, are never queried and
      waste disk until an operator removes them.
    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: read_cursor
    type: io.kestra.plugin.core.kv.Get
    description: Read the first seen cursor, a KV map from database::table::name to
      the epoch second this guard first saw the part. A missing key is the first
      run and returns null.
    key: "{{ render(vars.cursor_key) }}"
    errorOnMissing: false

  - id: compose
    type: io.kestra.plugin.core.output.OutputValues
    description: Build one JSON document holding the server clock, the grace window
      in seconds, the stored cursor and the detected parts, so every later step
      reads the same input.
    values:
      blob: >-
        {{ {'now': outputs.now_probe.rows[0].now_epoch, 'thr':
        (inputs.staleness_minutes * 60), 'old': (outputs.read_cursor.value ??
        ''), 'parts': outputs.detect.rows} | toJson }}

  - id: enrich
    type: io.kestra.plugin.core.output.OutputValues
    description: For each detected part, look up its first seen epoch (or set it to
      now on first sight), compute its age and flag it a target when the age is
      at or past the grace window.
    values:
      all_json: >-
        {{ outputs.compose.values.blob | jq('(.now|tonumber) as $N |
        (.thr|tonumber) as $T | (.old | if type == "string" then (if . == ""
        then "{}" else . end | fromjson) elif . == null then {} else . end) as
        $old | [ .parts[] | (.database+"::"+.table+"::"+.name) as $k |
        {database, table, name, reason, bytes:(.bytes_on_disk|tonumber), k:$k,
        first_seen:($old[$k] // $N)} | . + {age:($N - .first_seen)} | . +
        {is_target:(.age >= $T)} ]') | first | toJson }}

  - id: plan
    type: io.kestra.plugin.core.output.OutputValues
    description: Split the enriched list into the drop targets and the fresh parts,
      build the pruned cursor and the human readable lists once so the logs, the
      alert and the failure message all agree.
    values:
      targets_json: >-
        {{ outputs.enrich.values.all_json | jq('map(select(.is_target)) |
        map({database, table, name, reason, bytes, first_seen, age})') | first |
        toJson }}
      new_cursor_json: >-
        {{ outputs.enrich.values.all_json | jq('map({(.k): .first_seen}) | add
        // {}') | first | toJson }}
      targeted_names: >-
        {{ outputs.enrich.values.all_json | jq('map(select(.is_target)) |
        map(.database+"."+.table+"/"+.name+" ("+.reason+")") | join(", ")') |
        first }}
      targeted_bytes: >-
        {{ outputs.enrich.values.all_json | jq('map(select(.is_target) | .bytes)
        | add // 0') | first }}
      fresh_names: >-
        {{ outputs.enrich.values.all_json | jq('map(select(.is_target | not)) |
        map(.database+"."+.table+"/"+.name+" ("+.reason+", seen
        "+(.age|tostring)+"s ago)") | join(", ")') | first }}

  - id: write_cursor
    type: io.kestra.plugin.core.kv.Set
    description: Write the pruned first seen cursor back. It keeps the first seen
      epoch for parts still detached, records now for parts seen for the first
      time and drops entries for parts that are gone. No ttl, so a long lived
      part keeps its original first seen time.
    key: "{{ render(vars.cursor_key) }}"
    value: "{{ outputs.plan.values.new_cursor_json }}"

  - id: log_detect
    type: io.kestra.plugin.core.log.Log
    message: |
      ClickHouse {{ inputs.database }}: {{ outputs.detect.size }} debris detached part(s) found. Targets past the {{ inputs.staleness_minutes }} minute grace window: {{ outputs.plan.values.targets_json | jq('length') | first }} ({{ outputs.plan.values.targeted_bytes }} bytes).
      Targeted: {{ outputs.plan.values.targeted_names }}
      Within grace window (reported, not dropped): {{ outputs.plan.values.fresh_names }}
      Mode: dry_run={{ inputs.dry_run }} auto_remediate={{ inputs.auto_remediate }} include_manual_detach={{ inputs.include_manual_detach }}

  - 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 detached debris even when nothing is dropped.
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {"text": "ClickHouse detached parts guard: {{ outputs.detect.size }} debris part(s) in {{ inputs.database }}, {{ outputs.plan.values.targets_json | jq('length') | first }} past the {{ inputs.staleness_minutes }}m grace window ({{ outputs.plan.values.targeted_bytes }} bytes). Targeted {{ outputs.plan.values.targeted_names }}. 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 and
          (outputs.plan.values.targets_json | jq('length') | first) > 0 }}"
        then:
          - id: approval_gate
            type: io.kestra.plugin.core.flow.Pause
            description: Hold the run until a human approves the drop. The destructive step
              never starts on its own.
            pauseDuration: PT2H
            behavior: FAIL
            onResume:
              - id: approved
                type: BOOL
                defaults: false
                description: Set true to drop the listed detached parts. Anything else leaves
                  them in place.

          - id: apply
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.approval_gate.onResume.approved ?? false }}"
            then:
              - id: drop_detached_parts
                type: io.kestra.plugin.core.flow.Loop
                description: Remediation loop. One ALTER TABLE DROP DETACHED PART per target,
                  concurrencyLimit 1 so the statements run in turn and parallel
                  drops never race on the same table.
                concurrencyLimit: 1
                values: "{{ outputs.plan.values.targets_json }}"
                tasks:
                  - id: drop_one
                    type: io.kestra.plugin.jdbc.clickhouse.Query
                    description: Drop one detached part by its full detached name. The SETTINGS
                      clause carries allow_drop_detached, the ClickHouse safety
                      flag this statement needs. allowFailure so one blocked
                      drop does not stop the pass and the re-read stays 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: "ALTER TABLE {{ fromJson(item.value).database }}.{{
                      fromJson(item.value).table }} DROP DETACHED PART '{{
                      fromJson(item.value).name }}' SETTINGS allow_drop_detached
                      = 1"

                  - id: drop_log
                    type: io.kestra.plugin.core.log.Log
                    message: "Issued DROP DETACHED PART '{{ fromJson(item.value).name }}' on {{
                      fromJson(item.value).database }}.{{
                      fromJson(item.value).table }}."

              - id: reverify
                type: io.kestra.plugin.jdbc.clickhouse.Query
                description: Re-read system.detached_parts with the same filter. The parts still
                  here are compared against the drop targets, so a part left
                  inside its grace window does not count as a failure.
                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: verify_plan
                type: io.kestra.plugin.core.output.OutputValues
                description: Keep only the drop targets that are still present after the pass.
                  The fresh parts left in place are excluded by design.
                values:
                  still_json: >-
                    {{ {'targets': outputs.plan.values.targets_json,
                    'remaining': outputs.reverify.rows} | toJson |
                    jq('(.remaining | map(.database+"::"+.table+"::"+.name)) as
                    $rem | (.targets | if type == "string" then fromjson else .
                    end) | map((.database+"::"+.table+"::"+.name) as $k |
                    select($rem | index($k)))') | first | toJson }}
                  still_names: >-
                    {{ {'targets': outputs.plan.values.targets_json,
                    'remaining': outputs.reverify.rows} | toJson |
                    jq('(.remaining | map(.database+"::"+.table+"::"+.name)) as
                    $rem | (.targets | if type == "string" then fromjson else .
                    end) | map((.database+"::"+.table+"::"+.name) as $k |
                    select($rem | index($k)) | .database+"."+.table+"/"+.name) |
                    join(", ")') | first }}

              - id: verify_result
                type: io.kestra.plugin.core.flow.If
                condition: "{{ (outputs.verify_plan.values.still_json | jq('length') | first) >
                  0 }}"
                then:
                  - id: remediation_failed
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: >-
                      {{ outputs.verify_plan.values.still_json | jq('length') |
                      first }} targeted detached part(s) still present after the
                      drop pass in {{ inputs.database }}: {{
                      outputs.verify_plan.values.still_names }}. The connecting
                      user may lack ALTER privilege, allow_drop_detached may be
                      blocked or a new part landed in the detached directory
                      during the run.
                else:
                  - id: remediated
                    type: io.kestra.plugin.core.log.Log
                    message: "Dropped and verified. {{ outputs.plan.values.targets_json |
                      jq('length') | first }} target(s) before, 0 of them
                      remaining after."
            else:
              - id: not_approved
                type: io.kestra.plugin.core.log.Log
                message: "Approval not granted, so nothing was dropped. {{
                  outputs.plan.values.targets_json | jq('length') | first }}
                  stale detached part(s) still remain in {{ inputs.database }}."
        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.plan.values.targets_json |
              jq('length') | first }} stale detached part(s) would be dropped:
              {{ outputs.plan.values.targeted_names }}. No change made."
    else:
      - id: clean
        type: io.kestra.plugin.core.log.Log
        message: "No debris detached parts in {{ inputs.database }} for reasons {{
          inputs.target_reasons }}. Nothing to drop."

errors:
  - id: alert_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Post to Slack on any failure, including a connection error or
      targeted detached parts still present after a drop pass.
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {"text": "ClickHouse detached parts guard FAILED for {{ inputs.database }} (execution {{ execution.id }}). A connection error, a DROP error or targeted parts still present after a drop pass. Check the run."}