#!/usr/bin/env python3 # -*- coding: utf-8 -*- """ArchiveBox-Runner-Telemetrie fuer den Knoten archibald. Laeuft als Nutzer `mm` per cron (jede Minute) und schreibt Prometheus-Exposition in eine .prom-Datei, die eine Telegraf-Container-Instanz liest und per Remote Write an den Collector (imac) schickt. Warum ueberhaupt ein Exporteur: ArchiveBox' eigener Healthcheck prueft nur den Webserver, nicht den Runner. Am 03.10.2026 blieb der Runner sieben Tage in einer Neustart-Schleife (uid-Mismatch im Archivbaum, PermissionError beim mkdir des Crawl-Verzeichnisses), waehrend Healthcheck und API gruen waren und jeder neue Snapshot dauerhaft `queued` blieb. Warum die Werte aus dem Container gelesen werden: der Host-Nutzer `mm` darf index.sqlite3 (uid 911) nicht oeffnen; `docker exec` laeuft als Container-User. Die Logdateien dagegen liegen im Host-Dateisystem und sind direkt lesbar. Bewusst KEIN `node`-Label in der Ausgabe -- das setzt Telegraf per [global_tags] (sonst entstehen zwei Serien derselben Identitaet). """ import os import re import subprocess import sys import time ARCHIVEBOX_DIR = "/home/mm/archivebox" OUT_DIR = "/home/mm/abx-telemetrie" OUT_FILE = os.path.join(OUT_DIR, "abx.prom") LOG_FILE = os.path.join(OUT_DIR, "exporteur.log") CONTAINER = "archivebox-archivebox-1" MAX_LOG_BYTES = 512 * 1024 LOGS = os.path.join(ARCHIVEBOX_DIR, "data", "logs") SUPERVISORD_LOG = os.path.join(LOGS, "supervisord.log") ERRORS_LOG = os.path.join(LOGS, "errors.log") # Im Container: Status-Verteilung + Alter des aeltesten offenen Snapshots/Crawls. DB_QUERY = r""" import json, sqlite3, time from datetime import datetime, timezone c = sqlite3.connect("file:/data/index.sqlite3?mode=ro", uri=True) now = time.time() def age_of(value): # ArchiveBox (Django, USE_TZ) speichert created_at als TEXT in UTC. if value is None: return None if isinstance(value, str): try: dt = datetime.fromisoformat(value.replace(" ", "T")) except ValueError: return None if dt.tzinfo is None: dt = dt.replace(tzinfo=timezone.utc) else: dt = value return now - dt.timestamp() out = {} out["snapshots"] = {str(s): int(n) for s, n in c.execute( "select status, count(*) from core_snapshot group by 1")} out["snapshots_total"] = int(c.execute( "select count(*) from core_snapshot").fetchone()[0]) row = c.execute("select min(created_at) from core_snapshot where status!='sealed'").fetchone() out["oldest_open_snapshot_age"] = age_of(row[0] if row else None) # Alarmsignal: ein Snapshot im Zustand `queued` ist NICHT abgeholt. Ein # `started`-Snapshot darf legitim lange laufen (Video-Crawl, gemessen > 1 h) -- # deshalb getrennt fuehren und `queued` explizit als 0 ausgeben, wenn nichts # wartet: eine fehlende Serie waere von "keine Warteschlange" nicht zu # unterscheiden. row = c.execute("select min(created_at) from core_snapshot where status='queued'").fetchone() out["oldest_queued_snapshot_age"] = age_of(row[0] if row else None) or 0 out["crawls"] = {str(s): int(n) for s, n in c.execute( "select status, count(*) from crawls_crawl group by 1")} row = c.execute("select min(created_at) from crawls_crawl where status!='sealed'").fetchone() out["oldest_open_crawl_age"] = age_of(row[0] if row else None) print("JSON:" + json.dumps(out)) """ def sh(args, timeout=20): try: p = subprocess.run(args, capture_output=True, text=True, timeout=timeout) return p.returncode, p.stdout, p.stderr except subprocess.TimeoutExpired: return 124, "", "timeout" except Exception as exc: # noqa: BLE001 return 1, "", str(exc) def line_count(path): try: with open(path, "rb") as fh: return sum(1 for _ in fh) except OSError: return None def stat_or_none(path): try: st = os.stat(path) return st.st_size, st.st_mtime except OSError: return None, None def main(): now = time.time() metrics = [] # (name, wert) ohne Labels labeled = [] # (name, labels-dict, wert) notes = [] # --- Logdateien (Host-Sicht) --- size, mtime = stat_or_none(ERRORS_LOG) if size is not None: metrics.append(("archivebox_runner_errors_log_bytes", size)) lines = line_count(ERRORS_LOG) if lines is not None: metrics.append(("archivebox_runner_errors_log_lines", lines)) metrics.append(("archivebox_runner_errors_log_age_seconds", int(now - mtime))) spawns = None try: with open(SUPERVISORD_LOG, "r", errors="replace") as fh: spawns = fh.read().count("spawned: 'worker_runner'") except OSError: notes.append("supervisord.log nicht lesbar") if spawns is not None: metrics.append(("archivebox_runner_spawns_total", spawns)) # --- Container: laeuft er, laeuft der Runner, was haengt in der Queue --- rc, out, err = sh(["docker", "inspect", "-f", "{{.State.Running}}", CONTAINER], timeout=15) container_up = 1 if (rc == 0 and out.strip() == "true") else 0 metrics.append(("archivebox_container_up", container_up)) runner_up = 0 if container_up: rc, out, _ = sh(["docker", "exec", CONTAINER, "sh", "-c", "pgrep -f 'archivebox run' >/dev/null && echo 1 || echo 0"], timeout=20) runner_up = 1 if out.strip().endswith("1") else 0 metrics.append(("archivebox_runner_up", runner_up)) if container_up: rc, out, err = sh(["docker", "exec", CONTAINER, "python3", "-c", DB_QUERY], timeout=30) payload = None for line in out.splitlines(): if line.startswith("JSON:"): import json payload = json.loads(line[5:]) if payload is None: notes.append("DB-Abfrage fehlgeschlagen: %s" % (err.strip()[:120] or rc)) else: metrics.append(("archivebox_snapshots_total", payload["snapshots_total"])) for status, n in sorted(payload["snapshots"].items()): labeled.append(("archivebox_snapshots", {"status": status}, n)) for status, n in sorted(payload["crawls"].items()): labeled.append(("archivebox_crawls", {"status": status}, n)) age = payload.get("oldest_open_snapshot_age") if age is not None: metrics.append(("archivebox_oldest_open_snapshot_age_seconds", int(age))) metrics.append(("archivebox_oldest_queued_snapshot_age_seconds", int(payload.get("oldest_queued_snapshot_age") or 0))) age = payload.get("oldest_open_crawl_age") if age is not None: metrics.append(("archivebox_oldest_open_crawl_age_seconds", int(age))) else: notes.append("Container laeuft nicht") metrics.append(("archivebox_telemetrie_timestamp_seconds", int(now))) # --- Ausgabe (atomar) --- os.makedirs(OUT_DIR, mode=0o755, exist_ok=True) buf = ["# HELP archivebox_telemetrie_timestamp_seconds Zeitpunkt der Erhebung.", "# TYPE archivebox_telemetrie_timestamp_seconds gauge", "# TYPE archivebox_container_up gauge", "# TYPE archivebox_runner_up gauge", "# TYPE archivebox_runner_spawns_total counter", "# TYPE archivebox_snapshots gauge", "# TYPE archivebox_crawls gauge"] for name, val in metrics: buf.append("%s %s" % (name, val)) for name, labels, val in labeled: lbl = ",".join('%s="%s"' % (k, v) for k, v in sorted(labels.items())) buf.append("%s{%s} %s" % (name, lbl, val)) tmp = OUT_FILE + ".tmp" with open(tmp, "w") as fh: fh.write("\n".join(buf) + "\n") os.chmod(tmp, 0o644) os.replace(tmp, OUT_FILE) # --- Laufprotokoll (gedeckelt) --- try: if os.path.exists(LOG_FILE) and os.path.getsize(LOG_FILE) > MAX_LOG_BYTES: os.replace(LOG_FILE, LOG_FILE + ".1") with open(LOG_FILE, "a") as fh: fh.write("%s ok container=%d runner=%d spawns=%s %s\n" % ( time.strftime("%Y-%m-%dT%H:%M:%S"), container_up, runner_up, spawns, ("; ".join(notes) if notes else ""))) except OSError: pass if notes: print("Hinweise: " + "; ".join(notes), file=sys.stderr) return 0 if __name__ == "__main__": sys.exit(main())