Crony

id: crony
namespace: company.team

description: |
  Crony keeps an eye on your Kubernetes CronJobs. Every day it checks
  whether each CronJob ran on schedule and whether its recent runs
  succeeded. If one missed its run or failed, it pings Slack.
  If everything is healthy, it stays quiet.

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

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

  - id: list_jobs
    type: io.kestra.plugin.scripts.shell.Commands
    description: lists all the Jobs in the namespace, so we can see which runs failed
    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') }}"
    outputFiles:
      - jobs.json
    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

  - id: find_trouble
    type: io.kestra.plugin.scripts.python.Script
    description: checks every CronJob for missed and failed runs, drops the ones we
      already alerted on
    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


      def parse_time(s):
          # "2026-10-06T12:00:00Z" -> datetime
          return datetime.fromisoformat(s.replace('Z', '+00:00'))


      def interval_minutes(schedule):
          # guess how often a CronJob should run from its schedule,
          # only handles the common shapes, good enough for a missed check
          parts = schedule.split()
          if len(parts) != 5:
              return None
          minute, hour, dom, month, dow = parts
          if minute.startswith('*/'):
              try:
                  return int(minute[2:])
              except ValueError:
                  return None
          if minute == '*':
              return 1
          if hour == '*':
              return 60
          if dom == '*' and month == '*' and dow == '*':
              return 24 * 60
          if dom == '*' and month == '*':
              return 7 * 24 * 60
          return 30 * 24 * 60


      def job_failed(job):
          # a Job failed if Kubernetes says so
          status = job.get('status', {})
          if status.get('failed', 0) > 0:
              return True
          for cond in status.get('conditions', []):
              if cond.get('type') == 'Failed' and cond.get('status') == 'True':
                  return True
          return False


      namespace = """{{ inputs.namespace }}"""
      now = datetime.now(timezone.utc)

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

      # group Jobs under the CronJob that created them
      jobs_by_owner = {}
      for job in jobs:
          owners = job.get('metadata', {}).get('ownerReferences', [])
          for owner in owners:
              if owner.get('kind') == 'CronJob':
                  jobs_by_owner.setdefault(owner['uid'], []).append(job)

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

      troubled = []
      for cj in cronjobs:
          meta = cj.get('metadata', {})
          spec = cj.get('spec', {})
          status = cj.get('status', {})
          name = meta['name']
          if spec.get('suspend'):
              continue  # paused on purpose, not our problem
          reasons = []
          # missed: last schedule is older than twice the expected interval
          interval = interval_minutes(spec.get('schedule', ''))
          last = status.get('lastScheduleTime')
          if last:
              age_min = (now - parse_time(last)).total_seconds() / 60
              if interval and age_min > 2 * interval:
                  reasons.append(f"missed: last run {age_min / 60:.1f}h ago, expected every {interval}m")
          else:
              age_min = (now - parse_time(meta['creationTimestamp'])).total_seconds() / 60
              if interval and age_min > 2 * interval:
                  reasons.append("missed: never ran")
          # failed: any owned Job failed
          owned = jobs_by_owner.get(meta.get('uid'), [])
          failed = [j for j in owned if job_failed(j)]
          if failed:
              reasons.append(f"failed: {len(failed)} failed run(s)")
          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;
      # recovered names drop out on their own, so a future failure alerts again

      lines = [f"*Crony - troubled CronJobs in '{namespace}'*", ""]
      for t in new_trouble:
          lines.append(f"- *{t['name']}*: " + "; ".join(t['reasons']))
          lines.append(f"  inspect: `kubectl describe cronjob {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 pings Slack 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: sends the troubled CronJobs 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 recovered ones alert again next time
    key: "crony-state-{{ inputs.namespace }}"
    value: "{{ outputs.find_trouble.vars.new_state_json }}"
    overwrite: true

triggers:
  - id: daily
    type: io.kestra.plugin.core.trigger.Schedule
    cron: "0 8 * * *"
    timezone: UTC
    inputs:
      namespace: default
      slack_webhook: "{{ secret('SLACK_WEBHOOK') }}"