diff --git a/scripts/docker-events-loki.py b/scripts/docker-events-loki.py new file mode 100644 index 0000000..cdb1f51 --- /dev/null +++ b/scripts/docker-events-loki.py @@ -0,0 +1,276 @@ +#!/usr/bin/env python3 +"""docker-events-loki.py — Docker-Events (Neustart, OOM-Kill, Stop, Health) +dauerhaft nach Loki. Damit ist "wer hat den Container neu gestartet" in +Grafana beantwortet statt in Rätselraten. + +ARCHITEKTUR (nach zwei echten Fehlern in der ersten Fassung): + * Docker hält Events nur im Speicher. Ein Poller mit `--since` verliert bei + Ausfällen alles dazwischen. Deshalb ein dauerhafter `docker events`-Stream. + * Ein RAM-Puffer verliert Events, wenn Loki nicht mithält. GEMESSEN auf x05: + ein POST über den NetBird-Reverse-Proxy braucht 11-19 s (TLS-Handshake + allein nur 0,5-1,6 s, Loki direkt 1,3 ms) — der Weg ist der Flaschenhals. + Mit RAM-Puffer und 5-s-Flush lief die Queue voll und die erste Fassung + VERWARF Events. Bei einem Audit-Stream ist Verwerfen inakzeptabel. + * Daher: jedes Event geht zuerst in eine Spool-DATEI (append, O_APPEND). + Der Sender arbeitet die Datei ab und trunkiert nach bestätigtem Push. + Loki-Ausfall oder langsamer Proxy kosten damit nur Verzögerung, keine Daten. + * Die Spool-Größe ist begrenzt; bei Überlauf werden die ÄLTESTEN verworfen + und laut gemeldet (auf diesem Host mit 967 MiB RAM darf ein Log-Forwarder + nie die Ursache des nächsten OOM werden). + +ENV (aus /opt/observability-telegraf/.env): + LOKI_URL, TELEGRAF_TOKEN, optional NODE_NAME +""" +import json +import os +import subprocess +import sys +import threading +import time +import urllib.error +import urllib.request + +STATE_DIR = os.environ.get("DOCKER_EVENTS_STATE", "/var/lib/docker-events") +LOKI_URL = os.environ.get("LOKI_URL", "").rstrip("/") +TOKEN = os.environ.get("TELEGRAF_TOKEN", "") +NODE = os.environ.get("NODE_NAME") or os.uname().nodename + +SPOOL = os.path.join(STATE_DIR, "spool.jsonl") +CURSOR = os.path.join(STATE_DIR, "cursor") +HEALTH_FILE = os.path.join(STATE_DIR, "heartbeat") +SPOOL_MAX_BYTES = 20 * 1024 * 1024 # ~50k Events; danach älteste weg +BATCH_LINES = 50 +BATCH_SECONDS = 10.0 +PUSH_TIMEOUT = 120 # Proxy-Weg braucht 11-19 s (gemessen) +IDLE_SLEEP = 2.0 + + +def log(msg): + print(f"{time.strftime('%Y-%m-%dT%H:%M:%S')} {msg}", flush=True) + + +def push(lines): + """lines: Liste von (ns_string, message_string, labels_dict)""" + if not lines: + return True + streams = {} + for ns, msg, labels in lines: + key = tuple(sorted(labels.items())) + streams.setdefault(key, {"stream": dict(labels), "values": []}) + streams[key]["values"].append([ns, msg]) + body = json.dumps({"streams": list(streams.values())}).encode() + req = urllib.request.Request( + f"{LOKI_URL}/loki/api/v1/push", data=body, method="POST", + headers={"Content-Type": "application/json", + "Authorization": f"Bearer {TOKEN}"}) + try: + with urllib.request.urlopen(req, timeout=PUSH_TIMEOUT) as r: + return 200 <= r.status < 300 + except urllib.error.HTTPError as e: + detail = "" + try: + detail = e.read()[:200].decode(errors="replace") + except Exception: + pass + log(f"LOKI HTTP {e.code}: {detail!r}") + # 4xx sind dauerhaft -> Batch verwerfen, sonst blockiert er ewig. + return "permanent" if 400 <= e.code < 500 else False + except Exception as e: + log(f"LOKI nicht erreichbar ({type(e).__name__}): {e}") + return False + + +def spool_append(rec): + """Ein Event atomar anhängen. O_APPEND: kein Verlust bei gleichzeitigem Lesen.""" + line = json.dumps(rec, ensure_ascii=False) + "\n" + fd = os.open(SPOOL, os.O_WRONLY | os.O_CREAT | os.O_APPEND, 0o600) + try: + os.write(fd, line.encode()) + finally: + os.close(fd) + + +def spool_rotate_if_big(): + """Zu groß? Die ÄLTESTEN verwerfen (neue Events sind die wertvollen).""" + try: + size = os.path.getsize(SPOOL) + except OSError: + return + if size <= SPOOL_MAX_BYTES: + return + keep = SPOOL_MAX_BYTES // 2 + try: + with open(SPOOL, "rb") as fh: + fh.seek(-keep, os.SEEK_END) + tail = fh.read() + # an der ersten vollständigen Zeile abschneiden + nl = tail.find(b"\n") + if nl >= 0: + tail = tail[nl + 1:] + tmp = SPOOL + ".rot" + with open(tmp, "wb") as fh: + fh.write(tail) + os.replace(tmp, SPOOL) + log(f"WARNUNG: Spool war {size/1e6:.1f} MB, aelteste Haelfte verworfen") + except OSError as e: + log(f"WARNUNG: Spool-Rotation fehlgeschlagen: {e}") + + +def collect_thread(): + """docker events -> Spool. Läuft nie blockierend, damit der Kernel-Puffer + nicht überläuft (sonst verliert Docker selbst Events).""" + while True: + try: + proc = subprocess.Popen( + ["docker", "events", "--format", "{{json .}}"], + stdout=subprocess.PIPE, stderr=subprocess.DEVNULL, + text=True, bufsize=1) + except FileNotFoundError: + log("FEHLER: docker nicht gefunden — warte 30 s") + time.sleep(30) + continue + + assert proc.stdout is not None + log("docker-events-Stream verbunden") + for raw in proc.stdout: + raw = raw.strip() + if not raw: + continue + try: + ev = json.loads(raw) + except json.JSONDecodeError: + continue + + ts = int(ev.get("timeNano") or ev.get("time") or 0) + if ts and ts < 10**12: # war in Sekunden + ts *= 10**9 + if not ts: + ts = int(time.time() * 10**9) + + act = ev.get("Action") or ev.get("status") or "unknown" + actor = ev.get("Actor") or {} + attrs = actor.get("Attributes") or {} + name = attrs.get("name") or str(ev.get("id", ""))[:12] or "?" + + try: + spool_append({ + "ns": str(ts), + "labels": { + "node": NODE, + "source": "docker-events", + "action": act.replace(" ", "_")[:40], + "name": name[:60], + }, + "msg": json.dumps({ + "action": act, "name": name, + "image": str(attrs.get("image", ""))[:80], + "exitCode": attrs.get("exitCode", ""), + "oomKilled": attrs.get("oomKilled", ""), + "signal": attrs.get("signal", ""), + "time": time.strftime("%Y-%m-%dT%H:%M:%S", + time.gmtime(ts / 1e9)), + }, ensure_ascii=False), + }) + except OSError as e: + log(f"FEHLER beim Spool-Schreiben: {e}") + + log("WARNUNG: docker-events-Stream endete — Neustart in 10 s") + time.sleep(10) + + +def send_thread(): + """Spool -> Loki über einen Cursor. + + Der Cursor (Byte-Offset) wird NACH bestätigtem Push gespeichert. Damit + gibt es kein Umschreiben der Datei und kein Rennen mit dem Collector: + der Collector hängt nur an, der Sender liest nur vorwärts. + at-least-once: nach einem Absturz wird der letzte Batch erneut gesendet. + Für ein Audit-Log ist ein doppeltes Event harmlos, ein fehlendes nicht. + """ + while True: + try: + if not os.path.exists(SPOOL): + time.sleep(IDLE_SLEEP) + continue + + try: + with open(CURSOR) as fh: + offset = int(fh.read().strip() or 0) + except (OSError, ValueError): + offset = 0 + + size = os.path.getsize(SPOOL) + if offset > size: # Spool wurde rotiert + offset = 0 + if offset == size: + spool_rotate_if_big() + time.sleep(IDLE_SLEEP) + continue + + with open(SPOOL, "rb") as fh: + fh.seek(offset) + chunk = fh.read(SPOOL_MAX_BYTES // 8) + + # nur vollständige Zeilen verarbeiten + cut = chunk.rfind(b"\n") + if cut < 0: + time.sleep(IDLE_SLEEP) + continue + complete = chunk[:cut + 1] + + batch = [] + for ln in complete.decode(errors="replace").splitlines()[:BATCH_LINES]: + try: + rec = json.loads(ln) + batch.append((rec["ns"], rec["msg"], rec["labels"])) + except (json.JSONDecodeError, KeyError): + continue + + # Offset bis zum Ende des verarbeiteten Teils + used = 0 + for ln in complete.decode(errors="replace").splitlines()[:BATCH_LINES]: + used += len(ln.encode()) + 1 + + if not batch: + offset += used + with open(CURSOR, "w") as fh: + fh.write(str(offset)) + continue + + res = push(batch) + if res is True or res == "permanent": + offset += used + tmp = CURSOR + ".tmp" + with open(tmp, "w") as fh: + fh.write(str(offset)) + os.replace(tmp, CURSOR) + try: + with open(HEALTH_FILE, "w") as fh: + fh.write(str(int(time.time()))) + except OSError: + pass + if res == "permanent": + log(f"{len(batch)} Ereignisse dauerhaft abgelehnt, verworfen") + else: + time.sleep(IDLE_SLEEP * 2) # Loki down: kurz warten + except Exception as e: + log(f"FEHLER im Sender ({type(e).__name__}): {e}") + time.sleep(5) + + +def main(): + if not LOKI_URL or not TOKEN: + log("FEHLER: LOKI_URL oder TELEGRAF_TOKEN fehlt") + return 2 + os.makedirs(STATE_DIR, exist_ok=True) + log(f"starte docker-events -> {LOKI_URL} (node={NODE})") + + threading.Thread(target=collect_thread, daemon=True).start() + threading.Thread(target=send_thread, daemon=True).start() + while True: + time.sleep(3600) + return 0 + + +if __name__ == "__main__": + sys.exit(main()) diff --git a/scripts/docker-events-watcher.sh b/scripts/docker-events-watcher.sh new file mode 100644 index 0000000..f163b1f --- /dev/null +++ b/scripts/docker-events-watcher.sh @@ -0,0 +1,14 @@ +#!/bin/bash +# docker-events-watcher.sh — Wrapper fuer docker-events-loki.py (systemd). +# +# Liest LOKI_URL und TELEGRAF_TOKEN aus der .env des Telegraf-Stacks, damit +# keine zweite Credential-Datei entsteht. Pfad per ENV uebersteuerbar. +set -u +ENVF="${DOCKER_EVENTS_ENV:-/opt/observability-telegraf/.env}" +if [ -r "$ENVF" ]; then + set -a + # shellcheck disable=SC1090 + . "$ENVF" + set +a +fi +exec /usr/bin/python3 /usr/local/bin/docker-events-loki.py diff --git a/telegraf/base.conf b/telegraf/base.conf index 562b879..1ae4007 100644 --- a/telegraf/base.conf +++ b/telegraf/base.conf @@ -27,7 +27,10 @@ [[outputs.http]] url = "${PROM_URL}" data_format = "prometheusremotewrite" - timeout = "10s" + # 60s, NICHT 10s: laeuft die Node hinter dem NetBird-Reverse-Proxy (x05), + # dauert ein POST 11-19 s. Ein 10-s-Timeout bricht dann JEDEN Batch ab — + # gemessen 702 abgelehnte Requests in 20 min, die Metriken fehlen still. + timeout = "60s" # Roh-Logs gehoeren NICHT in Prometheus — sie sind Text, keine Zeitreihen. # `apache_error` hier ergaenzen, sonst versucht der Remote-Write-Output # Textzeilen als Metriken zu senden. @@ -42,7 +45,8 @@ [[outputs.loki]] domain = "${LOKI_URL}" endpoint = "/loki/api/v1/push" - timeout = "10s" + # 60s aus demselben Grund wie oben: der Proxy-Weg braucht 11-19 s. + timeout = "60s" # `apache_error` MUSS hier stehen: ohne den Namen sammelt der Input die # Error-Logs ein und dieser Output verwirft sie wieder — kein Fehler, keine # Meldung, die Panels bleiben leer. Derselbe stille Fehler wie bei `journal`. diff --git a/telegraf/docker-events-loki.service b/telegraf/docker-events-loki.service new file mode 100644 index 0000000..c699e33 --- /dev/null +++ b/telegraf/docker-events-loki.service @@ -0,0 +1,26 @@ +[Unit] +Description=Docker-Events nach Loki (Neustart/OOM/Stop sichtbar machen) +Documentation=file:///usr/local/bin/docker-events-loki.py +After=network-online.target docker.service +Wants=network-online.target +Requires=docker.service + +[Service] +Type=simple +Restart=always +RestartSec=10 +# Kein Timeout: der Prozess ist ein Stream, kein Job. +TimeoutStopSec=15 +ExecStart=/usr/local/bin/docker-events-watcher.sh +# Schutz: der Forwarder darf nie die Ursache des naechsten OOM werden. +# Auf Nodes mit wenig RAM (x05: 967 MB) ist das keine Formalie — dort hat ein +# OOM-Kill bereits netbird-proxy erwischt. +MemoryMax=128M +MemorySwapMax=0 +Nice=10 +StandardOutput=journal +StandardError=journal +SyslogIdentifier=docker-events-loki + +[Install] +WantedBy=multi-user.target