ConfigJanitor - Kubernetes unused ConfigMap and Secret cleaner

id: configjanitor
namespace: company.team
description: Finds ConfigMaps and Secrets no pod references, deletes them, and
  reports the cleanup on Slack.

labels:
  project: configjanitor
  team: platform

inputs:
  - id: namespace
    type: STRING
    displayName: "Namespace"
    description: the namespace whose configs to clean
    defaults: default
  - id: dry_run
    type: BOOL
    displayName: "Dry run"
    description: when true, only reports what would be deleted without deleting anything
    defaults: true
  - id: slack_webhook
    type: STRING
    displayName: "Slack Webhook URL"
    description: Incoming webhook URL of the Slack channel for the cleanup report

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: cleans up unused configs every Sunday morning
    cron: "0 8 * * 0"
    timezone: UTC
    inputs:
      namespace: default
      dry_run: true
      slack_webhook: "{{ secret('SLACK_WEBHOOK') }}"

tasks:
  - id: list_resources
    type: io.kestra.plugin.scripts.shell.Commands
    description: lists configmaps, secrets, and pods in the namespace as JSON
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
      user: "0"
    containerImage: bitnami/kubectl
    inputFiles:
      client.crt: "{{ secret('CLIENT_CRT') }}"
      client.key: "{{ secret('CLIENT_KEY') }}"
    commands:
      - kubectl get configmaps -n {{ inputs.namespace }} -o json
        --client-certificate=client.crt --client-key=client.key --server={{
        secret('MASTERURL') }} --insecure-skip-tls-verify=true > configmaps.json
      - kubectl get secrets -n {{ inputs.namespace }} -o json
        --client-certificate=client.crt --client-key=client.key --server={{
        secret('MASTERURL') }} --insecure-skip-tls-verify=true > secrets.json
      - kubectl get pods -n {{ inputs.namespace }} -o json
        --client-certificate=client.crt --client-key=client.key --server={{
        secret('MASTERURL') }} --insecure-skip-tls-verify=true > pods.json
    outputFiles:
      - configmaps.json
      - secrets.json
      - pods.json

  - id: find_unused
    type: io.kestra.plugin.scripts.python.Script
    description: finds configs no pod references through env, volumes, or image pull secrets
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
      user: "0"
    containerImage: python:3.11
    dependencies:
      - kestra
    outputFiles:
      - unused.txt
    script: |
      import json
      from kestra import Kestra

      namespace = """{{ inputs.namespace }}"""
      dry_run = """{{ inputs.dry_run }}""".strip().lower() == "true"

      with open("""{{ outputs.list_resources.outputFiles['configmaps.json'] }}""") as f:
          configmaps = json.load(f)['items']
      with open("""{{ outputs.list_resources.outputFiles['secrets.json'] }}""") as f:
          secrets = json.load(f)['items']
      with open("""{{ outputs.list_resources.outputFiles['pods.json'] }}""") as f:
          pods = json.load(f)['items']

      # every (kind, name) pair a pod actually references
      referenced = set()
      for pod in pods:
          spec = pod.get('spec', {})
          for container in spec.get('containers', []) + spec.get('initContainers', []):
              for ef in container.get('envFrom', []) or []:
                  if 'configMapRef' in ef:
                      referenced.add(('configmap', ef['configMapRef']['name']))
                  if 'secretRef' in ef:
                      referenced.add(('secret', ef['secretRef']['name']))
              for e in container.get('env', []) or []:
                  vf = e.get('valueFrom', {})
                  if 'configMapKeyRef' in vf:
                      referenced.add(('configmap', vf['configMapKeyRef']['name']))
                  if 'secretKeyRef' in vf:
                      referenced.add(('secret', vf['secretKeyRef']['name']))
          for vol in spec.get('volumes', []) or []:
              if 'configMap' in vol:
                  referenced.add(('configmap', vol['configMap']['name']))
              if 'secret' in vol:
                  referenced.add(('secret', vol['secret']['name']))
              if 'projected' in vol:
                  for src in vol['projected'].get('sources', []) or []:
                      if 'configMap' in src:
                          referenced.add(('configmap', src['configMap']['name']))
                      if 'secret' in src:
                          referenced.add(('secret', src['secret']['name']))
          for ips in spec.get('imagePullSecrets', []) or []:
              referenced.add(('secret', ips['name']))

      unused = []
      for cm in configmaps:
          name = cm['metadata']['name']
          if ('configmap', name) not in referenced:
              unused.append({'kind': 'configmap', 'name': name})
      for secret in secrets:
          name = secret['metadata']['name']
          if ('secret', name) not in referenced:
              unused.append({'kind': 'secret', 'name': name})

      # one kind/name per line for the delete step to read
      with open('unused.txt', 'w') as f:
          for u in unused:
              f.write(f"{u['kind']}/{u['name']}\n")

      if dry_run:
          title = f"*ConfigJanitor - dry run in '{namespace}'*"
          tail = "would delete {n} unused config(s). set dry_run to false to actually clean."
      else:
          title = f"*ConfigJanitor - cleaned '{namespace}'*"
          tail = "deleted {n} unused config(s)."
      lines = [title, ""]
      for u in sorted(unused, key=lambda x: (x['kind'], x['name'])):
          lines.append(f"- *{u['kind']}/{u['name']}*: not referenced by any pod")
      lines.append("")
      lines.append(tail.format(n=len(unused)))
      summary_text = "\n".join(lines)

      Kestra.outputs({
          'unused': unused,
          'count': len(unused),
          'summary_text': summary_text,
      })

  - id: maybe_clean
    type: io.kestra.plugin.core.flow.If
    description: only acts when there is something unused to report
    condition: "{{ outputs.find_unused.vars.count > 0 }}"
    then:
      - id: do_clean
        type: io.kestra.plugin.core.flow.If
        description: deletes the unused configs, live mode only
        condition: "{{ inputs.dry_run == false }}"
        then:
          - id: delete_unused
            type: io.kestra.plugin.scripts.shell.Commands
            description: deletes each unused configmap and secret
            taskRunner:
              type: io.kestra.plugin.scripts.runner.docker.Docker
              user: "0"
            containerImage: bitnami/kubectl
            inputFiles:
              client.crt: "{{ secret('CLIENT_CRT') }}"
              client.key: "{{ secret('CLIENT_KEY') }}"
            commands:
              - |
                while read -r line; do
                  if [ -n "$line" ]; then
                    kind=$(echo "$line" | cut -d/ -f1)
                    name=$(echo "$line" | cut -d/ -f2-)
                    kubectl delete "$kind" "$name" -n {{ inputs.namespace }} --client-certificate=client.crt --client-key=client.key --server={{ secret('MASTERURL') }} --insecure-skip-tls-verify=true
                  fi
                done < {{ outputs.find_unused.outputFiles['unused.txt'] }}
      - id: send_summary
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: posts the cleanup summary to Slack
        url: "{{ inputs.slack_webhook }}"
        messageText: "{{ outputs.find_unused.vars.summary_text }}"