Kalina
63bc054371
Die 10-s-Timeouts waren kuerzer als die gemessene Zustellzeit (11-19 s durch den NetBird-Reverse-Proxy auf x05) -> jeder Batch brach ab. Gemessen: 702 abgelehnte Requests in 20 min, Metriken fehlten still. Neu: scripts/docker-events-loki.py + Wrapper + systemd-Unit. Docker-Events (Neustart/OOM/Stop) gehen nach Loki, Spool-Datei + Cursor statt RAM-Puffer.
277 lines
10 KiB
Python
277 lines
10 KiB
Python
#!/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())
|