Reap Exited Docker Containers by Age and Label

id: docker-exited-container-reaper-guard
namespace: company.team

description: |
  Reap the terminal containers a CI runner or build host leaves behind. Only containers carrying
  a label you set yourself are read, only those in a terminal state (exited, created, dead) past
  an age threshold and absent from a protect-list become targets. Each one is removed with its
  anonymous volumes and never with --force, so a container that started running in the
  meantime is refused rather than killed. The host is then listed again. The run fails if any targeted
  container survived.

inputs:
  - id: docker_host
    type: STRING
    displayName: Docker host
    description: TCP endpoint of the daemon to clean, for example
      tcp://build-01:2375. TCP is required because the container list is read
      from the Engine API, which a unix socket does not expose to the worker.
    defaults: "tcp://docker-host:2375"

  - id: label_selector
    type: STRING
    displayName: Label selector
    description: Only containers carrying this label are read, in key=value form.
      Nothing outside the selector is ever inspected or removed, so a container
      the flow does not own cannot be a target.
    defaults: "kestra.reap=true"

  - id: target_states
    type: MULTISELECT
    displayName: Terminal states to reap
    description: Container states treated as terminal. Running and paused containers
      are never in this list.
    values:
      - exited
      - created
      - dead
    defaults:
      - exited
      - created
      - dead

  - id: protected_name_pattern
    type: STRING
    displayName: Protected name pattern
    description: A container whose name matches this regular expression is never
      removed, even when it carries the label and is old. The protect-list wins
      over every other rule.
    defaults: "^(prod|db|state|kestra)([-_.].*)?$"

  - id: min_age_minutes
    type: INT
    displayName: Minimum age in minutes
    description: A container is a target only when it stopped (or was created, for a
      never-started one) longer ago than this. Protects the container a build
      exited two minutes ago.
    defaults: 1440

  - id: max_removals
    type: INT
    displayName: Removal ceiling
    description: At most this many containers are removed in one run. The rest are
      reported and left for the next run, so a wrong selector cannot empty a
      host in one go.
    defaults: 25

  - id: auto_remediate
    type: BOOL
    displayName: Auto remediate
    description: When false the flow only reports. Must be true, together with
      dry_run false and the approval, for anything to be removed.
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: When true the plan is only logged. Must be false, together with
      auto_remediate, for anything to be removed.
    defaults: true

variables:
  api: "{{ inputs.docker_host | replace({'tcp://': 'http://'}) }}"

