Observable jobs: exit contracts, logs, metrics and audit trails

"""The exit contract for this course's jobs: what each status means, and what the caller does."""
from enum import IntEnum
 
 
class Exit(IntEnum):
    OK = 0          # everything converged (or already was): nothing to do
    FAILED = 1      # a human must look: a permanent failure, unusable input, or findings from a checker
    USAGE = 2       # the command line itself is wrong: unknown flag, missing or malformed argument
    SOFTWARE = 70   # crashed on an error nobody classified: a bug; traceback in the log (EX_SOFTWARE)
    TEMPFAIL = 75   # only transient failures: a later run can finish the work (EX_TEMPFAIL)
    CONFIG = 78     # the configuration is unusable; nothing was attempted (EX_CONFIG)
    # 126, 127 and 128+n belong to the shell and the kernel: 137 is 128+9, killed by SIGKILL.
 
 
class PermanentError(Exception):
    """Retrying cannot help (refused by policy, bad data): the run ends with Exit.FAILED."""
 
 
class TransientError(Exception):
    """May work later (timeout, outage, another run busy): the run ends with Exit.TEMPFAIL."""

"""sweep: bring every target to the wanted key generation and leave four traces behind: a JSON
log, a metric, an audit trail and an exit status from the contract in exit_codes.py."""
import argparse
import sys
from pathlib import Path
 
import metrics
from config import ConfigError, load_config
from converge import Report, converge
from exit_codes import Exit, TransientError
from jsonlog import get_logger
from lock import single_instance
 
log = get_logger("sweep")
 
 
def outcome(report: Report) -> Exit:
    """A failure that needs a human outranks one that may clear up by itself."""
    if report.permanent:
        return Exit.FAILED
    if report.transient:
        return Exit.TEMPFAIL
    return Exit.OK
 
 
def main(argv: list[str]) -> int:
    parser = argparse.ArgumentParser(prog="sweep")
    parser.add_argument("--config", type=Path, default=Path("sweep.toml"))
    parser.add_argument("--dry-run", action="store_true", help="report what would change, change nothing")
    args = parser.parse_args(argv)  # a bad flag exits 2 here, before anything else runs
    try:
        cfg = load_config(args.config)
    except ConfigError as err:
        log.error("config error", extra={"fields": {"error": str(err)}})
        return Exit.CONFIG
 
    report = Report()
    try:
        with single_instance(Path(cfg["lock"])):
            converge(cfg, args.dry_run, report)
        status = outcome(report)
    except TransientError as err:  # the lock is held: another run is doing the work
        log.warning("not started", extra={"fields": {"error": str(err)}})
        return Exit.TEMPFAIL
    except Exception:  # no rule for this error: stop, and say it is a bug, not a retry
        log.exception("crashed")
        status = Exit.SOFTWARE
    if not args.dry_run:
        status = write_metric(cfg, report, status)
    log.info("run complete", extra={"fields": {"changed": report.changed, "permanent": report.permanent,
                                               "transient": report.transient, "exit": int(status)}})
    return status
 
 
def write_metric(cfg: dict, report: Report, status: Exit) -> Exit:
    """The metric is how monitoring sees this job. If it cannot be written, the run reports
    a software failure (70) in one log line, instead of a traceback and Python's exit 1."""
    failed = len(report.permanent) + len(report.transient)
    try:
        metrics.record_run(Path(cfg["metrics"]), status, report.changed, failed)
    except Exception as err:
        detail = getattr(err, "strerror", None) or str(err)
        log.error("metric not written", extra={"fields": {
            "path": cfg["metrics"], "error": f"{type(err).__name__}: {detail}"}})
        return Exit.SOFTWARE
    return status
 
 
if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))

# What sweep changes, and where it leaves its traces. Reviewed and version-controlled.
targets = ["web01", "web02", "db01"]
generation = 2
fleet = "fleet"
metrics = "/var/lib/scr-observability/textfile/sweep.prom"
audit = "audit/sweep.jsonl"
lock = "sweep.lock"

