Cron Jobs for Developers: Patterns and Best Practices

# ── Basic patterns ──

# Every minute
* * * * * /script.sh

# Every 5 minutes
*/5 * * * * /script.sh

# Every 15 minutes
*/15 * * * * /script.sh

# Every hour at minute 0
0 * * * * /script.sh

# Every hour at minute 30
30 * * * * /script.sh

# Daily at midnight
0 0 * * * /script.sh

# Daily at 2:30 AM
30 2 * * * /script.sh

# Weekdays at 9 AM
0 9 * * 1-5 /script.sh

# Once a week (Sunday at 3 AM)
0 3 * * 0 /script.sh

# First day of every month at midnight
0 0 1 * * /script.sh

# Every 6 hours
0 */6 * * * /script.sh

# Every Monday and Thursday at 8 AM
0 8 * * 1,4 /script.sh

# daily_report.py
# Cron-friendly script with logging and error handling

#!/usr/bin/env python3
"""Generate daily analytics report. Run via cron at 6 AM."""
import sys
import json
import logging
from datetime import datetime, timedelta
from pathlib import Path

# Configure logging
LOG_DIR = Path("/var/log/app")
LOG_DIR.mkdir(parents=True, exist_ok=True)
LOG_FILE = LOG_DIR / "daily_report.log"

logging.basicConfig(
    level=logging.INFO,
    format="%(asctime)s [%(levelname)s] %(message)s",
    handlers=[
        logging.FileHandler(LOG_FILE),
        logging.StreamHandler(sys.stdout),
    ]
)
logger = logging.getLogger(__name__)

def generate_report():
    """Generate daily analytics report from database."""
    logger.info("Starting daily report generation")

    yesterday = datetime.now() - timedelta(days=1)
    report_date = yesterday.strftime("%Y-%m-%d")

    # Simulate data collection
    data = {
        "date": report_date,
        "total_users": 15234,
        "new_signups": 234,
        "active_sessions": 8912,
        "revenue": {
            "subscriptions": 45230.50,
            "one_time": 3200.00,
        },
        "errors": {
            "critical": 0,
            "warning": 12,
            "info": 45,
        }
    }

    # Write report
    report_path = Path(f"/var/reports/daily_{report_date}.json")
    report_path.parent.mkdir(parents=True, exist_ok=True)

    with open(report_path, 'w') as f:
        json.dump(data, f, indent=2)

    logger.info(f"Report saved to {report_path}")
    logger.info(f"Summary: {data['new_signups']} signups, "
                f"${data['revenue']['subscriptions']:.2f} revenue")

    # Return exit code: 0 for success
    return 0

if __name__ == "__main__":
    try:
        exit_code = generate_report()
        sys.exit(exit_code)
    except Exception as e:
        logger.critical(f"Report generation failed: {e}", exc_info=True)
        sys.exit(1)

# /etc/systemd/system/daily-report.service
[Unit]
Description=Generate daily analytics report

[Service]
Type=oneshot
ExecStart=/usr/bin/python3 /opt/app/scripts/daily_report.py
User=appuser
Group=appgroup

# Logging
StandardOutput=journal
StandardError=journal

# /etc/systemd/system/daily-report.timer
[Unit]
Description=Run daily report every morning at 6 AM

[Timer]
OnCalendar=*-*-* 06:00:00
Persistent=true  # Catch up if missed (e.g., after boot)
RandomizedDelaySec=300  # Random delay 0-5 min to avoid thundering herd

[Install]
WantedBy=timers.target

# Enable and start the timer
sudo systemctl daemon-reload
sudo systemctl enable daily-report.timer
sudo systemctl start daily-report.timer

# Check status
systemctl status daily-report.timer
systemctl list-timers

# cronjob.yaml
# Kubernetes CronJob for daily database backup
apiVersion: batch/v1
kind: CronJob
metadata:
  name: db-backup
