HelmDriftWatch - Helm release drift detector

id: helmdriftwatch
namespace: company.team
description: Detects when a Helm release's live workloads drift from the chart's
  desired state (out-of-band kubectl edits) and alerts on Slack.

labels:
  project: helmdriftwatch
  team: platform

inputs:
  - id: release_name
    type: STRING
    displayName: "Helm release name"
    description: the helm release to watch for drift
  - id: namespace
    type: STRING
    displayName: "Namespace"
    description: the namespace holding the release
    defaults: default
  - id: slack_webhook
    type: STRING
    displayName: "Slack Webhook URL"
    description: Incoming webhook URL of the Slack channel for alerts

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: checks for drift every thirty minutes
    cron: "*/30 * * * *"
    timezone: UTC
    inputs:
      release_name: my-app
      namespace: default
      slack_webhook: "{{ secret('SLACK_WEBHOOK') }}"

tasks:
  - id: collect
    type: io.kestra.plugin.scripts.shell.Commands
    description: grabs the helm desired state and the live workload state
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
      user: "0"
    containerImage: dtzar/helm-kubectl
    inputFiles:
      client.crt: "{{ secret('CLIENT_CRT') }}"
      client.key: "{{ secret('CLIENT_KEY') }}"
    commands:
      - CRT_B64=$(base64 client.crt | tr -d '\n')
      - KEY_B64=$(base64 client.key | tr -d '\n')
      - |
        printf 'apiVersion: v1\nkind: Config\nclusters:\n- cluster:\n    server: %s\n    insecure-skip-tls-verify: true\n  name: target\nusers:\n- user:\n    client-certificate-data: %s\n    client-key-data: %s\n  name: target\ncontexts:\n- context:\n    cluster: target\n    user: target\n  name: target\ncurrent-context: target\n' "{{ secret('MASTERURL') }}" "$CRT_B64" "$KEY_B64" > kubeconfig
      - export KUBECONFIG=$PWD/kubeconfig
      - helm get values {{ inputs.release_name }} -n {{ inputs.namespace }} -o
        json > values.json
      - helm get manifest {{ inputs.release_name }} -n {{ inputs.namespace }} >
        manifest.yaml
      - kubectl get deploy,sts -n {{ inputs.namespace }} -l
        app.kubernetes.io/instance={{ inputs.release_name }} -o json > live.json
    outputFiles:
      - values.json
      - manifest.yaml
      - live.json

  - id: check_drift
    type: io.kestra.plugin.scripts.python.Script
    description: diffs live workloads against the chart's desired state
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
      user: "0"
    containerImage: python:3.11
    dependencies:
      - kestra
      - pyyaml
    inputFiles:
      values.json: "{{ outputs.collect.outputFiles['values.json'] }}"
      manifest.yaml: "{{ outputs.collect.outputFiles['manifest.yaml'] }}"
      live.json: "{{ outputs.collect.outputFiles['live.json'] }}"
    script: |
      import hashlib
      import json
      import yaml
      from kestra import Kestra

      release = """{{ inputs.release_name }}"""
      namespace = """{{ inputs.namespace }}"""

      with open('values.json') as f:
          values = json.load(f)
      with open('manifest.yaml') as f:
          manifest_docs = [d for d in yaml.safe_load_all(f) if d]
      with open('live.json') as f:
          live_items = json.load(f).get('items', [])

      def workload_state(obj):
          spec = obj['spec']
          containers = spec['template']['spec']['containers']
          return {
              'replicas': spec.get('replicas', 1),
              'images': {c['name']: c['image'] for c in containers},
              'env': {
                  c['name']: sorted((e['name'], e.get('value', '')) for e in c.get('env', []) or [])
                  for c in containers
              },
          }

      desired = {}
      for doc in manifest_docs:
          if doc.get('kind') in ('Deployment', 'StatefulSet'):
              desired[doc['metadata']['name']] = workload_state(doc)

      live = {}
      for item in live_items:
          if item.get('kind') in ('Deployment', 'StatefulSet'):
              live[item['metadata']['name']] = workload_state(item)

      drift = []
      for name, want in desired.items():
          got = live.get(name)
          if got is None:
              drift.append(f"{name}: in the chart but missing live (deleted out-of-band?)")
              continue
          if want['replicas'] != got['replicas']:
              drift.append(f"{name}: replicas live={got['replicas']} chart={want['replicas']}")
          for cname, image in want['images'].items():
              if got['images'].get(cname) != image:
                  drift.append(f"{name}/{cname}: image live={got['images'].get(cname)} chart={image}")
          for cname, env in want['env'].items():
              if got['env'].get(cname) != env:
                  drift.append(f"{name}/{cname}: env vars differ from the chart")
      for name in live:
          if name not in desired:
              drift.append(f"{name}: live but not in the chart (created out-of-band?)")

      # values snapshot: note chart-side changes, but they never alert alone.
      # a helm upgrade legitimately changes values; the drift that matters is live vs desired.
      raw_state = """{{ kv('helmdriftwatch-' ~ inputs.namespace ~ '-' ~ inputs.release_name, errorOnMissing=false) }}"""
      state = json.loads(raw_state) if raw_state.strip() else {}
      values_changed = state.get('values') != values
      last_sig = state.get('drift_sig', '')

      sig = hashlib.sha256(json.dumps(sorted(drift)).encode()).hexdigest() if drift else ''
      is_new = bool(drift) and sig != last_sig

      lines = [f"*HelmDriftWatch - drift in release '{release}' ({namespace})*", ""]
      for d in drift:
          lines.append(f"- {d}")
      if values_changed and state.get('values') is not None:
          lines.append("")
          lines.append("chart values also changed since the last run (helm upgrade or --set?).")
      lines.append("")
      lines.append("reconcile with: helm upgrade " + release + " <chart> -n " + namespace + " --reuse-values")
      alert_text = "\n".join(lines)

      Kestra.outputs({
          'drift': drift,
          'is_new': is_new,
          'alert_text': alert_text,
          'new_state_json': json.dumps({'values': values, 'drift_sig': sig}),
      })

  - id: maybe_alert
    type: io.kestra.plugin.core.flow.If
    description: alerts only on new drift
    condition: "{{ outputs.check_drift.vars.is_new }}"
    then:
      - id: send_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: posts the drift alert to Slack
        url: "{{ inputs.slack_webhook }}"
        messageText: "{{ outputs.check_drift.vars.alert_text }}"

  - id: save_state
    type: io.kestra.plugin.core.kv.Set
    description: remembers the values snapshot and the last drift signature
    key: "helmdriftwatch-{{ inputs.namespace }}-{{ inputs.release_name }}"
    value: "{{ outputs.check_drift.vars.new_state_json }}"
    overwrite: true