#!/usr/bin/env bash
# A stand-in fleet for sweep: one JSON file per host, holding the key generation deployed there.
# On real hosts this state is on the host itself; sweep reads it before it changes anything.
set -euo pipefail
cd "$(dirname "$0")"
mkdir -p fleet audit metrics
for host in web01 web02 db01; do
  printf '{"generation": 1}\n' >"fleet/$host.json"
done
printf '{"generation": 1, "frozen": true}\n' >fleet/pay01.json     # change freeze: a human decides
printf '{"generation": 1, "reachable": false}\n' >fleet/dr01.json  # outage: may answer later
ls fleet

#!/usr/bin/env bash
# Run sweep from cron. cron retries nothing and mails whatever a job prints, so this wrapper
# turns the exit contract into "stay quiet" or "tell a human".
cd "$(dirname "$0")" || exit 78
python3 sweep.py --config "${1:-sweep.toml}" 2>>sweep.log
status=$?
case $status in
  0) ;;
  75) printf 'sweep: transient failure (75); the next run retries\n' >>sweep.log ;;
  *) printf 'sweep: exit %s needs a human; the log is %s/sweep.log\n' "$status" "$PWD" ;;
esac
exit "$status"

"""JSON logs on stderr with UTC-aware timestamps, so stdout stays free for the tool's real output."""
import json
import logging
import sys
from datetime import datetime, timezone
 
 
class JsonFormatter(logging.Formatter):
    def format(self, record: logging.LogRecord) -> str:
        payload = {
            # datetime.now(timezone.utc) is the supported UTC clock. datetime.utcnow() is deprecated
            # since 3.12: it returned a naive datetime that only pretended to be UTC.
            "ts": datetime.fromtimestamp(record.created, timezone.utc).isoformat(),
            "level": record.levelname.lower(),
            "logger": record.name,
            "msg": record.getMessage(),
        }
        for key, value in (getattr(record, "fields", None) or {}).items():
            # A field never replaces ts, level, logger or msg: a value taken from input could
            # otherwise forge them inside a well-formed line. A clashing key gets a prefix.
            payload[f"field_{key}" if key in payload else key] = value
        if record.exc_info:
            payload["exc"] = self.formatException(record.exc_info)
        # json.dumps escapes control characters, so a newline or an ESC in a value becomes
        # \n or \u001b and cannot start a second log line.
        return json.dumps(payload, ensure_ascii=True)
 
 
def get_logger(name: str) -> logging.Logger:
    handler = logging.StreamHandler(sys.stderr)
    handler.setFormatter(JsonFormatter())
    log = logging.getLogger(name)
    log.handlers = [handler]
    log.setLevel(logging.INFO)
    log.propagate = False
    return log

"""Write a node_exporter textfile metric atomically, readable by the scraper's own user."""
import os
import tempfile
import time
from pathlib import Path
 
 
def write_textfile(path: Path, body: str) -> None:
    """A scrape sees the old file or the whole new one, never half of one, and the node_exporter
    user (not this job's user) can read it."""
    fd, tmp = tempfile.mkstemp(dir=path.parent, prefix="." + path.name + ".")
    try:
        with os.fdopen(fd, "w") as f:
            f.write(body)
            f.flush()
            os.fsync(f.fileno())
        # mkstemp creates the file 0600 for its own safety; node_exporter runs as its own user
        # and could not read that. Widen to 0644 before the rename makes the name visible.
        os.chmod(tmp, 0o644)
        os.replace(tmp, path)  # atomic within one filesystem
    except BaseException:
        os.unlink(tmp)
        raise
 
 
SERIES = {  # name: help text. Names end in _seconds where the value is a Unix time.
    "sweep_last_run_seconds": "Unix time the last run finished, whatever its outcome.",
    "sweep_last_success_seconds": "Unix time of the last fully successful run.",
    "sweep_last_exit_code": "Exit status of the last run.",
    "sweep_targets_changed": "Targets changed by the last run.",
    "sweep_targets_failed": "Targets that failed in the last run.",
}
 
 
def record_run(path: Path, status: int, changed: int, failed: int) -> None:
    """Every run stamps last_run; only a clean run moves last_success. Together they tell
    "stopped running" (last_run is old) from "runs and keeps failing" (only last_success is old)."""
    now = int(time.time())
    values = [now, now if status == 0 else read_last_success(path), status, changed, failed]
    write_textfile(path, "".join(f"# HELP {name} {text}\n# TYPE {name} gauge\n{name} {value}\n"
                                 for (name, text), value in zip(SERIES.items(), values)))
 
 