spec:
  schedule: "0 2 * * *"     # Daily at 2 AM
  timeZone: "America/New_York"
  jobTemplate:
    spec:
      template:
        spec:
          containers:
          - name: backup
            image: postgres:16
            command:
            - /bin/sh
            - -c
            - |
              pg_dump -h postgres-svc -U admin mydb | \
              gzip > /backups/$(date +\%Y-\%m-\%d).sql.gz
            env:
            - name: PGPASSWORD
              valueFrom:
                secretKeyRef:
                  name: db-secret
                  key: password
            volumeMounts:
            - mountPath: /backups
              name: backup-storage
          restartPolicy: OnFailure
          volumes:
          - name: backup-storage
            persistentVolumeClaim:
              claimName: backup-pvc
  successfulJobsHistoryLimit: 7
  failedJobsHistoryLimit: 3

# eventbridge-schedule.yaml
# AWS EventBridge scheduled rule (via CloudFormation)
ScheduledRule:
  Type: AWS::Events::Rule
  Properties:
    Name: daily-report-generation
    ScheduleExpression: "cron(0 6 ? * MON-FRI *)"
    State: ENABLED
    Targets:
      - Arn: !GetAtt ReportFunction.Arn
        Id: ReportTarget
        Input: '{"type": "daily", "format": "json"}'

# cron_monitor.py
# Health check monitor for cron jobs

#!/usr/bin/env python3
"""Monitor cron job health — detect missed or failed runs."""
import json
from pathlib import Path
from datetime import datetime, timedelta
import smtplib
import sys

CRON_LOG_DIR = Path("/var/log/cron-jobs")
ALERT_EMAIL = "admin@example.com"

def check_cron_health():
    """Check all cron jobs ran successfully within expected intervals."""
    jobs = {
        "hourly_cleanup": {"interval_minutes": 70, "max_age_minutes": 90},
        "daily_report": {"interval_minutes": 1440, "max_age_minutes": 1500},
        "db_backup": {"interval_minutes": 1440, "max_age_minutes": 1500},
        "cert_renewal": {"interval_minutes": 43200, "max_age_minutes": 43500},  # 30 days
    }

    now = datetime.now()
    failed_jobs = []

    for job_name, config in jobs.items():
        log_file = CRON_LOG_DIR / f"{job_name}.log"

        if not log_file.exists():
            failed_jobs.append(f"{job_name}: No log file found")
            continue

        # Check last modification time
        mod_time = datetime.fromtimestamp(log_file.stat().st_mtime)
        age_minutes = (now - mod_time).total_seconds() / 60

        if age_minutes > config["max_age_minutes"]:
            failed_jobs.append(
                f"{job_name}: Last run {age_minutes:.0f} minutes ago "
                f"(expected < {config['max_age_minutes']})"
            )

    if failed_jobs:
        alert_message = "CRON HEALTH ALERT\n" + "=" * 40 + "\n"
        alert_message += "\n".join(failed_jobs)
        print(alert_message)
        # Send alert
        # send_alert(alert_message)
        return 1
    else:
        print(f"[OK] All cron jobs healthy at {now.isoformat()}")
        return 0

if __name__ == "__main__":
    sys.exit(check_cron_health())

# robust_cron.py
# Cron job with comprehensive error handling

#!/usr/bin/env python3
"""Production-grade cron script with error handling."""
import os
import sys
import time
import json
import logging
import subprocess
from pathlib import Path
from datetime import datetime

# Setup
LOCK_DIR = Path("/var/lock/cron")
LOCK_DIR.mkdir(parents=True, exist_ok=True)
JOB_NAME = Path(__file__).stem

def acquire_lock():
    """Prevent concurrent execution (flock-based)."""
    lock_file = LOCK_DIR / f"{JOB_NAME}.lock"
    # Use flock to prevent race conditions
    # In practice: run with `flock -n /var/lock/cron/daily_report.lock`
    return lock_file

def send_alert(subject, body):
    """Send monitoring alert (PagerDuty/Slack/Email)."""
    print(f"[ALERT] {subject}: {body}")
    # Integrate with Slack webhook or PagerDuty API
    # subprocess.run(["curl", "-X", "POST", "-d", json.dumps(body), webhook_url])

