ImageWatch - Kubernetes ImagePullBackOff watcher

id: imagewatch
namespace: company.team
description: Watches pods stuck in ImagePullBackOff or ErrImagePull and alerts
  on Slack with cause hints.

labels:
  project: imagewatch
  team: platform

inputs:
  - id: namespace
    type: STRING
    displayName: "Namespace"
    description: the namespace whose pods to watch
    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 stuck image pulls every five minutes
    cron: "*/5 * * * *"
    timezone: UTC
    inputs:
      namespace: default
      slack_webhook: "{{ secret('SLACK_WEBHOOK') }}"

tasks:
  - id: list_pods
    type: io.kestra.plugin.scripts.shell.Commands
    description: lists all 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 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:
      - pods.json

  - id: find_stuck
    type: io.kestra.plugin.scripts.python.Script
    description: finds pods stuck pulling images and guesses why
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
      user: "0"
    containerImage: python:3.11
    dependencies:
      - kestra
    script: |
      import json
      from datetime import datetime, timezone
      from kestra import Kestra

      namespace = """{{ inputs.namespace }}"""

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

      # names we already alerted on, from the KV store (null on first run -> empty)
      raw_state = """{{ kv('imagewatch-state-' ~ inputs.namespace, errorOnMissing=false) }}"""
      already = set(json.loads(raw_state) if raw_state.strip() else [])

      def hint(image, message, pull_secrets):
          msg = (message or '').lower()
          hints = []
          if 'toomanyrequests' in msg or 'rate limit' in msg:
              hints.append('registry rate limit - wait or use an authenticated pull')
          if 'unauthorized' in msg or 'denied' in msg or '401' in msg or '403' in msg:
              hints.append('registry auth failed - check imagePullSecrets')
          first = image.split('/')[0] if '/' in image else ''
          if '.' in first and not pull_secrets:
              hints.append('private registry with no imagePullSecrets on the pod')
          if not hints:
              hints.append('image has never been seen - check for typos in the name or tag')
          return '; '.join(hints)

      stuck = []
      now = datetime.now(timezone.utc)
      for pod in pods:
          meta = pod.get('metadata', {})
          spec = pod.get('spec', {})
          status = pod.get('status', {})
          name = meta['name']
          pull_secrets = spec.get('imagePullSecrets', [])
          created = datetime.fromisoformat(
              meta['creationTimestamp'].replace('Z', '+00:00')
          )
          age_min = int((now - created).total_seconds() // 60)
          for cs in (status.get('containerStatuses', []) or []) + (status.get('initContainerStatuses', []) or []):
              waiting = (cs.get('state', {}) or {}).get('waiting')
              if not waiting:
                  continue
              reason = waiting.get('reason', '')
              if reason not in ('ImagePullBackOff', 'ErrImagePull'):
                  continue
              image = cs.get('image', '')
              stuck.append({
                  'pod': name,
                  'image': image,
                  'reason': reason,
                  'age_min': age_min,
                  'hint': hint(image, waiting.get('message'), pull_secrets),
              })
              break  # one alert per pod is enough

      stuck_names = sorted(s['pod'] for s in stuck)
      new = [s for s in stuck if s['pod'] not in already]

      lines = [f"*ImageWatch - image pull stuck in '{namespace}'*", ""]
      for s in new:
          lines.append(f"- *{s['pod']}*: `{s['image']}` stuck in {s['reason']} for ~{s['age_min']}m")
          lines.append(f"  likely cause: {s['hint']}")
      lines.append("")
      lines.append(f"inspect with: kubectl describe pod {new[0]['pod']} -n {namespace}" if new else "")
      alert_text = "\n".join(lines)

      Kestra.outputs({
          'new_stuck': new,
          'count': len(new),
          'alert_text': alert_text,
          'new_state_json': json.dumps(stuck_names),
      })

  - id: maybe_alert
    type: io.kestra.plugin.core.flow.If
    description: alerts only about newly stuck pods
    condition: "{{ outputs.find_stuck.vars.count > 0 }}"
    then:
      - id: send_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: posts the stuck image pull alert to Slack
        url: "{{ inputs.slack_webhook }}"
        messageText: "{{ outputs.find_stuck.vars.alert_text }}"

  - id: save_state
    type: io.kestra.plugin.core.kv.Set
    description: remembers who is stuck now, so recovered pods alert again next time
    key: "imagewatch-state-{{ inputs.namespace }}"
    value: "{{ outputs.find_stuck.vars.new_state_json }}"
    overwrite: true