Kubernetes CronJob Status Monitor

Kubernetes CronJob Status Monitor

Problem Statement

Kubernetes CronJobs are notoriously easy to misconfigure and hard to observe. A job can silently fail to schedule, get stuck in a Suspended state, or fail repeatedly without anyone noticing until a downstream data pipeline breaks. The kubectl get cronjobs output only shows the last schedule time and active job count — it doesn't tell you whether the last N runs succeeded, how long ago the last successful completion occurred, or whether failures are piling up.

This script solves that by querying the Kubernetes API directly, inspecting each CronJob's status.lastSuccessfulTime, status.lastScheduleTime, active jobs, and recent Job history. It then classifies each CronJob as HEALTHY, WARNING, or CRITICAL based on configurable thresholds, and emits both a human-readable summary and a machine-readable JSON report suitable for alerting pipelines or dashboards.

How It Works

The script uses the official kubernetes Python client to authenticate against the current kubeconfig context (or in-cluster service account if running inside a pod). For each CronJob in the target namespace, it fetches the parent CronJob object plus the child batch/v1 Jobs that it owns via ownerReferences. It computes three derived metrics: (1) missed schedules — the number of expected runs since lastSuccessfulTime that never completed, based on the cron expression and the current time; (2) consecutive failures — walking the child Jobs in reverse chronological order and counting failed ones until a success is found; and (3) staleness — the age of the last successful run relative to the cron schedule.

Classification is threshold-driven: --warn-missed N triggers a warning when missed schedules ≥ N; --alert-failed promotes any CronJob with ≥1 consecutive failed job to CRITICAL. The croniter library is used to compute schedule expectations from the cron expression. The script exits with code 0 if all CronJobs are healthy, 1 if any warnings, and 2 if any criticals — making it directly usable as a Kubernetes CronJob/Job gate or CI check.

Full Script

#!/usr/bin/env python3
"""
Kubernetes CronJob Status Monitor

Inspects CronJobs in a namespace, computes missed-schedule counts and
consecutive-failure streaks, and emits a health report (text or JSON).

Exit codes:
    0 = all healthy
    1 = at least one WARNING
    2 = at least one CRITICAL (or a fatal error occurred)
"""

from __future__ import annotations

import argparse
import json
import logging
import os
import sys
from dataclasses import dataclass, field, asdict
from datetime import datetime, timezone
from typing import Any, Iterable

from croniter import croniter
from kubernetes import client, config
from kubernetes.client.exceptions import ApiException

# ---------------------------------------------------------------------------
# Configuration (env vars with sensible defaults)
# ---------------------------------------------------------------------------
DEFAULT_NAMESPACE = os.environ.get("K8S_NAMESPACE", "default")
DEFAULT_WARN_MISSED = int(os.environ.get("K8S_WARN_MISSED", "2"))
DEFAULT_ALERT_FAILED = os.environ.get("K8S_ALERT_FAILED", "false").lower() == "true"
DEFAULT_CONTEXT = os.environ.get("K8S_CONTEXT")  # None => use current context
LOG_LEVEL = os.environ.get("LOG_LEVEL", "INFO").upper()

logging.basicConfig(
    level=LOG_LEVEL,
    format="%(asctime)s %(levelname)-7s %(message)s",
    datefmt="%Y-%m-%dT%H:%M:%S%z",
)
log = logging.getLogger("k8s-cronjob-monitor")

# ---------------------------------------------------------------------------
# Data model
# ---------------------------------------------------------------------------
@dataclass
class CronJobReport:
    name: str
    namespace: str
    schedule: str
    suspended: bool
    last_schedule_time: str | None
    last_successful_time: str | None
    missed_schedules: int
    consecutive_failures: int
    active_jobs: int
    status: str = "HEALTHY"
    reasons: list[str] = field(default_factory=list)

    def to_dict(self) -> dict[str, Any]:
        return asdict(self)