def read_last_success(path: Path) -> int:
    """The last-success time already on disk, so a failed run keeps it instead of resetting it."""
    if path.exists():
        for line in path.read_text().splitlines():
            if line.startswith("sweep_last_success_seconds "):
                return int(line.split()[1])
    return 0

"""The atomic-but-unreadable metric writer: correct rename, wrong mode. Shown to be avoided."""
import os
import sys
import tempfile
from pathlib import Path
 
path = Path(sys.argv[1])
fd, tmp = tempfile.mkstemp(dir=path.parent, prefix="." + path.name + ".")
with os.fdopen(fd, "w") as f:
    f.write("sweep_up 1\n")
os.replace(tmp, path)  # atomic, but the file keeps mkstemp's 0600: the scraper cannot read it

"""A freshness gate over sweep's metric file: is the job running, and is it succeeding?
 
freshness.py --max-age SECONDS FILE
  exit 0  the last run and the last success are both within --max-age
  exit 1  a human must look: no metric file, the job stopped running, or it runs and keeps failing
"""
import argparse
import sys
import time
from pathlib import Path
 
 
def main(argv: list[str]) -> int:
    parser = argparse.ArgumentParser(prog="freshness")
    parser.add_argument("--max-age", type=int, required=True, metavar="SECONDS")
    parser.add_argument("file", type=Path)
    args = parser.parse_args(argv)
 
    if not args.file.exists():
        print(f"freshness: no {args.file}: the job never wrote it, or it writes somewhere else")
        return 1
    samples = dict(line.split() for line in args.file.read_text().splitlines()
                   if line and not line.startswith("#"))
    run_age = time.time() - float(samples["sweep_last_run_seconds"])
    success_age = time.time() - float(samples["sweep_last_success_seconds"])
    if run_age > args.max_age:
        print(f"freshness: last run {run_age:.0f}s ago: the job has stopped running")
        return 1
    if success_age > args.max_age:
        when = "never" if samples["sweep_last_success_seconds"] == "0" else f"{success_age:.0f}s ago"
        print(f"freshness: runs (last exit {samples['sweep_last_exit_code']}) but last success"
              f" {when}: it keeps failing")
        return 1
    print(f"freshness: last run {run_age:.0f}s ago, last success {success_age:.0f}s ago")
    return 0
 
 
if __name__ == "__main__":
    sys.exit(main(sys.argv[1:]))

"""The targets sweep changes. Each host is stood in for by one JSON file under fleet/."""
import json
import os
import time
from pathlib import Path
 
from exit_codes import PermanentError, TransientError
 
 
def read_generation(host: Path) -> int:
    """Ask the target which key generation it runs now: the target is the truth, not a memo."""
    state = json.loads(host.read_text())  # a corrupt file raises JSONDecodeError: no rule for it
    if not state.get("reachable", True):
        raise TransientError("host did not answer")
    if state.get("frozen"):
        raise PermanentError("host is under a change freeze")
    return state["generation"]
 
 
def rotate(host: Path, generation: int, work_seconds: float) -> None:
    """Stand-in for the real change (rotate a key, renew a certificate): slow, then one atomic write."""
    time.sleep(work_seconds)
    tmp = host.with_name(host.name + ".tmp")
    tmp.write_text(json.dumps({"generation": generation}) + "\n")
    os.replace(tmp, host)

"""Bring every target to the wanted generation. Each target's change is final as soon as it is made."""
from dataclasses import dataclass, field
from datetime import datetime, timezone
from pathlib import Path
 
import audit
from exit_codes import PermanentError, TransientError
from jsonlog import get_logger
from targets import read_generation, rotate
 
log = get_logger("sweep")
 
 
@dataclass
class Report:
    ok: int = 0
    changed: int = 0
    permanent: list[str] = field(default_factory=list)
    transient: list[str] = field(default_factory=list)
 
 
