Docker Container Log Rotation and Disk Guard

id: docker-log-rotation-disk-guard
namespace: company.team

description: |
  Find the Docker containers whose json-file logs have no size limit, measure the real disk
  bytes each one holds under the Docker data root, compare them with the last run and report
  the free space left on that filesystem. Containers over the threshold can have their log
  files truncated behind a human approval, then the flow measures the files again and fails if
  the space did not come back.

inputs:
  - id: docker_host
    type: STRING
    displayName: Docker daemon API endpoint
    description: Daemon to inspect, reachable from the task containers, for example
      tcp://build-01:2375.
    defaults: "tcp://docker-host:2375"

  - id: docker_network
    type: STRING
    displayName: Network for the task containers
    description: Docker network the CLI task containers join, so docker_host
      resolves by name.
    defaults: "bridge"

  - id: docker_cli_image
    type: STRING
    displayName: Docker CLI image
    defaults: "docker:27-cli"

  - id: helper_image
    type: STRING
    displayName: Measuring helper image
    description: Small image started on the target daemon with the data root
      mounted, to read the real file sizes.
    defaults: "alpine:3.20"

  - id: log_size_warn_mib
    type: INT
    displayName: Log size threshold in MiB
    description: A container with no max-size whose log files already hold this much
      disk is urgent.
    defaults: 256

  - id: min_free_percent
    type: INT
    displayName: Minimum free space percent
    description: Free space on the Docker data root filesystem below this is
      reported as at risk.
    defaults: 15

  - id: recommended_max_size
    type: STRING
    displayName: Recommended max-size
    description: The value the report tells the operator to set, as Docker spells it.
    defaults: "100m"

  - id: recommended_max_file
    type: INT
    displayName: Recommended max-file
    defaults: 3

  - id: truncate_allowlist
    type: STRING
    displayName: Truncation allowlist
    description: Regex on the container name. Only a matching container can be
      truncated. The default matches nothing.
    defaults: "^$"

  - id: free_space_slack_mib
    type: INT
    displayName: Free space slack in MiB
    description: Other writers share the data root filesystem, so the re-check
      allows free space to end this much below where it started.
    defaults: 128

  - id: auto_remediate
    type: BOOL
    displayName: Allow truncation
    description: Must be true, with dry_run false and an approval, before any log
      file is touched.
    defaults: false

  - id: dry_run
    type: BOOL
    displayName: Dry run
    description: When true the flow only measures and reports.
    defaults: true

variables:
  baseline_key: "docker_log_bytes_{{ inputs.docker_host | replace({'tcp://': '',
    'unix://': '', 'https://': '', '/': '_', ':': '_', '.': '_', '-': '_'}) }}"