# ---------------------------------------------------------------------------
# Helpers
# ---------------------------------------------------------------------------
def parse_k8s_time(value: Any) -> datetime | None:
    """Kubernetes returns timestamps as datetime objects (or None)."""
    if value is None:
        return None
    if isinstance(value, datetime):
        return value if value.tzinfo else value.replace(tzinfo=timezone.utc)
    # Fallback: parse string form
    try:
        return datetime.fromisoformat(str(value).replace("Z", "+00:00"))
    except ValueError:
        log.warning("Could not parse timestamp %r", value)
        return None


def count_missed_schedules(
    schedule: str,
    last_success: datetime | None,
    now: datetime,
    suspend: bool,
) -> int:
    """
    Number of cron fires since last_success (exclusive of the success itself).
    If we've never succeeded, we count fires since the CronJob was created —
    but we don't know that here, so we conservatively return 0.
    """
    if suspend or last_success is None:
        return 0
    try:
        it = croniter(schedule, last_success)
    except (ValueError, KeyError) as exc:
        log.warning("Invalid cron expression %r: %s", schedule, exc)
        return 0

    missed = 0
    # Iterate forward; cap at 10k to avoid pathological expressions.
    for _ in range(10_000):
        nxt = it.get_next(datetime)
        if nxt.tzinfo is None:
            nxt = nxt.replace(tzinfo=timezone.utc)
        if nxt > now:
            break
        missed += 1
    return missed


def count_consecutive_failures(jobs: Iterable[client.V1Job]) -> int:
    """
    Walk child Jobs newest-first; count failed ones until we hit a success.
    A Job is considered successful if it has a `Complete` condition.
    """
    def sort_key(j: client.V1Job) -> datetime:
        ts = j.status.completion_time or j.status.start_time or j.metadata.creation_timestamp
        return parse_k8s_time(ts) or datetime.min.replace(tzinfo=timezone.utc)

    ordered = sorted(jobs, key=sort_key, reverse=True)
    streak = 0
    for job in ordered:
        conditions = (job.status.conditions or []) if job.status else []
        succeeded = any(c.type == "Complete" and c.status == "True" for c in conditions)
        failed = any(c.type == "Failed" and c.status == "True" for c in conditions)
        if succeeded:
            break
        if failed:
            streak += 1
    return streak


# ---------------------------------------------------------------------------
# Core logic
# ---------------------------------------------------------------------------
def build_clients(context: str | None) -> tuple[client.BatchV1Api, client.CoreV1Api]:
    """Load kubeconfig (or in-cluster) and return API clients."""
    try:
        if os.environ.get("KUBERNETES_SERVICE_HOST"):
            config.load_incluster_config()
            log.info("Loaded in-cluster configuration")
        else:
            config.load_kube_config(context=context)
            log.info("Loaded kubeconfig (context=%s)", context or "current")
    except Exception as exc:  # ConfigException, FileNotFoundError, etc.
        log.error("Failed to load Kubernetes config: %s", exc)
        raise SystemExit(2)

    return client.BatchV1Api(), client.CoreV1Api()


def list_cronjobs(batch: client.BatchV1Api, namespace: str) -> list[client.V1CronJob]:
    try:
        result = batch.list_namespaced_cron_job(namespace=namespace)
        return result.items or []
    except ApiException as exc:
        log.error("Kubernetes API error listing CronJobs: %s %s", exc.status, exc.reason)
        raise SystemExit(2)


def list_jobs_for(
    batch: client.BatchV1Api,
    namespace: str,
    cronjob_name: str,
    uid: str,
) -> list[client.V1Job]:
    """
    Fetch Jobs owned by this CronJob. We filter server-side by label
    (Kubernetes sets `job-name` but not a stable cronjob label by default),
    so we list all Jobs and filter by ownerReference UID client-side.
    """
    try:
        jobs = batch.list_namespaced_job(namespace=namespace).items or []
    except ApiException as exc:
        log.warning("Could not list Jobs in %s: %s", namespace, exc.reason)
        return []

    owned: list[client.V1Job] = []
    for job in jobs:
        refs = job.metadata.owner_references or []
        if any(r.uid == uid and r.kind == "CronJob" for r in refs):
            owned.append(job)
    return owned


