threadnet-operating/monitoring/alertmanager/matrix-alerts.py

#!/usr/bin/env python3
# Minimaler Alertmanager-Webhook -> Matrix-Raum (gitops#32). Gleiche Machart wie
# maintenance-notify (Issue #24 im gitops-Repo): purer Stdlib-HTTP-Server, Bot-Token
# aus der Umgebung, Nachricht per Client-Server-API.
#
# Ein Alarm = EINE Nachricht: beim Firing wird pro Alarm (Alertmanager-Fingerprint)
# eine Nachricht gesendet und deren Event-ID gemerkt; beim Resolved wird dieselbe
# Nachricht per m.replace-Edit durchgestrichen und abgehakt statt eine neue zu
# posten (Wunsch sorb 2026-08-01: append-only wird unuebersichtlich). Die Zuordnung
# ueberlebt Container-Restarts via State-Datei; ohne Zuordnung (z.B. nach Neubau)
# faellt Resolved auf eine eigenstaendige ✅-Nachricht zurueck.
import html
import json
import os
import time
import urllib.parse
import urllib.request
import uuid
from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer

HOMESERVER = os.environ["MATRIX_HOMESERVER"]
ROOM_ID = os.environ["MATRIX_ROOM_ID"]
TOKEN = os.environ["MATRIX_TOKEN"]
STATE_FILE = os.environ.get("MATRIX_STATE_FILE", "/tmp/matrix-alerts-state.json")
# Raum-Routing (gitops#47): Alerts mit Label room="<name>" landen im Raum aus
# MATRIX_ROOM_<NAME> (z.B. room="security" -> MATRIX_ROOM_SECURITY), alles
# andere im Default-Raum. Der Bot muss in jedem Zielraum Mitglied sein.
ROOM_MAP = {k[len("MATRIX_ROOM_"):].lower(): v
            for k, v in os.environ.items()
            if k.startswith("MATRIX_ROOM_") and k != "MATRIX_ROOM_ID" and v}

try:
    with open(STATE_FILE) as f:
        _persisted = json.load(f)
    state = _persisted.get("alerts", _persisted)  # alt: flaches Format
    recent_resolved = _persisted.get("recent_resolved", [])
except Exception:
    state = {}
    recent_resolved = []


def save_state():
    try:
        with open(STATE_FILE, "w") as f:
            json.dump({"alerts": state, "recent_resolved": recent_resolved[-200:]}, f)
    except Exception as e:
        print(f"state save failed: {e}", flush=True)


def matrix_put(content, room=ROOM_ID):
    url = (f"{HOMESERVER}/_matrix/client/v3/rooms/{urllib.parse.quote(room)}"
           f"/send/m.room.message/{uuid.uuid4()}")
    req = urllib.request.Request(url, data=json.dumps(content).encode(), method="PUT",
                                 headers={"Authorization": f"Bearer {TOKEN}",
                                          "Content-Type": "application/json"})
    return json.loads(urllib.request.urlopen(req, timeout=10).read()).get("event_id")


def send_text(text, room=ROOM_ID):
    return matrix_put({"msgtype": "m.text", "body": text}, room)


def edit_resolved(event_id, old_text, room=ROOM_ID):
    # m.replace-Edit: Original wird in Element in-place ersetzt (durchgestrichen + Haken)
    new_body = f"✅ ~~{old_text}~~"
    new_html = f"✅ <del>{html.escape(old_text)}</del>"
    matrix_put({
        "msgtype": "m.text",
        "body": f"* {new_body}",
        "m.new_content": {"msgtype": "m.text", "body": new_body,
                          "format": "org.matrix.custom.html", "formatted_body": new_html},
        "m.relates_to": {"rel_type": "m.replace", "event_id": event_id},
    }, room)


class Handler(BaseHTTPRequestHandler):
    def do_POST(self):
        if self.path != "/alert":
            self.send_response(404); self.end_headers(); return
        data = json.loads(self.rfile.read(int(self.headers.get("Content-Length", 0))))
        failed = 0
        for a in data.get("alerts", []):
            name = a.get("labels", {}).get("alertname", "?")
            summary = a.get("annotations", {}).get("summary", "")
            fp = a.get("fingerprint", "")
            room = ROOM_MAP.get(a.get("labels", {}).get("room", ""), ROOM_ID)
            try:
                if a.get("status") == "firing":
                    if fp in state:  # re-notify/Batch-Retry -> keine Doppelnachricht
                        continue
                    text = f"\U0001F534 [firing] {name}: {summary}"
                    event_id = send_text(text, room)
                    if fp and event_id:
                        state[fp] = {"event_id": event_id, "text": text, "room": room}
                else:
                    if fp in recent_resolved:  # Batch-Retry -> Fallback nicht doppeln
                        continue
                    known = state.pop(fp, None)
                    if known:
                        edit_resolved(known["event_id"], known["text"], known.get("room", ROOM_ID))
                    else:
                        send_text(f"✅ [resolved] {name}: {summary}", room)
                    recent_resolved.append(fp)
                save_state()  # inkrementell: Teilfortschritt uebersteht Fehler/Retry
                time.sleep(1)  # Synapse-Ratelimit (rc_message) nicht reizen
            except Exception as e:
                failed += 1
                self.log_message("matrix send failed (%s): %s", name, e)
                save_state()
                time.sleep(2)
        if failed:
            # Alertmanager wiederholt den Batch; Dedup liefert nur den Rest nach
            self.send_response(502); self.end_headers(); return
        self.send_response(200); self.end_headers()

    def log_message(self, fmt, *args):
        print(fmt % args, flush=True)


ThreadingHTTPServer(("0.0.0.0", 8080), Handler).serve_forever()