tasks:
  - id: prepare_helper
    type: io.kestra.plugin.docker.cli.Pull
    description: Make sure the measuring image is present on the target daemon
      before anything tries to run it.
    host: "{{ inputs.docker_host }}"
    image: "{{ inputs.helper_image }}"

  - id: measure
    type: io.kestra.plugin.scripts.shell.Commands
    description: >-
      Read the daemon default log driver and data root, the per container log
      driver with its max-size and max-file options, then start the helper on
      the same daemon with the data root mounted read-only to get the allocated
      KiB and the apparent bytes of every json-file log, plus df on that
      filesystem.
    containerImage: "{{ inputs.docker_cli_image }}"
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
      networkMode: "{{ inputs.docker_network }}"
    env:
      DOCKER_HOST: "{{ inputs.docker_host }}"
      HELPER_IMAGE: "{{ inputs.helper_image }}"
    commands:
      - |
        set -e
        command -v jq >/dev/null 2>&1 || apk add --no-cache jq >/dev/null 2>&1
        INFO=$(docker info --format json)
        DR=$(echo "$INFO" | jq -r '.DockerRootDir')
        LD=$(echo "$INFO" | jq -r '.LoggingDriver')
        IDS=$(docker ps -aq)
        if [ -z "$IDS" ]; then echo "no containers on $DOCKER_HOST"; exit 1; fi
        docker inspect $IDS | jq -c '[ .[] | { id: .Id, name: (.Name | ltrimstr("/")), image: .Config.Image, state: .State.Status, log_type: .HostConfig.LogConfig.Type, max_size: (.HostConfig.LogConfig.Config["max-size"] // ""), max_file: (.HostConfig.LogConfig.Config["max-file"] // "") } ]' > containers.json
        docker run --rm -v "$DR":/hostroot:ro "$HELPER_IMAGE" sh -c 'find /hostroot/containers -maxdepth 2 -name "*-json.log*" -exec du -k {} \; ; echo "#SPLIT"; find /hostroot/containers -maxdepth 2 -name "*-json.log*" -exec stat -c "%s %n" {} \; ; echo "#SPLIT"; df -Pk /hostroot | tail -1' > raw.txt
        awk 'BEGIN { s=0 } /^#SPLIT$/ { s++; next } s==0 && NF>=2 { print "{ \"path\": \"" $2 "\", \"kib\": " $1 " }" } s==1 && NF>=2 { print "{ \"path\": \"" $2 "\", \"apparent\": " $1 " }" } s==2 && NF>=6 { print "{ \"blocks_kib\": " $2 ", \"used_kib\": " $3 ", \"avail_kib\": " $4 " }" }' raw.txt > rows.jsonl
        jq -s -c '{ files: [ .[] | select(.kib != null) ], apparent: [ .[] | select(.apparent != null) ], disk: ( [ .[] | select(.avail_kib != null) ] | first ) }' rows.jsonl > measured.json
        jq -c -n --slurpfile c containers.json --slurpfile m measured.json --arg root "$DR" --arg driver "$LD" '{ root: $root, daemon_default_driver: $driver, containers: $c[0], files: $m[0].files, apparent: $m[0].apparent, disk: $m[0].disk }' > report.json
        echo "data root $DR on $DOCKER_HOST, daemon default driver $LD, $(jq '.containers | length' report.json) containers, $(jq '.files | length' report.json) json-file logs"
        echo "::$(jq -c -n --rawfile r report.json '{ outputs: { report: $r } }')::"

  - id: assess
    type: io.kestra.plugin.core.output.OutputValues
    description: >-
      Attribute every log file to its container by id, classify the container,
      rank by disk held and compare each one with the baseline stored by the
      last run.
    values:
      r: >-
        {{ {'r': fromJson(outputs.measure.vars.report), 'warn':
        inputs.log_size_warn_mib, 'minfree': inputs.min_free_percent,
            'allow': inputs.truncate_allowlist, 'base': (kv(render(vars.baseline_key), errorOnMissing=false) ?? {})} | jq('
           .warn as $warn | .minfree as $minfree | .allow as $allow | .base as $base | .r as $r
           | [ $r.containers[] | . as $c | .id as $id
               | ([ $r.files[] | select(.path | contains($id)) ]) as $f
               | (($f | map(.kib) | add) // 0) as $kib
               | ((([ $r.apparent[] | select(.path | contains($id)) ] | map(.apparent) | add) // 0)) as $bytes
               | ($base[$c.name] // null) as $prev
               | { name: $c.name, short_id: $id[0:12], image: $c.image, state: $c.state,
                   log_type: $c.log_type, max_size: $c.max_size, max_file: $c.max_file,
                   log_dir: ($r.root + "/containers/" + $id), files: ($f | length), kib: $kib, bytes: $bytes,
                   prev_kib: $prev, growth_kib: (if $prev == null then null else ($kib - ($prev | tonumber)) end),
                   verdict: (if $c.log_type != "json-file" then "OTHER_DRIVER"
                             elif ($c.max_size | length) > 0 then "OK"
                             elif $kib >= ($warn * 1024) then "UNBOUNDED_AND_LARGE"
                             else "UNBOUNDED" end) } ]
           | sort_by(-.kib) as $cs
           | $r.disk as $d
           | (($d.avail_kib * 100 / $d.blocks_kib) | floor) as $freepct
           | [ $cs[] | select(.verdict == "UNBOUNDED_AND_LARGE") ] as $off
           | { root: $r.root, daemon_default_driver: $r.daemon_default_driver, containers: $cs,
               worst: ($cs | first), offenders: $off,
               targets: [ $off[] | select(.name | test($allow)) ],
               skipped: [ $off[] | select(.name | test($allow) | not) | .name ],
               latent: [ $cs[] | select(.verdict == "UNBOUNDED") | .name ],
               grower: ($cs | map(select(.growth_kib != null)) | sort_by(-.growth_kib) | first),
               attributed_kib: (($cs | map(.kib) | add) // 0),
               total_log_kib: (($r.files | map(.kib) | add) // 0),
               disk: ($d + { free_pct: $freepct, at_risk: ($freepct < $minfree) }),
               state: (if (($off | length) > 0) or ($freepct < $minfree) then "ACTION"
                       elif (([ $cs[] | select(.verdict == "UNBOUNDED") ] | length) > 0) then "LATENT"
                       else "OK" end) }') | first | toJson }}

  - id: report
    type: io.kestra.plugin.core.log.Log
    description: One line per container, worst first, plus the exact options to set
      and the honest note about what truncation can and cannot fix.
    message: >-
      Docker log guard on {{ inputs.docker_host }}: state {{
      fromJson(outputs.assess.values.r).state }}, daemon default driver {{
      fromJson(outputs.assess.values.r).daemon_default_driver }}, data root {{
      fromJson(outputs.assess.values.r).root }}. json-file logs hold {{
      fromJson(outputs.assess.values.r).total_log_kib | jq('. / 1024 | floor |
      tostring | sub("\\.0$"; "")') | first }} MiB, free space {{
      fromJson(outputs.assess.values.r).disk.free_pct | jq('tostring |
      sub("\\.0$"; "")') | first }}% ({{
      fromJson(outputs.assess.values.r).disk.avail_kib | jq('. / 1048576 | floor
      | tostring | sub("\\.0$"; "")') | first }} GiB free, threshold {{
      inputs.min_free_percent }}%, at risk {{
      fromJson(outputs.assess.values.r).disk.at_risk }}). Worst offender {{
      fromJson(outputs.assess.values.r).worst.name }} ({{
      fromJson(outputs.assess.values.r).worst.kib | jq('. / 1024 | floor |
      tostring | sub("\\.0$"; "")') | first }} MiB, {{
      fromJson(outputs.assess.values.r).worst.files }} files, verdict {{
      fromJson(outputs.assess.values.r).worst.verdict }}). Per container: {{
      fromJson(outputs.assess.values.r).containers | jq('map("\(.name)
      \(.verdict) \((.kib / 1024) | floor | tostring | sub("\\.0$"; ""))MiB" +
      (if .max_size == "" then "" else " cap \(.max_size)x\(.max_file)" end) +
      (if .growth_kib == null then " no baseline" else " growth \((.growth_kib /
      1024) | floor | tostring | sub("\\.0$"; ""))MiB" end)) | join("; ")') |
      first }}. Unbounded but under the threshold: {{
      fromJson(outputs.assess.values.r).latent | jq('if length == 0 then "none"
      else join(", ") end') | first }}. Fix: recreate each unbounded container
      with --log-opt max-size={{ inputs.recommended_max_size }} --log-opt
      max-file={{ inputs.recommended_max_file }}. Set the same log-opts as the
      daemon default so new containers inherit them. Changing a log driver
      option needs the container recreated, so this flow never does it.
      Truncation only buys time.

  - id: findings_alert
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    description: Only an urgent container or a filesystem running out of space is
      worth a message.
    runIf: "{{ fromJson(outputs.assess.values.r).state == 'ACTION' }}"
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Docker log guard on ' ~ inputs.docker_host ~ ': ' ~ (fromJson(outputs.assess.values.r).offenders | length) ~ ' containers log without a size limit and are already large ('
          ~ (fromJson(outputs.assess.values.r).offenders | jq('if length == 0 then "none" else map("\(.name) \((.kib / 1024) | floor | tostring | sub("\\.0$"; ""))MiB") | join(", ") end') | first)
          ~ '). Free space on ' ~ fromJson(outputs.assess.values.r).root ~ ' is ' ~ (fromJson(outputs.assess.values.r).disk.free_pct | jq('tostring | sub("\\.0$"; "")') | first)
          ~ '%. Recreate them with --log-opt max-size=' ~ inputs.recommended_max_size ~ ' --log-opt max-file=' ~ inputs.recommended_max_file ~ '.'} | toJson }}

  - id: apply_gate
    type: io.kestra.plugin.core.flow.If
    description: Truncation needs auto_remediate true, dry_run false and at least
      one offender the allowlist names.
    condition: "{{ inputs.auto_remediate and not inputs.dry_run and
      (fromJson(outputs.assess.values.r).targets | length) > 0 }}"
    then:
      - id: approve_truncation
        type: io.kestra.plugin.core.flow.Pause
        description: Truncating a log discards lines nobody shipped anywhere else, so a
          human has to say yes.
        pauseDuration: PT2H
        behavior: FAIL
        onResume:
          - id: approved
            type: BOOL
            defaults: false
            description: Set to true to truncate the log files listed in the plan.
          - id: reason
            type: STRING
            defaults: "disk pressure"
            description: Recorded in the execution so the decision is auditable.

      - id: truncation
        type: io.kestra.plugin.core.flow.If
        condition: "{{ (outputs.approve_truncation.onResume is defined) and
          outputs.approve_truncation.onResume.approved }}"
        then:
          - id: remediate_truncate
            type: io.kestra.plugin.core.flow.Loop
            description: One container per iteration, strictly in order, so two truncations
              never write the same directory at once.
            values: "{{ fromJson(outputs.assess.values.r).targets | toJson }}"
            concurrencyLimit: 1
            tasks:
              - id: truncate_log_files
                type: io.kestra.plugin.scripts.shell.Commands
                description: Mount only this container's log directory on the target daemon and
                  truncate its json-file logs in place.
                containerImage: "{{ inputs.docker_cli_image }}"
                taskRunner:
                  type: io.kestra.plugin.scripts.runner.docker.Docker
                  networkMode: "{{ inputs.docker_network }}"
                env:
                  DOCKER_HOST: "{{ inputs.docker_host }}"
                  HELPER_IMAGE: "{{ inputs.helper_image }}"
                  LOG_DIR: "{{ fromJson(item.value).log_dir }}"
                  TARGET: "{{ fromJson(item.value).name }}"
                commands:
                  - |
                    set -e
                    echo "truncating logs of $TARGET in $LOG_DIR on $DOCKER_HOST"
                    docker run --rm -v "$LOG_DIR":/t "$HELPER_IMAGE" sh -c 'for f in /t/*-json.log*; do if [ -f "$f" ]; then before=$(du -k "$f" | cut -f1); : > "$f"; echo "truncated $f, was ${before} KiB"; fi; done'

          - id: measure_after
            type: io.kestra.plugin.scripts.shell.Commands
            description: Read the daemon again, exactly as before, so the result is measured
              and not assumed.
            containerImage: "{{ inputs.docker_cli_image }}"
            taskRunner:
              type: io.kestra.plugin.scripts.runner.docker.Docker
              networkMode: "{{ inputs.docker_network }}"
            env:
              DOCKER_HOST: "{{ inputs.docker_host }}"
              HELPER_IMAGE: "{{ inputs.helper_image }}"
            commands:
              - |
                set -e
                command -v jq >/dev/null 2>&1 || apk add --no-cache jq >/dev/null 2>&1
                INFO=$(docker info --format json)
                DR=$(echo "$INFO" | jq -r '.DockerRootDir')
                LD=$(echo "$INFO" | jq -r '.LoggingDriver')
                IDS=$(docker ps -aq)
                if [ -z "$IDS" ]; then echo "no containers on $DOCKER_HOST"; exit 1; fi
                docker inspect $IDS | jq -c '[ .[] | { id: .Id, name: (.Name | ltrimstr("/")), image: .Config.Image, state: .State.Status, log_type: .HostConfig.LogConfig.Type, max_size: (.HostConfig.LogConfig.Config["max-size"] // ""), max_file: (.HostConfig.LogConfig.Config["max-file"] // "") } ]' > containers.json
                docker run --rm -v "$DR":/hostroot:ro "$HELPER_IMAGE" sh -c 'find /hostroot/containers -maxdepth 2 -name "*-json.log*" -exec du -k {} \; ; echo "#SPLIT"; find /hostroot/containers -maxdepth 2 -name "*-json.log*" -exec stat -c "%s %n" {} \; ; echo "#SPLIT"; df -Pk /hostroot | tail -1' > raw.txt
                awk 'BEGIN { s=0 } /^#SPLIT$/ { s++; next } s==0 && NF>=2 { print "{ \"path\": \"" $2 "\", \"kib\": " $1 " }" } s==1 && NF>=2 { print "{ \"path\": \"" $2 "\", \"apparent\": " $1 " }" } s==2 && NF>=6 { print "{ \"blocks_kib\": " $2 ", \"used_kib\": " $3 ", \"avail_kib\": " $4 " }" }' raw.txt > rows.jsonl
                jq -s -c '{ files: [ .[] | select(.kib != null) ], apparent: [ .[] | select(.apparent != null) ], disk: ( [ .[] | select(.avail_kib != null) ] | first ) }' rows.jsonl > measured.json
                jq -c -n --slurpfile c containers.json --slurpfile m measured.json --arg root "$DR" --arg driver "$LD" '{ root: $root, daemon_default_driver: $driver, containers: $c[0], files: $m[0].files, apparent: $m[0].apparent, disk: $m[0].disk }' > report.json
                echo "re-measured $(jq '.files | length' report.json) json-file logs under $DR"
                echo "::$(jq -c -n --rawfile r report.json '{ outputs: { report: $r } }')::"

          - id: compare
            type: io.kestra.plugin.core.output.OutputValues
            description: >-
              Per targeted container, the allocated KiB before and after, plus
              free space before and after. Free space is allowed to end
              free_space_slack_mib below the start, because other processes
              write to the same filesystem while the flow runs.
            values:
              c: >-
                {{ {'b': fromJson(outputs.assess.values.r), 'a':
                fromJson(outputs.measure_after.vars.report), 'slack':
                inputs.free_space_slack_mib} | jq('
                   .slack as $slack | .b as $b | .a as $a
                   | [ $b.targets[] | .name as $n | .short_id as $sid
                       | (($a.files | map(select(.path | contains($sid))) | map(.kib) | add) // 0) as $after
                       | { name: $n, before_kib: .kib, after_kib: $after, freed_kib: (.kib - $after), shrank: ($after < .kib) } ] as $t
                   | { targets: $t, freed_kib: (($t | map(.freed_kib) | add) // 0),
                       not_shrunk: [ $t[] | select(.shrank | not) | .name ],
                       avail_before_kib: $b.disk.avail_kib, avail_after_kib: $a.disk.avail_kib,
                       space_recovered: ($a.disk.avail_kib >= ($b.disk.avail_kib - ($slack * 1024))),
                       total_log_kib_before: $b.total_log_kib, total_log_kib_after: (($a.files | map(.kib) | add) // 0) }
                   | . + { ok: ((.not_shrunk | length) == 0 and .space_recovered) }') | first | toJson }}

          - id: verify
            type: io.kestra.plugin.core.flow.If
            description: A truncation that freed no bytes must fail the run rather than read
              as fixed. So must a filesystem that is still losing space.
            condition: "{{ not fromJson(outputs.compare.values.c).ok }}"
            then:
              - id: truncation_did_not_land
                type: io.kestra.plugin.core.execution.Fail
                errorMessage: >-
                  Truncation did not land on {{ inputs.docker_host }}.
                  Containers whose logs did not shrink: {{
                  fromJson(outputs.compare.values.c).not_shrunk | jq('if length
                  == 0 then "none" else join(", ") end') | first }}. Free space
                  recovered: {{
                  fromJson(outputs.compare.values.c).space_recovered }} ({{
                  fromJson(outputs.compare.values.c).avail_before_kib }} KiB
                  before, {{ fromJson(outputs.compare.values.c).avail_after_kib
                  }} KiB after, slack {{ inputs.free_space_slack_mib }} MiB).
                  Check whether the daemon holds the file open through another
                  driver or whether something else fills the filesystem faster
                  than the flow frees it.
            else:
              - id: reclaimed
                type: io.kestra.plugin.core.log.Log
                message: >-
                  Truncated {{ fromJson(outputs.compare.values.c).targets |
                  length }} container logs on {{ inputs.docker_host }}, freeing
                  {{ fromJson(outputs.compare.values.c).freed_kib | jq('. / 1024
                  | floor | tostring | sub("\\.0$"; "")') | first }} MiB. Per
                  container: {{ fromJson(outputs.compare.values.c).targets |
                  jq('map("\(.name) \(.before_kib) KiB to \(.after_kib) KiB") |
                  join("; ")') | first }}. All json-file logs under the data
                  root went from {{
                  fromJson(outputs.compare.values.c).total_log_kib_before }} KiB
                  to {{ fromJson(outputs.compare.values.c).total_log_kib_after
                  }} KiB. Free space {{
                  fromJson(outputs.compare.values.c).avail_before_kib }} KiB
                  before, {{ fromJson(outputs.compare.values.c).avail_after_kib
                  }} KiB after. Approved by a human with reason "{{
                  outputs.approve_truncation.onResume.reason }}". The containers
                  still have no max-size, so recreate them with --log-opt
                  max-size={{ inputs.recommended_max_size }} --log-opt
                  max-file={{ inputs.recommended_max_file }} to fix this for
                  good.

              - id: save_baseline_after
                type: io.kestra.plugin.core.kv.Set
                description: The baseline the next run compares growth against, taken after the
                  truncation so growth means new logs.
                key: "{{ render(vars.baseline_key) }}"
                kvType: JSON
                overwrite: true
                value: "{{ fromJson(outputs.measure_after.vars.report) | jq('. as $r | [
                  $r.containers[] | .id as $id | .name as $n | { ($n): ((([
                  $r.files[] | select(.path | contains($id)) ] | map(.kib) |
                  add)) // 0) } ] | add // {}') | first | toJson }}"
        else:
          - id: not_approved
            type: io.kestra.plugin.core.log.Log
            message: "Truncation on {{ inputs.docker_host }} was not approved, so no log
              file was touched."
    else:
      - id: no_change
        type: io.kestra.plugin.core.log.Log
        message: >-
          No log file was touched on {{ inputs.docker_host }}. dry_run {{
          inputs.dry_run }}, auto_remediate {{ inputs.auto_remediate }}, urgent
          containers {{ fromJson(outputs.assess.values.r).offenders | length }},
          allowed by truncate_allowlist {{
          fromJson(outputs.assess.values.r).targets | length }}, urgent but not
          in the allowlist: {{ fromJson(outputs.assess.values.r).skipped |
          jq('if length == 0 then "none" else join(", ") end') | first }}.

      - id: save_baseline
        type: io.kestra.plugin.core.kv.Set
        description: Store the measured KiB per container so the next run can report
          growth. No ttl, the key has to survive.
        key: "{{ render(vars.baseline_key) }}"
        kvType: JSON
        overwrite: true
        value: "{{ fromJson(outputs.assess.values.r).containers | jq('map({ (.name):
          .kib }) | add // {}') | first | toJson }}"

errors:
  - id: alert_flow_failure
    type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
    url: "{{ secret('SLACK_WEBHOOK_URL') }}"
    payload: |
      {{ {'text': 'Docker log rotation guard failed on ' ~ inputs.docker_host ~ ' (execution ' ~ execution.id ~ '). Container log growth is unmeasured until this run succeeds.'} | toJson }}

triggers:
  - id: hourly
    type: io.kestra.plugin.core.trigger.Schedule
    description: Shipped disabled and in safe mode. Enable it once the thresholds
      match your hosts.
    cron: "20 * * * *"
    disabled: true
    inputs:
      dry_run: true
      auto_remediate: false