def analyse_cronjob(
    batch: client.BatchV1Api,
    cj: client.V1CronJob,
    warn_missed: int,
    alert_failed: bool,
    now: datetime,
) -> CronJobReport:
    name = cj.metadata.name
    namespace = cj.metadata.namespace
    schedule = cj.spec.schedule
    suspend = bool(cj.spec.suspend)

    last_sched = parse_k8s_time(cj.status.last_schedule_time if cj.status else None)
    last_success = parse_k8s_time(cj.status.last_successful_time if cj.status else None)
    active = len(cj.status.active or []) if cj.status else 0

    jobs = list_jobs_for(batch, namespace, name, cj.metadata.uid)
    failures = count_consecutive_failures(jobs)
    missed = count_missed_schedules(schedule, last_success, now, suspend)

    report = CronJobReport(
        name=name,
        namespace=namespace,
        schedule=schedule,
        suspended=suspend,
        last_schedule_time=last_sched.isoformat() if last_sched else None,
        last_successful_time=last_success.isoformat() if last_success else None,
        missed_schedules=missed,
        consecutive_failures=failures,
        active_jobs=active,
    )

    # Classify
    if suspend:
        report.reasons.append("CronJob is suspended")
        report.status = "WARNING"
    if missed >= warn_missed and warn_missed > 0:
        report.reasons.append(f"{missed} missed schedules (threshold {warn_missed})")
        report.status = "CRITICAL" if report.status == "CRITICAL" else "WARNING"
    if alert_failed and failures > 0:
        report.reasons.append(f"{failures} consecutive failed jobs")
        report.status = "CRITICAL"

    return report


def render_text(reports: list[CronJobReport]) -> str:
    lines: list[str] = []
    for r in sorted(reports, key=lambda x: (x.status, x.name)):
        icon = {"HEALTHY": "OK  ", "WARNING": "WARN", "CRITICAL": "FAIL"}[r.status]
        lines.append(
            f"[{icon}] {r.namespace}/{r.name}  schedule={r.schedule}  "
            f"missed={r.missed_schedules}  failures={r.consecutive_failures}  "
            f"active={r.active_jobs}"
        )
        for reason in r.reasons:
            lines.append(f"         └─ {reason}")
    return "\n".join(lines)


# ---------------------------------------------------------------------------
# Entry point
# ---------------------------------------------------------------------------
def main(argv: list[str] | None = None) -> int:
    parser = argparse.ArgumentParser(
        description="Monitor Kubernetes CronJob health and emit a report.",
    )
    parser.add_argument("--namespace", "-n", default=DEFAULT_NAMESPACE,
                        help="Namespace to inspect (default: %(default)s)")
    parser.add_argument("--warn-missed", type=int, default=DEFAULT_WARN_MISSED,
                        help="Warn if missed schedules >= N (default: %(default)s)")
    parser.add_argument("--alert-failed", action="store_true",
                        default=DEFAULT_ALERT_FAILED,
                        help="Mark CRITICAL on any consecutive failure")
    parser.add_argument("--output", "-o", choices=["text", "json"], default="text",
                        help="Output format (default: %(default)s)")
    parser.add_argument("--output-file", default=None,
                        help="Write report to file instead of stdout")
    parser.add_argument("--context", default=DEFAULT_CONTEXT,
                        help="kubeconfig context to use")
    args = parser.parse_args(argv)

    now = datetime.now(timezone.utc)
    batch, _core = build_clients(args.context)
    cronjobs = list_cronjobs(batch, args.namespace)

    if not cronjobs:
        log.warning("No CronJobs found in namespace %s", args.namespace)

    reports = [
        analyse_cronjob(batch, cj, args.warn_missed, args.alert_failed, now)
        for cj in cronjobs
    ]

    if args.output == "json":
        payload = {
            "generated_at": now.isoformat(),
            "namespace": args.namespace,
            "cronjobs": [r.to_dict() for r in reports],
            "summary": {
                "total": len(reports),
                "healthy": sum(r.status == "HEALTHY" for r in reports),
                "warning": sum(r.status == "WARNING" for r in reports),
                "critical": sum(r.status == "CRITICAL" for r in reports),
            },
        }
        rendered = json.dumps(payload, indent=2)
    else:
        rendered = render_text(reports) or "(no cronjobs to report)"

    if args.output_file:
        with open(args.output_file, "w", encoding="utf-8") as fh:
            fh.write(rendered + "\n")
        log.info("Wrote report to %s", args.output_file)
    else:
        print(rendered)

    if any(r.status == "CRITICAL" for r in reports):
        return 2
    if any(r.status == "WARNING" for r in reports):
        return 1
    return 0


