fix: Output-Timeouts 10s->60s; docker-events-Watcher fuer Docker-Nodes
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.
This commit is contained in:
Kalina committed
1 parent
1fe25163d5
commit
63bc054371
4 files changed
+322
-2
No files matched your search
@@ -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())
|
||||
@@ -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
|
||||
Reference in new issue
Block a user