Cron Monitoring Best Practices

#!/bin/bash
# /usr/local/bin/monitored-cron.sh
# Cron job with push-based health check monitoring

JOB_NAME="$1"
PING_URL="https://hc-ping.com/YOUR-UUID"
START_TIME=$(date +%s)

shift

# Signal start
curl -fsS -m 10 "${PING_URL}/start" > /dev/null 2>&1

# Execute the job
"$@" 2>&1
EXIT_CODE=$?

END_TIME=$(date +%s)
DURATION=$((END_TIME - START_TIME))

if [ $EXIT_CODE -eq 0 ]; then
    # Signal success with duration
    curl -fsS -m 10 "${PING_URL}/${DURATION}" > /dev/null 2>&1
else
    # Signal failure
    curl -fsS -m 10 "${PING_URL}/fail" > /dev/null 2>&1
fi

exit $EXIT_CODE

# Usage in crontab:
# 0 3 * * * /usr/local/bin/monitored-cron.sh daily-backup /usr/local/bin/backup.sh

#!/usr/bin/env python3
"""Export cron job metrics to Prometheus."""
import time
import os
import json
from prometheus_client import start_http_server, Gauge, Counter, Histogram

cron_success = Counter('cron_job_success_total', 'Successful cron executions', ['job_name'])
cron_failure = Counter('cron_job_failure_total', 'Failed cron executions', ['job_name'])
cron_duration = Histogram('cron_job_duration_seconds', 'Cron job duration', ['job_name'])
cron_last_success = Gauge('cron_job_last_success_timestamp', 'Last successful execution', ['job_name'])
cron_last_duration = Gauge('cron_job_last_duration_seconds', 'Last execution duration', ['job_name'])

class CronMetrics:
    def __init__(self, port=8000):
        self.port = port

    def start(self):
        start_http_server(self.port)
        print(f"Cron metrics server on :{self.port}")

    def record_success(self, job_name, duration):
        cron_success.labels(job_name=job_name).inc()
        cron_last_success.labels(job_name=job_name).set(time.time())
        cron_last_duration.labels(job_name=job_name).set(duration)
        cron_duration.labels(job_name=job_name).observe(duration)

    def record_failure(self, job_name, duration):
        cron_failure.labels(job_name=job_name).inc()
        cron_last_duration.labels(job_name=job_name).set(duration)

metrics = CronMetrics()
metrics.start()

metrics.record_success('daily-backup', 145.2)
metrics.record_success('hourly-cache', 2.1)
metrics.record_failure('weekly-report', 60.5)

try:
    while True:
        time.sleep(1)
except KeyboardInterrupt:
    pass

# Prometheus scrape config
scrape_configs:
  - job_name: 'cron-metrics'
    static_configs:
      - targets: ['localhost:8000']

#!/usr/bin/env python3
"""Simple cron job status dashboard using SQLite."""
import sqlite3
import time
import os
from datetime import datetime

class CronDashboard:
    def __init__(self, db_path='/var/lib/cron-monitor/dashboard.db'):
        os.makedirs(os.path.dirname(db_path), exist_ok=True)
        self.conn = sqlite3.connect(db_path)
        self._init_db()

    def _init_db(self):
        self.conn.execute('''
            CREATE TABLE IF NOT EXISTS cron_jobs (
                id INTEGER PRIMARY KEY AUTOINCREMENT,
                job_name TEXT NOT NULL,
                hostname TEXT NOT NULL,
                status TEXT NOT NULL,
                exit_code INTEGER,
                duration REAL,
                started_at TIMESTAMP,
                finished_at TIMESTAMP,
                output TEXT
            )
        ''')
        self.conn.execute('''
            CREATE TABLE IF NOT EXISTS cron_schedules (
                job_name TEXT PRIMARY KEY,
                expected_interval INTEGER,
                last_seen TIMESTAMP,
                alert_email TEXT
            )
        ''')
        self.conn.commit()

    def record_job(self, job_name, status, exit_code, duration):
        self.conn.execute('''
            INSERT INTO cron_jobs
            (job_name, hostname, status, exit_code, duration, started_at, finished_at)
            VALUES (?, ?, ?, ?, ?, ?, ?)
        ''', (
            job_name,
            os.uname().nodename,
            status,
            exit_code,
            duration,
            datetime.now(),
            datetime.now(),
        ))
        self.conn.execute('''
            INSERT OR REPLACE INTO cron_schedules
            (job_name, last_seen)
            VALUES (?, ?)
        ''', (job_name, datetime.now()))
        self.conn.commit()

    def get_summary(self):
        cur = self.conn.execute('''
            SELECT job_name,
                   COUNT(*) as total,
                   SUM(CASE WHEN status = 'success' THEN 1 ELSE 0 END) as successes,
                   SUM(CASE WHEN status = 'failure' THEN 1 ELSE 0 END) as failures,
                   ROUND(AVG(duration), 2) as avg_duration
            FROM cron_jobs
            WHERE started_at > datetime('now', '-7 days')
            GROUP BY job_name
        ''')
        return cur.fetchall()

    def get_recent_failures(self, limit=10):
        cur = self.conn.execute('''
            SELECT job_name, hostname, exit_code, started_at
            FROM cron_jobs
            WHERE status = 'failure'
            ORDER BY started_at DESC
            LIMIT ?
        ''', (limit,))
        return cur.fetchall()

