96 lines
3.3 KiB
Python
96 lines
3.3 KiB
Python
#!/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())
|