netbird-telegraf-installer/scripts/archivebox-telemetrie.py

#!/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())