if __name__ == "__main__":
    try:
        sys.exit(main())
    except KeyboardInterrupt:
        log.error("Interrupted")
        sys.exit(130)

Configuration

CLI flags --namespace, --warn-missed, --alert-failed, --output, --output-file, and --context override the corresponding env vars.

Usage Examples

1. Quick text report against a namespace:

$ python3 k8s_cronjobs.py --namespace jobs --warn-missed 2 --alert-failed
[OK  ] jobs/nightly-backup   schedule=0 2 * * *  missed=0  failures=0  active=0
[WARN] jobs/report-builder   schedule=*/15 * * * *  missed=3  failures=0  active=0
         └─ 3 missed schedules (threshold 2)
[FAIL] jobs/etl-ingest       schedule=*/5 * * * *  missed=0  failures=4  active=0
         └─ 4 consecutive failed jobs
$ echo $?
2

2. JSON report written to a file for a dashboard:

$ python3 k8s_cronjobs.py -n jobs --output json --output-file report.json
2025-01-14T09:12:03+0000 INFO    Wrote report to report.json
$ jq '.summary' report.json
{ "total": 3, "healthy": 1, "warning": 1, "critical": 1 }

3. In-cluster run inside a Kubernetes CronJob:

# A "CronJob to monitor CronJobs" — meta but useful.
apiVersion: batch/v1
kind: CronJob
metadata: { name: cronjob-monitor, namespace: jobs }
spec:
  schedule: "*/30 * * * *"
  jobTemplate:
    spec:
      template:
        spec:
          serviceAccountName: cronjob-monitor  # needs get/list on cronjobs & jobs
          restartPolicy: OnFailure
          containers:
            - name: monitor
              image: myorg/k8s-cronjob-monitor:1.0
              args: ["--namespace", "jobs", "--alert-failed", "--output", "json"]

Error Handling

• Config loading failures (~/.kube/config missing, bad context, expired token) are caught in build_clients and exit with code 2 after logging the reason.
• API errors (ApiException) — 403 from RBAC, 404 from a deleted namespace, etc. — are logged with status + reason. Fatal for the CronJob list, non-fatal (returns empty) for the child Job list, so a single permission gap doesn't kill the whole report.
• Malformed cron expressions are caught in count_missed_schedules; the CronJob is reported with missed=0 and a warning is logged rather than crashing.
• Missing timestamps (a CronJob that has never run) are handled by parse_k8s_time returning None, and missed-schedule counting short-circuits to 0.
• Interruptions (Ctrl-C) exit cleanly with code 130.

Extending Ideas

• Prometheus exporter mode: add a --format prometheus option that emits gauges like k8s_cronjob_missed_schedules{namespace,job} so you can scrape it from a sidecar or node_exporter textfile collector.
• Webhook alerting: on CRITICAL, POST the JSON payload to Slack/PagerDuty/Opsgenie using requests. Add retry with exponential backoff for transient webhook failures.
• Historical persistence: write each run's summary to a SQLite/Postgres table so you can compute MTBF, mean time to recovery, and failure trends over time.
• Multi-namespace / cluster fan-out: accept --namespace multiple times (or --all-namespaces) and iterate client.BatchV1Api().list_cron_job_for_all_namespaces(), aggregating results into a single report with per-namespace summaries.