JobWatch: Kubernetes failed Job watcher blueprint for Kestra

id: jobwatch
namespace: company.team
description: Watches Kubernetes Jobs and alerts on Slack when one fails.

labels:
  project: jobwatch
  team: platform

inputs:
  - id: namespace
    type: STRING
    displayName: "Namespace"
    description: the namespace whose jobs to watch
    defaults: default
  - id: slack_webhook
    type: STRING
    displayName: "Slack Webhook URL"
    description: Incoming webhook URL of the Slack channel for the alerts

triggers:
  - id: schedule
    type: io.kestra.plugin.core.trigger.Schedule
    description: checks job health every morning
    cron: "0 8 * * *"
    timezone: UTC
    inputs:
      namespace: default
      slack_webhook: "{{ secret('SLACK_WEBHOOK') }}"

tasks:
  - id: list_jobs
    type: io.kestra.plugin.scripts.shell.Commands
    description: lists all jobs 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 jobs -n {{ inputs.namespace }} -o json
        --client-certificate=client.crt --client-key=client.key --server={{
        secret('MASTERURL') }} --insecure-skip-tls-verify=true > jobs.json
    outputFiles:
      - jobs.json

  - id: find_trouble
    type: io.kestra.plugin.scripts.python.Script
    description: finds failed jobs, drops ones already alerted on
    taskRunner:
      type: io.kestra.plugin.scripts.runner.docker.Docker
      user: "0"
    containerImage: python:3.11
    dependencies:
      - kestra
    script: |
      import json
      from kestra import Kestra

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

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

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

      troubled = []
      for job in jobs:
          name = job['metadata']['name']
          status = job.get('status', {})
          reasons = []

          failed = status.get('failed', 0) or 0
          if failed > 0:
              reasons.append(f"{failed} failed pod(s)")

          for cond in status.get('conditions', []):
              if cond.get('type') == 'Failed' and cond.get('status') == 'True':
                  reason = cond.get('reason', 'failed')
                  reasons.append(f"condition: {reason}")

          # dedupe the reason list while keeping order
          seen = set()
          reasons = [r for r in reasons if not (r in seen or seen.add(r))]

          if reasons:
              troubled.append({'name': name, 'reasons': reasons})

      troubled_names = set(t['name'] for t in troubled)
      new_trouble = [t for t in troubled if t['name'] not in already]
      # the new state is simply who is troubled right now;
      # deleted jobs drop out on their own, so a future failure alerts again

      lines = [f"*JobWatch - failed jobs in '{namespace}'*", ""]
      for t in new_trouble:
          lines.append(f"- *{t['name']}*: " + "; ".join(t['reasons']))
          lines.append(f"  inspect: `kubectl describe job {t['name']} -n {namespace}`")
      lines.append("")
      alert_text = "\n".join(lines)

      Kestra.outputs({
          'troubled': new_trouble,
          'count': len(new_trouble),
          'new_state_json': json.dumps(sorted(troubled_names)),
          'alert_text': alert_text,
      })

  - id: maybe_alert
    type: io.kestra.plugin.core.flow.If
    description: only alerts when there is something new to report
    condition: "{{ outputs.find_trouble.vars.count > 0 }}"
    then:
      - id: send_alert
        type: io.kestra.plugin.slack.notifications.SlackIncomingWebhook
        description: posts the failed jobs to Slack
        url: "{{ inputs.slack_webhook }}"
        messageText: "{{ outputs.find_trouble.vars.alert_text }}"

  - id: save_state
    type: io.kestra.plugin.core.kv.Set
    description: remembers who is troubled now, so deleted jobs alert again next time
    key: "jobwatch-state-{{ inputs.namespace }}"
    value: "{{ outputs.find_trouble.vars.new_state_json }}"
    overwrite: true

How it works

list_jobs (io.kestra.plugin.scripts.shell.Commands) runs kubectl get jobs in the target namespace and saves the full job list as JSON. It authenticates with a client certificate and key passed as Kestra secrets, the same pattern the Kubernetes gallery blueprints use.
find_trouble (io.kestra.plugin.scripts.python.Script) reads the remembered state from the KV store, then checks every job's status. Trouble is any failed pods, or a Failed condition being True. Anything already alerted on is dropped.
maybe_alert (io.kestra.plugin.core.flow.If) runs only when there is something new to report, and send_alert posts the failed jobs to Slack with the reasons and a kubectl command to inspect each one.
save_state (io.kestra.plugin.core.kv.Set) writes back the new list of troubled job names as one JSON value. Deleted jobs drop out on their own, so a future failure alerts again.

What you get

A daily job health check for any namespace, no agent to install inside the cluster.
Slack alerts only for new trouble: which job, why it failed, and the kubectl command to look at it.
No alert spam: the KV-backed memory remembers who was already reported, and forgets jobs that were deleted.
Quiet runs when every job succeeded.

Who it's for

Anyone running batch work on Kubernetes: migrations, backups, report generators, data loads. A failed job that nobody notices is a silent outage. Works on any cluster you can reach with a client certificate: minikube, kind, managed clusters.

Why orchestrate this with Kestra

A shell script on a cron server can list jobs, but then you need somewhere to keep the "already alerted" state, somewhere to hold the secrets, and a place to see the run history when an alert looks wrong. Kestra gives you all three: the KV store is the memory, secrets stay out of the YAML, and every run keeps its logs and outputs where you can inspect them.

Prerequisites

A Kubernetes cluster reachable from where Kestra runs, with a client certificate and key for authentication.
Kestra with the Docker task runner available (used for the kubectl and Python steps).
A Slack Incoming Webhook URL.

Secrets

Secret	What it is
`CLIENT_CRT`	base64-encoded client certificate for the cluster
`CLIENT_KEY`	base64-encoded client key for the cluster
`MASTERURL`	base64-encoded Kubernetes API server URL, reachable from inside the Kestra container
`SLACK_WEBHOOK`	base64-encoded Slack Incoming Webhook URL

In Kestra OSS, secrets come from environment variables prefixed with SECRET_ and must be base64-encoded.

Quick start

Start Kestra with the four secrets above as SECRET_ environment variables.
Import this flow and run it with the namespace you want to watch.
Create a test job that always fails (kubectl create job jw-fail --image=busybox -- sh -c 'exit 1') and run again: the Slack alert lists the job and the failure.
Run once more without changing anything: no second alert. Delete the job and run again: quiet, and the state is cleared.

How to extend

Add a failed-pod threshold so a job with one retry failure stays quiet.
Lower the schedule to every hour for faster detection by changing the cron trigger.
Deploy one copy of the flow per namespace you want to watch, changing only the namespace input.
Route alerts to PagerDuty or Opsgenie by swapping the Slack task for another notification task.

Inputs

Input	Type	Default	What it does
`namespace`	STRING	`default`	the namespace whose jobs to watch
`slack_webhook`	STRING	(required)	Incoming webhook URL of the Slack channel for the alerts

Pitfalls

Jobs are not cleaned up automatically by default, so a failed job stays visible until you delete it. The dedup means you hear about it once, then it stays quiet until the job is gone.
A job whose pods are still retrying (backoff) may not have failed set yet. The Failed condition with reason BackoffLimitExceeded catches the terminal state.
Make sure the MASTERURL is reachable from inside the Kestra container. localhost inside the container is not your host: use host.docker.internal when Kestra runs in Docker on the same machine as the cluster.