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