dashboard = CronDashboard()

dashboard.record_job('daily-backup', 'success', 0, 145.2)
dashboard.record_job('hourly-cache', 'success', 0, 2.1)
dashboard.record_job('weekly-report', 'failure', 1, 60.5)

print("Weekly Summary:")
for row in dashboard.get_summary():
    print(f"  {row[0]}: {row[2]}/{row[1]} successful, avg {row[4]}s")

print("\nRecent Failures:")
for row in dashboard.get_recent_failures():
    print(f"  {row[0]} on {row[1]} at {row[3]}")

#!/usr/bin/env python3
"""Monitor cron job liveness: verify jobs are still running on schedule."""
import time
import redis
import requests

r = redis.Redis()

class CronLivenessMonitor:
    def __init__(self, tolerance_minutes=5):
        self.tolerance = tolerance_minutes * 60

    def check_job(self, job_name):
        last_seen = r.get(f"cron:last_run:{job_name}")
        if not last_seen:
            return {'status': 'unknown', 'message': 'Never seen'}

        last_time = float(last_seen)
        elapsed = time.time() - last_time

        if elapsed > self.tolerance:
            return {
                'status': 'missing',
                'message': f'Last run {elapsed:.0f}s ago (> {self.tolerance}s)',
            }
        return {
            'status': 'ok',
            'message': f'Last run {elapsed:.0f}s ago',
        }

    def check_all(self, job_names):
        results = {}
        for name in job_names:
            results[name] = self.check_job(name)
        return results

monitor = CronLivenessMonitor(tolerance_minutes=10)

r.setex('cron:last_run:daily-backup', 3600, time.time() - 300)
r.setex('cron:last_run:hourly-cache', 3600, time.time() - 60)
r.setex('cron:last_run:weekly-report', 3600, time.time() - 3600)

jobs = ['daily-backup', 'hourly-cache', 'weekly-report']
results = monitor.check_all(jobs)
for name, result in results.items():
    status_icon = "OK" if result['status'] == 'ok' else "MISSING"
    print(f"  [{status_icon}] {name}: {result['message']}")

# prometheus-rules.yml
groups:
- name: cron-jobs
  rules:
  - alert: CronJobFailing
    expr: rate(cron_job_failure_total[1h]) > 0
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "Cron job {{ $labels.job_name }} is failing"

  - alert: CronJobNotRunning
    expr: time() - cron_job_last_success_timestamp > 3600
    for: 10m
    labels:
      severity: critical
    annotations:
      summary: "Cron job {{ $labels.job_name }} has not run recently"

  - alert: CronJobSlow
    expr: cron_job_last_duration_seconds > 300
    for: 5m
    labels:
      severity: warning
    annotations:
      summary: "Cron job {{ $labels.job_name }} is running slowly"

# alertmanager.yml
receivers:
- name: 'cron-alerts'
  slack_configs:
  - api_url: 'https://hooks.slack.com/services/...'
    channel: '#cron-alerts'
    send_resolved: true
  pagerduty_configs:
  - routing_key: 'YOUR_PD_KEY'
    severity: 'critical'

#!/usr/bin/env python3
"""Complete cron monitoring system with heartbeats and alerts."""
import time
import json
import redis
import requests
from datetime import datetime

r = redis.Redis()

class CronMonitor:
    def __init__(self, heartbeat_ttl=300, alert_webhook=None):
        self.heartbeat_key = 'cron:heartbeats'
        self.alert_webhook = alert_webhook

    def heartbeat(self, job_name, duration, exit_code):
        status = 'success' if exit_code == 0 else 'failure'
        entry = {
            'job': job_name,
            'host': __import__('socket').gethostname(),
            'status': status,
            'duration': duration,
            'exit_code': exit_code,
            'timestamp': time.time(),
        }
        r.hset(self.heartbeat_key, job_name, json.dumps(entry))
        r.expire(self.heartbeat_key, self.heartbeat_ttl)

        if exit_code != 0:
            self._send_alert(f"Cron failed: {job_name} (exit: {exit_code})")

    def get_status(self, job_name):
        data = r.hget(self.heartbeat_key, job_name)
        if not data:
            return {'status': 'unknown', 'job': job_name}
        return json.loads(data)

    def get_all_status(self):
        data = r.hgetall(self.heartbeat_key)
        result = {}
        for name, entry in data.items():
            result[name.decode()] = json.loads(entry)
        return result

    def _send_alert(self, message):
        if self.alert_webhook:
            try:
                requests.post(self.alert_webhook, json={'text': message}, timeout=5)
            except requests.RequestException:
                pass
        print(f"ALERT: {message}")

monitor = CronMonitor(alert_webhook='https://hooks.slack.com/services/...')

monitor.heartbeat('daily-backup', 145.2, 0)
monitor.heartbeat('hourly-cache', 2.1, 0)
monitor.heartbeat('weekly-report', 60.5, 1)

statuses = monitor.get_all_status()
for name, status in statuses.items():
    print(f"  {name}: {status['status']} ({status['duration']}s)")