def converge(cfg: dict, dry_run: bool, report: Report) -> None:
    want, fleet, trail = cfg["generation"], Path(cfg["fleet"]), Path(cfg["audit"])
    for target in cfg["targets"]:
        try:
            have = read_generation(fleet / f"{target}.json")
        except PermanentError as err:
            report.permanent.append(target)
            log.error("failed", extra={"fields": {"target": target, "result": "permanent", "error": str(err)}})
            continue
        except TransientError as err:
            report.transient.append(target)
            log.warning("failed", extra={"fields": {"target": target, "result": "transient", "error": str(err)}})
            continue
        if have == want:  # converged already, by an earlier run or by hand: a no-op
            report.ok += 1
            log.info("unchanged", extra={"fields": {"target": target, "result": "unchanged"}})
            continue
        if dry_run:
            log.info("would change", extra={"fields": {"target": target, "from": have, "to": want}})
            continue
        # Intent before the change, completion after it: a run that dies in between leaves a
        # "start" with no "done", so the audit trail shows the change that may have happened.
        audit.append(trail, {"ts": _now(), "target": target, "event": "start", "from": have, "to": want})
        rotate(fleet / f"{target}.json", want, cfg.get("work_seconds", 0))
        audit.append(trail, {"ts": _now(), "target": target, "event": "done", "to": want})
        report.ok += 1
        report.changed += 1
        log.info("changed", extra={"fields": {"target": target, "result": "changed", "from": have, "to": want}})
 
 
def _now() -> str:
    return datetime.now(timezone.utc).isoformat(timespec="seconds")

"""One run at a time: an exclusive, non-blocking flock on a file that is never deleted."""
import fcntl
from collections.abc import Iterator
from contextlib import contextmanager
from pathlib import Path
 
from exit_codes import TransientError
 
 
@contextmanager
def single_instance(path: Path) -> Iterator[None]:
    with path.open("a") as f:
        try:
            fcntl.flock(f, fcntl.LOCK_EX | fcntl.LOCK_NB)
        except BlockingIOError:
            raise TransientError(f"another run holds {path}") from None
        yield  # the kernel drops the lock when the file is closed or the process dies, even by SIGKILL

"""An append-only audit trail: one JSON object per line, each carrying the hash of the line before.
 
Editing or deleting a line that has a successor breaks the chain, and verify() finds it. Cutting
lines off the end does not: the shorter file is still a valid chain. Only a copy of the newest
hash kept somewhere else (the lines shipped off the host as they are written) shows that. Whoever
can write the file can also recompute every hash, so the chain raises the bar; it does not seal
the file."""
import fcntl
import hashlib
import json
import os
from pathlib import Path
 
GENESIS = "0" * 64
 
 
def _hash(line: bytes) -> str:
    return hashlib.sha256(line).hexdigest()
 
 
def _last_line(fd: int) -> bytes:
    """Read the last line from the end, so an append does not reread the whole file."""
    end = os.lseek(fd, 0, os.SEEK_END)
    tail = os.pread(fd, min(end, 65536), max(0, end - 65536))  # a record is far below 64 KiB
    return tail.rstrip(b"\n").rsplit(b"\n", 1)[-1]
 
 
def append(path: Path, record: dict) -> None:
    fd = os.open(path, os.O_RDWR | os.O_CREAT | os.O_APPEND, 0o640)
    try:
        # Without the lock, two writers read the same last line and both chain to it. With it,
        # read-last-hash, write and fsync are one step.
        fcntl.flock(fd, fcntl.LOCK_EX)
        last = _last_line(fd)
        entry = {**record, "prev": _hash(last) if last else GENESIS}
        os.write(fd, json.dumps(entry, sort_keys=True).encode() + b"\n")
        os.fsync(fd)  # on disk before the job goes on to the next step
    finally:
        os.close(fd)  # closing the descriptor releases the lock
 
 
def verify(path: Path) -> bool:
    prev = GENESIS
    for raw in path.read_bytes().splitlines():
        if not raw:
            continue
        if json.loads(raw).get("prev") != prev:
            return False
        prev = _hash(raw)
    return True