tasks:
  - id: list_terminal
    type: io.kestra.plugin.core.http.Request
    description: Read-only. Asks the Engine API for containers carrying the label
      selector and sitting in one of the terminal states. A container outside
      the selector is never returned, so it cannot be seen or touched.
    uri: "{{ render(vars.api) }}/containers/json"
    method: GET
    params:
      all: "true"
      filters: "{{ {'label': [inputs.label_selector], 'status': inputs.target_states}
        | toJson }}"

  - id: candidates
    type: io.kestra.plugin.core.output.OutputValues
    description: The candidate ids. The list endpoint carries no stop time, which is
      why each one is inspected next.
    values:
      ids: >-
        {{ ({'c': fromJson(outputs.list_terminal.body)} | toJson) | jq('
           [.c[] | {id: .Id, name: (.Names[0] | ltrimstr("/"))}] | sort_by(.name)') | first | toJson }}

  - id: inspect_each
    type: io.kestra.plugin.core.flow.Loop
    description: One read-only inspect per candidate. The stop time, the exit code
      and the anonymous volumes exist only here.
    values: "{{ outputs.candidates.values.ids }}"
    concurrencyLimit: 4
    outputs:
      - id: state
        type: STRING
        value: >-
          {{ outputs.inspect.body | jq('{id: .Id, name: (.Name | ltrimstr("/")),
          image: .Config.Image,
             state: .State.Status, running: .State.Running, exit_code: .State.ExitCode,
             finished_at: (.State.FinishedAt // ""), created: (.Created // ""),
             volumes: [.Mounts[]? | select(.Type == "volume" and ((.Name // "") | test("^[0-9a-f]{64}$"))) | .Name]}') | first | toJson }}
    tasks:
      - id: inspect
        type: io.kestra.plugin.core.http.Request
        uri: "{{ render(vars.api) }}/containers/{{ fromJson(item.value).id }}/json"
        method: GET

  - id: window
    type: io.kestra.plugin.core.output.OutputValues
    description: The instant a container must have stopped before to count as old.
      Frozen once, so detection and verification share one cutoff.
    values:
      cutoff: >-
        {{ now(timeZone='UTC') | dateAdd(-1 * inputs.min_age_minutes, 'MINUTES')
        | date("yyyy-MM-dd'T'HH:mm:ss", 'UTC') }}

  - id: assess
    type: io.kestra.plugin.core.output.OutputValues
    description: >-
      One row per candidate with its verdict. A never-started container has no
      stop time, so its creation time is used instead. The protect-list and the
      removal ceiling both win over REAP.
    values:
      rows: >-
        {{ ({'rows': ((outputs.inspect_each.outputs ?? []) |
        jq('map(.outputs.state | fromjson)') | first),
            'cut': outputs.window.values.cutoff, 're': inputs.protected_name_pattern, 'cap': inputs.max_removals} | toJson) | jq('
           .cut as $cut | .re as $re | .cap as $cap
           | [.rows[] | . + {ref: (if ((.finished_at // "") | startswith("0001")) then (.created // "") else (.finished_at // "") end)}]
           | map(. + {old: ((.ref[0:19]) < $cut),
                      decision: (if .running then "KEEP_RUNNING"
                        elif (.name | test($re)) then "KEEP_PROTECTED"
                        elif ((.ref[0:19]) < $cut | not) then "KEEP_FRESH"
                        else "REAP" end)})
           | sort_by(.ref)
           | ([.[] | select(.decision == "REAP")] | to_entries
              | map(.value + {decision: (if .key < $cap then "REAP" else "KEEP_OVER_CAP" end)})) as $reap
           | ([.[] | select(.decision != "REAP")] + $reap) | sort_by(.decision, .name)') | first | toJson }}
      targets: >-
        {{ ({'rows': ((outputs.inspect_each.outputs ?? []) |
        jq('map(.outputs.state | fromjson)') | first),
            'cut': outputs.window.values.cutoff, 're': inputs.protected_name_pattern, 'cap': inputs.max_removals} | toJson) | jq('
           .cut as $cut | .re as $re | .cap as $cap
           | [.rows[] | . + {ref: (if ((.finished_at // "") | startswith("0001")) then (.created // "") else (.finished_at // "") end)}]
           | [.[] | select((.running | not) and ((.name | test($re)) | not) and ((.ref[0:19]) < $cut))]
           | sort_by(.ref) | .[0:$cap]
           | map({id: .id, name: .name, state: .state, ref: .ref, volumes: .volumes})') | first | toJson }}

  - id: summary
    type: io.kestra.plugin.core.log.Log
    message: >-
      Terminal container reaper on {{ inputs.docker_host }}, label {{
      inputs.label_selector }}: {{ fromJson(outputs.assess.values.rows) | length
      }} candidates, {{ fromJson(outputs.assess.values.targets) | length }} to
      reap, cutoff {{ outputs.window.values.cutoff }}Z. {{
      fromJson(outputs.assess.values.rows) | jq('group_by(.decision) |
      map("\(.[0].decision) \(length)") | join(", ")') | first }}. Targets: {{
      fromJson(outputs.assess.values.targets) | jq('if length == 0 then "none"
      else map("\(.name) \(.state) since \(.ref)") | join("; ") end') | first
      }}.

  - id: route
    type: io.kestra.plugin.core.flow.If
    condition: "{{ fromJson(outputs.assess.values.targets) | length > 0 }}"
    then:
      - id: findings_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        url: "{{ secret('SLACK_WEBHOOK_URL') }}"
        payload: |
          {{ {'text': 'Docker terminal container reaper on ' ~ inputs.docker_host ~ ': ' ~ (fromJson(outputs.assess.values.targets) | length) ~ ' of ' ~ (fromJson(outputs.assess.values.rows) | length) ~ ' labelled candidates are terminal and older than ' ~ inputs.min_age_minutes ~ ' minutes (' ~ (fromJson(outputs.assess.values.targets) | jq('map("\(.name) \(.state)") | join(", ")') | first) ~ '). ' ~ (inputs.auto_remediate and not inputs.dry_run ? 'Removal waits for approval on this execution.' : 'Nothing removed (auto_remediate=' ~ inputs.auto_remediate ~ ', dry_run=' ~ inputs.dry_run ~ ').')} | toJson }}

      - id: remediate_gate
        type: io.kestra.plugin.core.flow.If
        description: A container is removed only when auto_remediate is true, dry_run is
          false and the pause below is approved.
        condition: "{{ inputs.auto_remediate and not inputs.dry_run }}"
        then:
          - id: hold
            type: io.kestra.plugin.core.flow.Pause
            description: Removing a container destroys its writable layer and its anonymous
              volumes. Read the plan in the summary above, then resume with
              approved true.
            onResume:
              - id: approved
                type: BOOL
                required: false
                description: Set true to remove the listed containers. Anything else leaves the
                  host untouched.

          - id: approval_gate
            type: io.kestra.plugin.core.flow.If
            condition: "{{ outputs.hold.onResume.approved ?? false }}"
            then:
              - id: remediate_reap
                type: io.kestra.plugin.core.flow.Loop
                description: One docker rm per target, with its anonymous volumes, never forced.
                  Sequential so a failure is attributable to one container.
                values: "{{ outputs.assess.values.targets }}"
                concurrencyLimit: 1
                tasks:
                  - id: reap_one
                    type: io.kestra.plugin.docker.cli.Rm
                    description: force is false on purpose. A container that started running between
                      detection and here is refused, not killed.
                    allowFailure: true
                    host: "{{ inputs.docker_host }}"
                    containerIds:
                      - "{{ fromJson(item.value).id }}"
                    removeVolumes: true
                    force: false

              - id: relist
                type: io.kestra.plugin.core.http.Request
                description: Every container carrying the label, in any state. The state filter
                  is deliberately dropped here, so a target that started running
                  still shows up as present instead of reading as removed.
                uri: "{{ render(vars.api) }}/containers/json"
                method: GET
                params:
                  all: "true"
                  filters: "{{ {'label': [inputs.label_selector]} | toJson }}"

              - id: result
                type: io.kestra.plugin.core.output.OutputValues
                description: Compare the plan with the host as it is now. A target still present
                  was not removed. A container the plan kept must still be
                  there.
                values:
                  r: >-
                    {{ ({'targets': fromJson(outputs.assess.values.targets),
                        'kept': (fromJson(outputs.assess.values.rows) | jq('[.[] | select(.decision != "REAP")]') | first),
                        'after': fromJson(outputs.relist.body)} | toJson) | jq('
                       ([.after[] | .Id]) as $a
                       | {removed: [.targets[] | select(([.id] - $a) | length > 0)],
                          survivors: [.targets[] | select(([.id] - $a) | length == 0)],
                          kept_missing: [.kept[] | select((([.id] - $a) | length > 0) and (.running | not))],
                          after_count: ($a | length)}') | first | toJson }}

              - id: verify
                type: io.kestra.plugin.core.flow.If
                description: The run fails when a target survived or when a container the plan
                  kept is gone. Reporting success without reading the host again
                  is what this gate exists to prevent.
                condition: "{{ fromJson(outputs.result.values.r).survivors | length > 0 or
                  fromJson(outputs.result.values.r).kept_missing | length > 0
                  }}"
                then:
                  - id: reap_incomplete
                    type: io.kestra.plugin.core.execution.Fail
                    errorMessage: >-
                      Reap did not land on {{ inputs.docker_host }}. Still
                      present after removal: {{
                      fromJson(outputs.result.values.r).survivors | jq('if
                      length == 0 then "none" else map(.name) | join(", ") end')
                      | first }} (a container that started running again is
                      refused by design, since force is false). Gone but kept by
                      the plan: {{
                      fromJson(outputs.result.values.r).kept_missing | jq('if
                      length == 0 then "none" else map(.name) | join(", ") end')
                      | first }}.
                else:
                  - id: reaped
                    type: io.kestra.plugin.core.log.Log
                    message: >-
                      Reaped {{ fromJson(outputs.result.values.r).removed |
                      length }} containers on {{ inputs.docker_host }} ({{
                      fromJson(outputs.result.values.r).removed | jq('if length
                      == 0 then "none" else map("\(.name) \(.state)") | join(",
                      ") end') | first }}) with {{
                      fromJson(outputs.result.values.r).removed | jq('[.[] |
                      .volumes | length] | add // 0') | first }} anonymous
                      volumes. {{ fromJson(outputs.result.values.r).after_count
                      }} labelled containers remain, every one the plan kept is
                      still there.
            else:
              - id: approval_declined
                type: io.kestra.plugin.core.log.Log
                message: "Approval not granted on execution {{ execution.id }}. Nothing was
                  removed on {{ inputs.docker_host }}."
        else:
          - id: report_only
            type: io.kestra.plugin.core.log.Log
            message: >-
              Report only, nothing removed (auto_remediate={{
              inputs.auto_remediate }}, dry_run={{ inputs.dry_run }}). Run with
              auto_remediate true and dry_run false, then approve the pause, to
              remove {{ fromJson(outputs.assess.values.targets) | jq('map(.name)
              | join(", ")') | first }}.
    else:
      - id: nothing_to_reap
        type: io.kestra.plugin.core.log.Log
        message: >-
          No labelled container on {{ inputs.docker_host }} is terminal and
          older than {{ inputs.min_age_minutes }} minutes. {{
          fromJson(outputs.assess.values.rows) | length }} candidates read, all
          kept.

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Docker terminal container reaper failed on ' ~ inputs.docker_host ~ ' (execution ' ~ execution.id ~ '). Either the daemon is unreachable or a targeted container survived the removal. Nothing was force removed. Check the execution logs.'} | toJson }}

triggers:
  - id: nightly
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled and in safe mode. Enabling it only reports until
      you change these inputs.
    cron: "20 3 * * *"
    disabled: true
    inputs:
      dry_run: true
      auto_remediate: false