python-watchdog-cooldown

A tiny, dependency-free health watchdog for cron-driven Python jobs. It watches your log file and your last "heartbeat" from cron, decides whether a symptom warrants alerting, and enforces a per-reason cooldown so a single incident does not become a pager storm.

The problem

Cron jobs fail silently. Your log file grows for hours before anyone notices. When you finally wire up an alert, the first incident sends fifty duplicate messages because your script checks every minute and the underlying issue is the same one.

This library is what I extract every time I stand up a new cron-driven service. Now it lives in one place.

What it does

• Scans a log file for symptom patterns you declare (ERROR, Traceback, custom regex).
• Detects a stalled cron by comparing "wall clock now" to the last marker line the job wrote.
• Detects a task that started and never finished by comparing paired markers.
• Groups symptoms by reason and enforces a cooldown per reason (default 6 hours), stored in SQLite.
• Delegates the actual "notify" step to a callable you inject, so you can post to Slack, Teams, email, or anywhere else.

No daemon. No pip dependencies outside the stdlib. You run it as its own cron entry (typically every 5 or 15 minutes) and it decides what to do.

Design decisions

• SQLite is the source of truth for cooldowns. File-based, no daemon needed, atomic writes. Survives reboot.
• Reasons are stable strings, not stack traces. A "cron_stalled" reason should coalesce all stalled-cron alerts regardless of exact symptom text.
• Notifications are injected, not built in. The library never opens a socket; you pass a notify callable that receives a structured Alert object.
• Idempotent by construction. Running the watchdog twice in quick succession is safe. Cooldown windows prevent duplicate notifications.
• Log scanning is bounded. The watchdog reads only the tail of the log (configurable), never the whole file. Log rotation is respected.

Requirements

• Python 3.11 or later.
• No third-party runtime dependencies.

Install

pip install -e .

Usage

from pathlib import Path
from watchdog_cooldown import Watchdog, PatternRule, StaleMarkerRule

def notify(alert):
    print(f"[{alert.reason}] {alert.summary}")

wd = Watchdog(
    state_db=Path("state.sqlite"),
    log_path=Path("service.log"),
    tail_bytes=200_000,
    default_cooldown_seconds=6 * 3600,
    notifier=notify,
)

wd.add_rule(PatternRule(reason="log_error", pattern=r"\bERROR\b|Traceback"))
wd.add_rule(StaleMarkerRule(
    reason="cron_stalled",
    marker=r"\[CRON RUN END\]",
    max_age_seconds=45 * 60,
    now=None,
))

wd.check()

Wire it as a cron entry:

*/15 * * * * /usr/bin/python /opt/health/watch.py >> /var/log/watchdog.log 2>&1

Tests

pytest tests/

Tests cover: cooldown enforcement, multiple reasons in one run, stale marker detection, and log rotation handling.

License

MIT. See LICENSE.