Media-Server-Setup/observability/kibana-alarms.py

#!/usr/bin/env python3
"""The real alarm rules. Idempotent: creates what is missing, leaves what exists.

Every rule is an Elasticsearch query rule that counts matching documents in a window and writes one document
into the alert bus through the "Vanguard alert index" connector, which bin/vanguard-alerts.sh delivers to
Discord. Run observability/kibana-setup.py first — it creates that connector.

FIRING CADENCE: notify_when=onThrottleInterval with throttle=12h, so an alarm that stays true writes again
every 12 hours and stops when the condition clears. The delivery script's own 15-minute per-rule suppression
sits well inside that and never interferes.

WHAT IS DELIBERATELY NOT HERE
Elasticsearch, Kibana and Beat liveness, and the disks, are bin/vanguard-deadman.sh's job. Nothing that
depends on Kibana being alive can report that Kibana is dead, so those checks stay outside the stack and
duplicating them here would be worse than useless: they would go quiet in exactly the case that matters.

  pw=$(grep -m1 '^ES_ELASTIC_PASSWORD=' .env | cut -d= -f2-)
  KEY=$(curl -s -XPOST -u "elastic:$pw" localhost:9200/_security/api_key \
          -H 'Content-Type: application/json' -d '{"name":"vanguard-cli"}' \
        | python3 -c 'import json,sys; print(json.load(sys.stdin)["encoded"])')
  KEY="$KEY" python3 observability/kibana-alarms.py

To change a rule, edit it in Kibana or delete it there and re-run this. Thresholds here are the defaults:
  FREE_PCT        15   alarm when a real filesystem or memory has less than this percent free
  STUCK_SECONDS  3600  alarm when a finished recording has waited this long to be normalised
  THROTTLE        12h  how often an unresolved alarm repeats
"""
import json, os, urllib.error, urllib.request

KIBANA = "http://127.0.0.1:5601"
HEADERS = {"Content-Type": "application/json", "kbn-xsrf": "1",
           "Authorization": "ApiKey " + os.environ["KEY"]}
CONNECTOR_NAME = "Vanguard alert index"

FREE_PCT = float(os.environ.get("FREE_PCT", "15"))
STUCK_SECONDS = int(os.environ.get("STUCK_SECONDS", "3600"))
THROTTLE = os.environ.get("THROTTLE", "12h")
USED_MAX = round(1 - FREE_PCT / 100, 4)          # 15% free -> used.pct >= 0.85

# Filesystems that are not disks. Without this, every tmpfs and overlay mount reports ~100% used and the
# disk alarm would fire permanently on something that cannot fill up.
PSEUDO_FS = ["tmpfs", "devtmpfs", "overlay", "squashfs", "ramfs", "autofs", "nsfs", "tracefs", "debugfs",
             "configfs", "mqueue", "hugetlbfs", "pstore", "bpf", "devpts", "proc", "sysfs", "cgroup",
             "cgroup2", "fusectl", "securityfs", "efivarfs", "binfmt_misc", "fuse.lxcfs", "iso9660"]

def rule(name, level, why, query, *, index="metricbeat-*", window=10, comparator=">", threshold=0):
    return {"name": name, "level": level, "why": why, "query": query,
            "index": index, "window": window, "comparator": comparator, "threshold": threshold}

