fix(borg): update roles/borg/files/borg-sentinel-push.py

roles/borg/files/borg-sentinel-push.py

@@ -0,0 +1,96 @@
+#!/usr/bin/env python3
+"""Send latest Borg archive statistics to Backup Sentinel."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import subprocess
+import sys
+import urllib.error
+import urllib.request
+from pathlib import Path
+from typing import Any
+
+DEFAULT_URL = "http://10.1.1.111:9999/api/push"
+
+
+def load_borg_info(info_file: str | None) -> Any:
+    if info_file:
+        return json.loads(Path(info_file).read_text())
+
+    result = subprocess.run(
+        ["borgmatic", "info", "--archive", "latest", "--json"],
+        check=True,
+        text=True,
+        capture_output=True,
+    )
+    return json.loads(result.stdout)
+
+
+def latest_archive(info: Any) -> dict[str, Any]:
+    if not isinstance(info, list) or not info or not isinstance(info[0], dict):
+        raise ValueError("Unexpected borgmatic info JSON: expected a non-empty repository list")
+    archives = info[0].get("archives")
+    if not isinstance(archives, list) or not archives or not isinstance(archives[-1], dict):
+        raise ValueError("Unexpected borgmatic info JSON: no latest archive found")
+    return archives[-1]
+
+
+def build_payload(host: str, archive: dict[str, Any]) -> dict[str, Any]:
+    stats = archive.get("stats")
+    if not isinstance(stats, dict):
+        raise ValueError("Unexpected borgmatic info JSON: latest archive has no stats")
+
+    return {
+        "host": host,
+        "status": "ok",
+        # Borg exposes duration on the archive object, not below stats.
+        "duration_sec": int(round(float(archive.get("duration", 0)))),
+        "original_size": int(stats.get("original_size", 0)),
+        "deduplicated_size": int(stats.get("deduplicated_size", 0)),
+        "compressed_size": int(stats.get("compressed_size", 0)),
+        "nfiles_new": int(stats.get("nfiles", 0)),
+        "archive": str(archive.get("name", "")),
+    }
+
+
+def post_payload(url: str, payload: dict[str, Any]) -> None:
+    request = urllib.request.Request(
+        url,
+        data=json.dumps(payload, separators=(",", ":")).encode(),
+        method="POST",
+        headers={"Content-Type": "application/json"},
+    )
+    with urllib.request.urlopen(request, timeout=10) as response:
+        body = response.read().decode(errors="replace")
+        if response.status < 200 or response.status >= 300:
+            raise RuntimeError(f"Sentinel returned HTTP {response.status}: {body[:500]}")
+        print(body)
+
+
+def parse_args() -> argparse.Namespace:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--host", required=True)
+    parser.add_argument("--url", default=DEFAULT_URL)
+    parser.add_argument("--info-file", help="Read borgmatic info JSON from a fixture instead of running borgmatic")
+    parser.add_argument("--dry-run", action="store_true", help="Print payload without posting it")
+    return parser.parse_args()
+
+
+def main() -> int:
+    args = parse_args()
+    try:
+        payload = build_payload(args.host, latest_archive(load_borg_info(args.info_file)))
+        if args.dry_run:
+            print(json.dumps(payload, separators=(",", ":")))
+        else:
+            post_payload(args.url, payload)
+        return 0
+    except (ValueError, OSError, subprocess.SubprocessError, json.JSONDecodeError, urllib.error.URLError) as exc:
+        print(f"borg-sentinel-push: {exc}", file=sys.stderr)
+        return 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())