def main():
    """Main cron job logic with error handling."""
    start_time = time.time()
    logger = logging.getLogger(JOB_NAME)

    try:
        logger.info(f"Starting {JOB_NAME}")

        # 1. Pre-flight checks
        if not os.getenv("DATABASE_URL"):
            raise EnvironmentError("DATABASE_URL not set")
        if not Path("/var/data").exists():
            raise FileNotFoundError("/var/data directory missing")

        # 2. Business logic
        logger.info("Processing data...")
        time.sleep(2)  # Simulate work
        result = {"processed": 100, "errors": 0, "duration_s": 2.5}

        # 3. Write check-in file (for monitoring)
        checkin = Path(f"/var/checkin/{JOB_NAME}")
        checkin.parent.mkdir(parents=True, exist_ok=True)
        checkin.write_text(json.dumps({
            "status": "success",
            "timestamp": datetime.now().isoformat(),
            "duration": round(time.time() - start_time, 2),
        }))

        logger.info(f"Completed in {time.time() - start_time:.2f}s")
        return 0

    except Exception as e:
        logger.critical(f"Fatal error: {e}", exc_info=True)
        send_alert(
            f"Cron failure: {JOB_NAME}",
            f"Failed at {datetime.now().isoformat()}: {str(e)}"
        )
        return 1

if __name__ == "__main__":
    logging.basicConfig(
        level=logging.INFO,
        format="%(asctime)s [%(levelname)s] %(message)s",
        handlers=[
            logging.FileHandler(f"/var/log/cron/{JOB_NAME}.log"),
            logging.StreamHandler(sys.stdout),
        ]
    )
    sys.exit(main())

# log_analyzer.py
# Analyze cron logs for failures and timing patterns

#!/usr/bin/env python3
"""Analyze cron job logs to detect failures and performance trends."""
import re
from pathlib import Path
from datetime import datetime, timedelta
from collections import Counter, defaultdict

LOG_DIR = Path("/var/log/cron")

def analyze_cron_logs(days=7):
    """Parse cron logs and produce a health report."""
    cutoff = datetime.now() - timedelta(days=days)
    job_stats = defaultdict(lambda: {"runs": 0, "failures": 0, "durations": []})

    # Parse log files
    for log_file in sorted(LOG_DIR.glob("*.log")):
        job_name = log_file.stem

        with open(log_file) as f:
            for line in f:
                # Match: "2026-06-20 10:00:00 [INFO] Starting job_name"
                timestamp_match = re.match(
                    r"(\d{4}-\d{2}-\d{2} \d{2}:\d{2}:\d{2})", line
                )
                if not timestamp_match:
                    continue

                log_time = datetime.strptime(
                    timestamp_match.group(1), "%Y-%m-%d %H:%M:%S"
                )
                if log_time < cutoff:
                    continue

                job_stats[job_name]["runs"] += 1

                if "ERROR" in line or "CRITICAL" in line:
                    job_stats[job_name]["failures"] += 1

    # Print report
    print(f"{'Job':<20} {'Runs':<8} {'Failures':<10} {'Health':<10}")
    print("-" * 50)

    for job_name, stats in sorted(job_stats.items()):
        failure_rate = stats["failures"] / max(stats["runs"], 1) * 100
        health = "✅" if failure_rate < 1 else "⚠️" if failure_rate < 10 else "❌"
        print(f"{job_name:<20} {stats['runs']:<8} "
              f"{stats['failures']:<10} {health:<10}")

    total_jobs = len(job_stats)
    total_failures = sum(s["failures"] for s in job_stats.values())
    print(f"\nTotal jobs: {total_jobs}")
    print(f"Total failures: {total_failures}")
    print(f"Overall health rate: {(1 - total_failures / max(sum(s['runs'] for s in job_stats.values()), 1)) * 100:.1f}%")

if __name__ == "__main__":
    analyze_cron_logs()