ALARMS = [
    rule("Disk free below %g%%" % FREE_PCT, "error",
         "A filesystem is nearly full. Elasticsearch turns its indices READ-ONLY at the flood watermark and "
         "does not undo that by itself, and recordings stop being published.",
         {"query": {"bool": {
             "filter": [{"range": {"system.filesystem.used.pct": {"gte": USED_MAX}}}],
             "must_not": [{"terms": {"system.filesystem.type": PSEUDO_FS}}]}}}),

    rule("Memory free below %g%%" % FREE_PCT, "warn",
         "Memory is nearly exhausted. Elasticsearch has a pinned 2 GB heap, so the pressure is usually "
         "ffmpeg or Jellyfin, and the kernel will start killing processes rather than queueing.",
         {"query": {"range": {"system.memory.actual.used.pct": {"gte": USED_MAX}}}}),

    rule("Discord bot not running", "error",
         "No metrics from the discord-bot container for 5 minutes: it is stopped, removed, or restarting too "
         "fast to be sampled. Nothing it does is happening.",
         {"query": {"term": {"container.name": "discord-bot"}}},
         window=5, comparator="<", threshold=1),

    rule("Discord bot unhealthy", "error",
         "The bot is running but reports itself degraded: its /health says the Discord gateway is "
         "disconnected or the database does not answer. It is up and doing nothing.",
         {"query": {"bool": {"filter": [{"term": {"container.name": "discord-bot"}},
                                        {"term": {"docker.healthcheck.status": "unhealthy"}}]}}},
         window=5),

    rule("Recordings stuck waiting to be normalised", "warn",
         "A finished recording has waited over %d minutes in .live/. It is not in Jellyfin, not in copyparty "
         "and not announced, so to everyone else it does not exist." % (STUCK_SECONDS // 60),
         {"query": {"range": {"prometheus.vanguard_media_pending_oldest_seconds.value": {"gt": STUCK_SECONDS}}}}),

    rule("Normalise failures parked", "warn",
         "One or more files have a normalize-failed log, so the sweep skips them forever until it is deleted. "
         "The log now contains ffmpeg's stderr, which says why.",
         {"query": {"range": {"prometheus.vanguard_media_failed_files.value": {"gt": 0}}}}),
]

def req(method, path, body=None):
    request = urllib.request.Request(KIBANA + path, method=method,
                                     data=None if body is None else json.dumps(body).encode())
    for key, value in HEADERS.items():
        request.add_header(key, value)
    try:
        with urllib.request.urlopen(request, timeout=30) as response:
            raw = response.read()
            return json.loads(raw) if raw else {}
    except urllib.error.HTTPError as error:
        print(f"  {method} {path} -> HTTP {error.code}")
        print("  " + error.read().decode(errors="replace")[:600])
        raise SystemExit(1)

connectors = {c["name"]: c["id"] for c in req("GET", "/api/actions/connectors")}
if CONNECTOR_NAME not in connectors:
    raise SystemExit(f"connector {CONNECTOR_NAME!r} is missing — run observability/kibana-setup.py first")
connector_id = connectors[CONNECTOR_NAME]

present = {r["name"] for r in req("GET", "/api/alerting/rules/_find?per_page=100").get("data", [])}
created = skipped = 0
for alarm in ALARMS:
    if alarm["name"] in present:
        print("already present:", alarm["name"])
        skipped += 1
        continue
    made = req("POST", "/api/alerting/rule", {
        "name": alarm["name"],
        "rule_type_id": ".es-query",
        "consumer": "stackAlerts",
        "enabled": True,
        "schedule": {"interval": "1m"},
        "params": {
            "searchType": "esQuery",
            "index": [alarm["index"]],
            "timeField": "@timestamp",
            "esQuery": json.dumps(alarm["query"]),
            "size": 1,
            "aggType": "count",
            "groupBy": "all",
            "thresholdComparator": alarm["comparator"],
            "threshold": [alarm["threshold"]],
            "timeWindowSize": alarm["window"],
            "timeWindowUnit": "m",
            "excludeHitsFromPreviousRun": False,
        },
        "actions": [{
            "group": "query matched",
            "id": connector_id,
            "params": {"documents": [{
                "kind": "alarm",
                "rule": "{{rule.name}}",
                "level": alarm["level"],
                "message": alarm["why"] + " [{{context.message}}]",
            }]},
            # every 12 h for as long as it stays true, then silence when it clears
            "frequency": {"summary": False, "notify_when": "onThrottleInterval", "throttle": THROTTLE},
        }],
    })
    print("created:", made["name"], "|", alarm["level"], "| repeats every", THROTTLE)
    created += 1

print(f"\n{created} created, {skipped} already there. Disk/memory threshold {FREE_PCT:g}% free; "
      f"stuck after {STUCK_SECONDS}s.")