Elastic security, our own metrics, and an alert pipeline that can actually reach Discord

Observability was running but could not report anything. This closes that end to end and makes every piece of it reproducible from the installers instead of from a shell history. Logs - Filebeat 9 REMOVED the container input; it refuses the config per container and then runs with zero harvesters and no index. Container logs now use filestream + the container parser with a stable id each, and add_docker_metadata is told where the id is. - Every line gets event.dataset, or Kibana's Logs rate chart is one "unknown" bucket. - Beats pin replicas to 0: a single node cannot place one, so the cluster sat yellow forever. Security - xpack.security.enabled=true, because Kibana's Observability apps ask Elasticsearch whether the user holds a privilege and that API does not exist when security is off (the Infrastructure page failed on every view). Nobody sees a login: Kibana's anonymous provider signs every visitor in as one native user, and Caddy + Authentik + DIAG_REQUIRE_GROUP stays the only gate. configure.py generates the service passwords, creates the accounts and writes a root-only netrc for the deadman; --es-security retries just that step. Alerting - A basic licence has no webhook connector (403, Gold+), so Kibana's rules write documents into a vanguard-alerts index through the free .index connector and vanguard-alerts.timer delivers them to Discord. Alerts are therefore searchable beside the logs that caused them. - Six rules in observability/kibana-alarms.py: disk and memory below 15% free, bot not running, bot unhealthy, recordings stuck over an hour, normalise failures parked. Each repeats every 12 h while unresolved. - The deadman watches both halves of that pipeline, since nothing inside the stack can report that the stack's own alerting died. Our own metrics - bin/vgmetrics.sh counters from the sweep and vgdeband, plus live gauges on dumpd /metrics (loopback only, unrouted by Caddy), scraped by Metricbeat. vanguard_media_pending is the measured answer to whether normalising keeps up, which was previously only a benchmark guess. - The sweep now captures ffmpeg's stderr: rc= alone was not diagnosable. - Its file loop reads a process substitution, not a pipe, or the subshell discards every counter. Housekeeping - configure.py --apply installs the media scripts (it never did, so applies left stale binaries), creates the alert index, injects an ILM delete phase and runs Kibana's first-run setup. - The discord-bot container asks the bot about itself via /health rather than probing a port. - Removed the dead diag landing page: Kibana occupies the root of that host. Co-Authored-By: Claude Opus 5 (1M context) <noreply@anthropic.com>
2026-11-03 19:58:06 -06:00
parent ef357c6d82
commit b38d40ab0c
20 changed files with 1322 additions and 110 deletions
--- a/bin/dumpd.py
+++ b/bin/dumpd.py
@@ -12,12 +12,17 @@ POST /me/clips  -> {"device_id": "..."}: create clips/<user> (+ .stignore), add
 POST /deploy    -> Gitea push webhook (bridge port only, HMAC-verified with DEPLOY_WEBHOOK_SECRET from
             /etc/vanguard/deploy.env). Runs /usr/local/bin/vanguard-deploy pull ONLY when DEPLOY_AUTO=1;
             otherwise logs the push and answers 202 "auto-deploy disabled".
-POST /heartbeat  -> Alertmanager's Watchdog route: touches a file the deadman timer checks, so a dead
-                   alerting pipeline becomes visible (bin/vanguard-deadman.sh).
+POST /heartbeat  -> LEGACY, nothing calls this now. It was the Watchdog route when Alertmanager, and then a
+                   Kibana webhook connector, posted here. A basic Elastic licence has no webhook connector, so
+                   Kibana's rules write into the vanguard-alerts index instead and the deadman reads the
+                   heartbeat from there. Kept because it costs nothing and is a ready-made target if an
+                   external monitor is ever added.
 POST /deploy-bot -> Gitea Actions (job container -> host gateway) hands over a Discord-bot release bundle,
                   HMAC-SHA256 signed with BOT_DEPLOY_SECRET; installs it with vanguard-deploy-bot.
                   May carry X-Vanguard-Env (base64 JSON): the bot's own settings, kept in Gitea, written
                   into the stack .env for the keys BOT_ENV_KEYS allows. Signature covers header+body.
+GET  /metrics -> Prometheus text for Metricbeat: the media scripts' counters (bin/vgmetrics.sh) plus live
+             gauges for the normalise queue and MediaMTX. Loopback listener only; Caddy routes nothing here.
 POST /mtx-auth -> MediaMTX authHTTP hook: allows read/playback; allows publish iff user/pass carries a valid
             token whose user owns the path prefix. Tokens are HMAC-signed with /etc/vanguard/stream-token.secret.

@@ -420,6 +425,140 @@ def verify_token(token):
        return None
    return user.lower()

+# ---- GET /metrics -----------------------------------------------------------------------------------
+# Prometheus text for Metricbeat's prometheus module. Two sources are merged:
+#   1. counters the media scripts keep (bin/vgmetrics.sh writes MARK/metrics/<component>.counters)
+#   2. gauges computed here at scrape time: unpublished recordings, encodes in flight, failures, MediaMTX
+# Only the loopback listener serves it, and Caddy routes no host to /metrics (every dumpd route over there is
+# an explicit path matcher), so this is reachable from the host and from nowhere else. Nothing here is
+# authenticated and nothing here is allowed to fail: a scrape must never be able to disturb the stack.
+METRICS_DIR = Path(os.environ.get("VG_METRICS_DIR", str(MARK / "metrics")))
+MTX_API = os.environ.get("MTX_API_URL", "http://127.0.0.1:9997")
+IDLE_S = int(os.environ.get("MINS_IDLE", "3")) * 60      # the sweep's own "finished being written" rule
+
+# counters key head -> (metric, type, help, the labels the remaining key fields fill, in order)
+COUNTER_MAP = {
+    "jobs":          ("vanguard_media_jobs_total", "counter", "normalise/deband jobs that finished", ("verdict", "outcome", "kind")),
+    "seconds":       ("vanguard_media_job_seconds_total", "counter", "wall-clock seconds spent in ffmpeg", ("verdict", "kind")),
+    "runs":          ("vanguard_media_runs_total", "counter", "times the script has started work", ()),
+    "deband":        ("vanguard_media_deband_total", "counter", "timecode bands cropped off", ()),
+    "last_finish":   ("vanguard_media_last_finish_timestamp_seconds", "gauge", "when the script last finished", ()),
+    "last_duration": ("vanguard_media_last_run_seconds", "gauge", "how long the last finished run took", ()),
+}
+
+def _num(value):
+    """Prometheus wants a plain number: %g would turn a unix timestamp into 1.76e+09."""
+    return "%d" % value if float(value).is_integer() else "%.3f" % value
+
+def _glob(*patterns):
+    found = []
+    for pattern in patterns:
+        try:
+            found.extend(ROOT.glob(pattern))
+        except OSError:
+            pass
+    return found
+
+def mtx_paths():
+    """MediaMTX's own view of what is live. None if it did not answer."""
+    try:
+        with urllib.request.urlopen(f"{MTX_API}/v3/paths/list?itemsPerPage=1000", timeout=1.5) as response:
+            return json.load(response).get("items") or []
+    except Exception:
+        return None
+
+def metrics_text():
+    order, grouped = [], {}
+    def add(metric, typ, helptext, labels, value):
+        if metric not in grouped:
+            grouped[metric] = (typ, helptext, [])
+            order.append(metric)
+        grouped[metric][2].append((labels, value))
+
+    # 1. whatever the scripts have counted
+    for counters in sorted(METRICS_DIR.glob("*.counters")):
+        component = counters.name[: -len(".counters")]
+        try:
+            lines = counters.read_text().splitlines()
+        except OSError:
+            continue
+        for line in lines:
+            field = line.split()
+            if len(field) != 2:
+                continue
+            try:
+                value = float(field[1])
+            except ValueError:
+                continue
+            head, *rest = field[0].split(":")
+            if head not in COUNTER_MAP:
+                continue
+            metric, typ, helptext, names = COUNTER_MAP[head]
+            labels = {"component": component}
+            labels.update(dict(zip(names, rest)))
+            add(metric, typ, helptext, labels, value)
+
+    # 2. is each script executing right now? (its pid file exists only while it runs; a crash leaves a stale
+    #    one, so the pid is checked against /proc rather than trusted)
+    for pid_file in sorted(METRICS_DIR.glob("*.pid")):
+        try:
+            pid = int(pid_file.read_text().strip() or 0)
+        except (OSError, ValueError):
+            pid = 0
+        add("vanguard_media_running", "gauge", "1 while the script is executing",
+            {"component": pid_file.name[: -len(".pid")]}, 1 if pid and Path(f"/proc/{pid}").is_dir() else 0)
+
+    # 3. the queue. Depth-limited globs on purpose: never a walk of the whole media tree.
+    now = _time.time()
+    recording, pending, oldest = 0, 0, 0.0
+    for f in _glob("streams/*/*/.live/*"):
+        if f.name.startswith(".") or not f.is_file():
+            continue
+        try:
+            age = now - f.stat().st_mtime
+        except OSError:
+            continue
+        # the sweep ignores anything touched in the last MINS_IDLE, so younger files are still being written
+        if age <= IDLE_S:
+            recording += 1
+        else:
+            pending += 1
+            oldest = max(oldest, age)
+    add("vanguard_media_recording", "gauge", "recordings MediaMTX is still writing into .live/", {}, recording)
+    add("vanguard_media_pending", "gauge", "finished recordings waiting to be normalised", {"kind": "stream"}, pending)
+    add("vanguard_media_pending_oldest_seconds", "gauge", "age of the oldest waiting recording", {}, round(oldest))
+    add("vanguard_media_encoding_in_flight", "gauge", "ffmpeg output files open right now", {},
+        len(_glob("streams/*/*.part", "streams/*/*/*.part", "streams/*/*/.live/*.part",
+                  "clips/*/*.part", "clips/*/*/*.part", "clips/*/*/*/*.part")))
+    add("vanguard_media_failed_files", "gauge", "files parked with a normalize-failed log", {},
+        len(_glob("streams/*/*.normalize-failed.log", "streams/*/*/*.normalize-failed.log",
+                  "clips/*/*.normalize-failed.log", "clips/*/*/*.normalize-failed.log",
+                  "clips/*/*/*/*.normalize-failed.log")))
+
+    # 4. MediaMTX. The Prometheus scrape of :9998 already has per-path detail; this is the count the portal
+    #    and the dock actually mean by "live", with the derived .webrtc audio mirrors separated out.
+    paths = mtx_paths()
+    add("vanguard_mediamtx_up", "gauge", "did the MediaMTX API answer this scrape", {}, 0 if paths is None else 1)
+    if paths is not None:
+        ready = [p for p in paths if p.get("ready")]
+        add("vanguard_streams_active", "gauge", "streams publishing right now, mirrors excluded", {},
+            sum(1 for p in ready if not str(p.get("name", "")).endswith(".webrtc")))
+        add("vanguard_streams_mirrors", "gauge", "derived .webrtc audio mirrors running", {},
+            sum(1 for p in ready if str(p.get("name", "")).endswith(".webrtc")))
+        add("vanguard_stream_readers", "gauge", "viewers attached across all paths", {},
+            sum(len(p.get("readers") or []) for p in ready))
+
+    out = []
+    for metric in order:
+        typ, helptext, series = grouped[metric]
+        if helptext:
+            out.append(f"# HELP {metric} {helptext}")
+        out.append(f"# TYPE {metric} {typ}")
+        for labels, value in series:
+            tags = ",".join(f'{k}="{v}"' for k, v in sorted(labels.items()))
+            out.append(f"{metric}{{{tags}}} {_num(value)}" if tags else f"{metric} {_num(value)}")
+    return "\n".join(out) + "\n"
+
 class H(BaseHTTPRequestHandler):
    server_version = "dumpd/1"
    def _json(self, code, obj):
@@ -429,6 +568,13 @@ class H(BaseHTTPRequestHandler):
        self.send_header("Content-Length", str(len(body)))
        self.end_headers()
        self.wfile.write(body)
+    def _text(self, code, body):
+        raw = body.encode()
+        self.send_response(code)
+        self.send_header("Content-Type", "text/plain; version=0.0.4; charset=utf-8")
+        self.send_header("Content-Length", str(len(raw)))
+        self.end_headers()
+        self.wfile.write(raw)
    def _deny(self):
        self._json(401, {"error": "not authenticated (no identity header from the proxy)"})
    def _parse(self):
@@ -457,6 +603,12 @@ class H(BaseHTTPRequestHandler):
    def do_GET(self):
        if self.server.server_port == AUTH_LISTEN[1]:
            return self._json(404, {"error": "not found"})          # the auth port serves POST /mtx-auth only
+        if self.path == "/metrics":
+            try:
+                return self._text(200, metrics_text())
+            except Exception as error:                  # a scrape must never take dumpd down
+                log("metrics_failed", level="error", error=str(error))
+                return self._text(500, f"# metrics unavailable: {error}\n")
        if self.path.startswith("/time"):
            # server clock for the timecode strip (the /clock/ page and Go Live measure their offset against this)
            return self._json(200, {"now_ms": int(_time.time() * 1000)})
--- a/bin/normalize-sweep.sh
+++ b/bin/normalize-sweep.sh
@@ -36,6 +36,15 @@ MAX_W=2560; MAX_H=1440; MAX_FPS=60
 exec 9>"${LOCK_FILE:-/run/lock/normalize-sweep.lock}"
 flock -n 9 || exit 0

+# Metrics (LOGGING.md): counters for dumpd's /metrics. Optional on purpose — if the helper is not installed
+# yet, these become no-ops and normalising carries on exactly as before.
+if [[ -r "$(dirname "$0")/vgmetrics.sh" ]]; then
+  VG_MET_COMPONENT=sweep
+  . "$(dirname "$0")/vgmetrics.sh" && vgmet_init
+else
+  vgmet_add() { :; }; vgmet_set() { :; }; vgmet_time() { echo 0; }; vgmet_since() { echo 0; }
+fi
+
 # logfmt (LOGGING.md): vglog <level> event=<verb> key=value ...   values with spaces must arrive pre-quoted
 vglog() { local lvl="$1"; shift; printf 'ts=%s level=%s component=sweep %s\n' "$(date -Is)" "$lvl" "$*"; }
 log() { vglog info "msg=\"$*\""; }
@@ -84,9 +93,6 @@ probe() {
  echo "ok"
 }

-find "$ROOT/streams" "$ROOT/clips" -type f \
-     \( -iname '*.mp4' -o -iname '*.mkv' -o -iname '*.mov' -o -iname '*.ts' -o -iname '*.m4v' -o -iname '*.webm' \) \
-     ! -path '*/.src/*' ! -path '*/.stfolder/*' ! -path '*/.stversions/*' ! -name '*.part' -mmin +"$MINS_IDLE" -print0 2>/dev/null |
 while IFS= read -r -d '' src; do
  base="${src%.*}"; ext="${src##*.}"
  live=0; stream=""
@@ -121,6 +127,7 @@ while IFS= read -r -d '' src; do
    [[ -e "$out" ]] && continue
  fi
  tmp="$out.part"
+  kind=clip; (( live )) && kind=stream           # metric label: stream encodes are the expensive ones
  vglog info "event=verdict verdict=$verdict file=\"$src\" reason=\"$reason\" stream=\"$stream\""
  tcargs=(); tcjson=""; band=0; deband=0
  if (( live )); then
@@ -157,9 +164,11 @@ PYEOF
      fi
    fi
  fi
+  job_t0=$(vgmet_time)
+  err=$(mktemp /tmp/vgsweep-err.XXXXXX)          # ffmpeg's own words; rc= alone is not diagnosable
  if [[ $verdict == remux ]]; then
    nice -n 15 ionice -c 3 ffmpeg -nostdin -hide_banner -loglevel error -y -i "$src" \
-      -map 0 -c copy "${tcargs[@]}" -movflags +faststart -f mp4 "$tmp"
+      -map 0 -c copy "${tcargs[@]}" -movflags +faststart -f mp4 "$tmp" 2>"$err"
  else
    # scale down into the cap (never up), keep aspect, even dimensions; cap fps; CFR; all audio -> AAC
    vf="scale='min(iw,$MAX_W)':'min(ih,$MAX_H)':force_original_aspect_ratio=decrease:force_divisible_by=2"
@@ -175,11 +184,14 @@ PYEOF
      -c:v libx264 -preset veryfast -crf 18 -maxrate 30M -bufsize 60M -pix_fmt yuv420p -profile:v high -threads "$ENCODE_THREADS" \
      -vf "$vf" -fps_mode "$fpsmode" "${fpsargs[@]}" \
      "${aargs[@]}" \
-      "${tcargs[@]}" -movflags +faststart -f mp4 "$tmp"
+      "${tcargs[@]}" -movflags +faststart -f mp4 "$tmp" 2>"$err"
  fi
  rc=$?
+  job_secs=$(vgmet_since "$job_t0")
  d1=$(dur "$src"); d2=$(dur "$tmp")
  if (( rc == 0 )) && [[ -n $d1 && -n $d2 ]] && (( d2 >= d1 - 2 && d2 <= d1 + 2 )); then
+    vgmet_add "jobs:$verdict:done:$kind"; vgmet_add "seconds:$verdict:$kind" "$job_secs"
+    (( deband )) && vgmet_add deband 1
    if (( live )); then
      mv -f "$tmp" "$out"; chown 1000:1000 "$out" 2>/dev/null
      if [[ -n $tcjson ]]; then                        # sidecar: exact capture time (read by the Jellyfin plugin) + drift samples
@@ -192,6 +204,7 @@ PYEOF
        rm -f "$src"                                   # band removal: the cropped file IS the recording; raw-with-band is not kept
      fi
      rmdir "$(dirname "$src")" "$titledir" 2>/dev/null || true   # drop .live and the empty title folder
+      rm -f "$err"
      vglog info "event=published file=\"$out\" stream=\"$stream\" verdict=$verdict"; continue
    elif (( synced )); then
      rel_src="${src#"$stroot"/}"; rel_out="${out#"$stroot"/}"
@@ -211,7 +224,13 @@ PYEOF
    vglog info "event=done file=\"$out\" verdict=$verdict"
  else
    rm -f "$tmp"
-    { echo "$(date '+%F %T') $verdict failed rc=$rc src_dur=${d1:-?} out_dur=${d2:-?} ($reason)"; } > "$base.normalize-failed.log"
-    vglog error "event=failed file=\"$src\" stream=\"$stream\" verdict=$verdict rc=$rc log=\"$base.normalize-failed.log\"" >&2
+    vgmet_add "jobs:$verdict:failed:$kind"; vgmet_add "seconds:$verdict:$kind" "$job_secs"
+    { echo "$(date '+%F %T') $verdict failed rc=$rc src_dur=${d1:-?} out_dur=${d2:-?} ($reason)"
+      echo "--- ffmpeg stderr ---"; tail -40 "$err" 2>/dev/null; } > "$base.normalize-failed.log"
+    why=$(tail -1 "$err" 2>/dev/null | tr -d '"' | cut -c1-200)
+    vglog error "event=failed file=\"$src\" stream=\"$stream\" verdict=$verdict rc=$rc reason=\"${why:-no stderr}\" log=\"$base.normalize-failed.log\"" >&2
  fi
-done
+  rm -f "$err"
+done < <(find "$ROOT/streams" "$ROOT/clips" -type f \
+     \( -iname '*.mp4' -o -iname '*.mkv' -o -iname '*.mov' -o -iname '*.ts' -o -iname '*.m4v' -o -iname '*.webm' \) \
+     ! -path '*/.src/*' ! -path '*/.stfolder/*' ! -path '*/.stversions/*' ! -name '*.part' -mmin +"$MINS_IDLE" -print0 2>/dev/null)
--- a/bin/vanguard-alerts.sh
+++ b/bin/vanguard-alerts.sh
@@ -0,0 +1,152 @@
+#!/bin/bash
+# Deliver Kibana's alerts to Discord. Runs from a systemd timer every minute; curl and python only.
+#
+# WHY THIS EXISTS
+# On a basic license Kibana's only connector types are .index and .server-log — every connector that leaves
+# the stack is Gold or above. So Kibana rules write their alerts into ONE Elasticsearch index (the alert bus,
+# created by observability/kibana-setup.py) and this script is the delivery side. Alerts are therefore
+# ordinary documents: queryable, replayable, and visible in Discover alongside the logs that caused them.
+#
+# Heartbeat documents (kind=heartbeat) are deliberately NOT delivered. They exist for bin/vanguard-deadman.sh
+# to notice their absence, which is the only way to catch the alerting pipeline itself dying.
+#
+# Repeat suppression is a safety net, not the primary throttle: a rule should use "On status change" or its
+# own throttle in Kibana. This stops a misconfigured rule turning into a message every minute, and it is per
+# rule name, so one noisy rule never silences another.
+#
+# Exit status: 0 when there was nothing to send or everything was sent; 1 when alerts existed and could not be
+# delivered, so `systemctl status vanguard-alerts` and the journal both carry the failure.
+set -uo pipefail
+
+ES="${ES_URL:-http://127.0.0.1:9200}"
+ES_NETRC="${ES_NETRC_FILE:-/etc/vanguard/es-netrc}"
+INDEX="${ALERT_INDEX:-vanguard-alerts}"
+STATE="${ALERT_STATE:-/var/lib/vanguard/alerts.state}"
+WEBHOOK_FILE="${ALERT_WEBHOOK_FILE:-/etc/vanguard/alert-webhook}"
+SUPPRESS_S="${ALERT_SUPPRESS_S:-900}"
+BATCH="${ALERT_BATCH:-50}"
+
+vglog() { local lvl="$1"; shift; printf 'ts=%s level=%s component=alerts %s\n' "$(date -Is)" "$lvl" "$*"; }
+es_curl() { if [[ -r $ES_NETRC ]]; then curl -s --netrc-file "$ES_NETRC" "$@"; else curl -s "$@"; fi; }
+
+state_get() { [[ -r $STATE ]] || return 0; grep -m1 "^$1=" "$STATE" 2>/dev/null | cut -d= -f2-; }
+state_set() {
+  local key="$1" value="$2" tmp
+  install -d "$(dirname "$STATE")" 2>/dev/null || return 0
+  tmp=$(mktemp "${STATE}.XXXXXX" 2>/dev/null) || return 0
+  { [[ -r $STATE ]] && grep -v "^${key}=" "$STATE"; printf '%s=%s\n' "$key" "$value"; } > "$tmp" 2>/dev/null
+  mv -f "$tmp" "$STATE" 2>/dev/null || rm -f "$tmp"
+}
+
+webhook=""
+[[ -r $WEBHOOK_FILE ]] && webhook=$(tr -d '[:space:]' < "$WEBHOOK_FILE")
+# Same guard as the deadman: a typo that is not an https URL must not be posted to, and must not look like
+# delivery succeeded. Treated as "no webhook", which means journal-only and a non-zero exit.
+case "${webhook:-}" in
+  "")       ;;
+  https://*) ;;
+  *)        vglog error 'event=bad_webhook msg="ALERT_WEBHOOK_URL is not an https URL; ignoring it"' >&2; webhook="" ;;
+esac
+
+now=$(date +%s)
+cursor=$(state_get cursor); cursor="${cursor:-0}"
+
+# Everything newer than the cursor that is not a heartbeat, oldest first so the cursor advances safely.
+query=$(printf '{"size":%d,"sort":[{"@timestamp":"asc"}],"query":{"bool":{"filter":[{"range":{"@timestamp":{"gt":%s,"format":"epoch_millis"}}}],"must_not":[{"term":{"kind":"heartbeat"}}]}}}' "$BATCH" "$cursor")
+response=$(es_curl -m 15 -H 'Content-Type: application/json' "$ES/$INDEX/_search" -d "$query")
+
+if [[ -z $response ]]; then
+  vglog error 'event=unreachable msg="Elasticsearch did not answer; cannot read the alert bus"' >&2
+  exit 1
+fi
+# A missing index just means no rule has fired yet. Not a fault, and not worth a line every minute.
+if grep -q 'index_not_found_exception' <<<"$response"; then
+  exit 0
+fi
+
+# Retention. A heartbeat document arrives every minute, so without this the alert bus grows without bound.
+# At most once every 6 h: a delete_by_query is cheap, but not worth doing sixty times an hour. Deliberately
+# before the "nothing new" exit below, because most runs have nothing to deliver and would skip it.
+retention_days="${ALERT_RETENTION_DAYS:-90}"
+last_purge=$(state_get purged); last_purge="${last_purge:-0}"
+if (( now - last_purge > 21600 )); then
+  purge=$(es_curl -m 120 -H 'Content-Type: application/json' \
+            -XPOST "$ES/$INDEX/_delete_by_query?conflicts=proceed" \
+            -d "{\"query\":{\"range\":{\"@timestamp\":{\"lt\":\"now-${retention_days}d\"}}}}")
+  gone=$(printf '%s' "${purge:-}" | python3 -c 'import json,sys
+try: print(json.load(sys.stdin).get("deleted", 0))
+except Exception: print("?")')
+  state_set purged "$now"
+  vglog info "event=purged older_than=${retention_days}d deleted=${gone:-?}"
+fi
+
+rows=$(printf '%s' "$response" | python3 -c '
+import json, sys
+from datetime import datetime
+try:
+    data = json.load(sys.stdin)
+except Exception:
+    sys.exit(0)
+for hit in data.get("hits", {}).get("hits", []):
+    source = hit.get("_source", {})
+    stamp = str(source.get("@timestamp", ""))
+    try:
+        when = datetime.fromisoformat(stamp.replace("Z", "+00:00"))
+        millis = int(when.timestamp() * 1000)
+    except Exception:
+        continue
+    def clean(value, limit):
+        return " ".join(str(value).split())[:limit] or "-"
+    print("\t".join((str(millis),
+                     clean(source.get("rule", "unnamed rule"), 120),
+                     clean(source.get("level", "warn"), 12),
+                     clean(source.get("message", json.dumps(source)[:400]), 1200))))
+')
+
+[[ -z $rows ]] && exit 0
+
+# One per line: "a=0 b=0 now" would make bash run `now` as a COMMAND with a and b as temporary
+# assignments, leaving every counter unset for the rest of the script.
+sent=0
+suppressed=0
+failed=0
+while IFS=$'\t' read -r millis rule level message; do
+  [[ -z ${millis:-} ]] && continue
+  last=$(state_get "sent:$rule"); last="${last:-0}"
+  if (( now - last < SUPPRESS_S )); then
+    suppressed=$((suppressed + 1))
+    vglog info "event=suppressed rule=\"$rule\" within=${SUPPRESS_S}s"
+    state_set cursor "$millis"
+    continue
+  fi
+  case "$level" in
+    error|critical) mark="[ERROR]" ;;
+    warn|warning)   mark="[WARN]" ;;
+    *)              mark="[$level]" ;;
+  esac
+  text="$mark $rule — $message"
+  if [[ -z $webhook ]]; then
+    vglog warn "event=undeliverable rule=\"$rule\" msg=\"no Discord webhook configured; alert only in the journal\"" >&2
+    vglog warn "event=alert rule=\"$rule\" level=$level msg=\"$message\"" >&2
+    failed=$((failed + 1))
+    state_set cursor "$millis"
+    continue
+  fi
+  payload=$(VG_TEXT="$text" python3 -c 'import json, os; print(json.dumps({"content": os.environ["VG_TEXT"][:1900]}))')
+  # curl already prints 000 on a connection failure; "|| echo 000" would concatenate and give 000000.
+  code=$(curl -s -o /dev/null -w '%{http_code}' -m 15 -X POST -H 'Content-Type: application/json' \
+           -d "$payload" "$webhook" || true)
+  code="${code:-000}"
+  if [[ $code == 2* ]]; then
+    sent=$((sent + 1)); state_set "sent:$rule" "$now"
+    vglog info "event=delivered rule=\"$rule\" level=$level"
+  else
+    failed=$((failed + 1))
+    vglog error "event=delivery_failed rule=\"$rule\" http=$code" >&2
+  fi
+  state_set cursor "$millis"
+done <<<"$rows"
+
+vglog info "event=run sent=$sent suppressed=$suppressed failed=$failed"
+(( failed > 0 )) && exit 1
+exit 0
--- a/bin/vanguard-deadman.sh
+++ b/bin/vanguard-deadman.sh
@@ -25,9 +25,19 @@
 set -uo pipefail

 ES="${ES_URL:-http://127.0.0.1:9200}"
+# Elasticsearch requires credentials (security is on; observability/kibana.yml says why). configure.py writes
+# this netrc, root-only: netrc rather than -u so the password is never visible in a process list. If the file
+# is missing, the calls below get 401 and the deadman reports Elasticsearch as unreachable — which is correct,
+# because from its point of view it is. Kibana needs nothing: its anonymous provider signs callers in.
+ES_NETRC="${ES_NETRC_FILE:-/etc/vanguard/es-netrc}"
+es_curl() {
+  if [[ -r $ES_NETRC ]]; then curl -s --netrc-file "$ES_NETRC" "$@"; else curl -s "$@"; fi
+}
 KIBANA="${KIBANA_URL:-http://127.0.0.1:5601}"
 WEBHOOK_FILE="${ALERT_WEBHOOK_FILE:-/etc/vanguard/alert-webhook}"
-HEARTBEAT="${ALERT_HEARTBEAT_FILE:-/srv/media/.markers/alert-heartbeat}"
+ALERT_INDEX="${ALERT_INDEX:-vanguard-alerts}"            # the alert bus Kibana's rules write into
+ALERT_STATE="${ALERT_STATE:-/var/lib/vanguard/alerts.state}"  # touched by every vanguard-alerts.sh run
+ALERT_STATE_MAX_S="${ALERT_STATE_MAX_S:-900}"
 HEARTBEAT_MAX_S="${HEARTBEAT_MAX_S:-900}"
 STALE_MAX_S="${BEAT_STALE_MAX_S:-900}"
 STATE="${DEADMAN_STATE:-/var/lib/vanguard/deadman.state}"
@@ -44,7 +54,7 @@ for k in sys.argv[1].split("."):
 print(d)' "$1" 2>/dev/null; }

 # 1. Elasticsearch
-health=$(curl -s -m 8 "$ES/_cluster/health" || true)
+health=$(es_curl -m 8 "$ES/_cluster/health" || true)
 if [[ -z $health ]]; then
  note "Elasticsearch is not answering on ${ES#http://} — nothing is being stored"
 else
@@ -56,7 +66,7 @@ else
  esac

  # 2. read-only indices. Elastic sets this when the disk passes the flood watermark and never unsets it.
-  blocked=$(curl -s -m 8 "$ES/_all/_settings/index.blocks.read_only_allow_delete" \
+  blocked=$(es_curl -m 8 "$ES/_all/_settings/index.blocks.read_only_allow_delete" \
            | python3 -c 'import json,sys
 try: d=json.load(sys.stdin)
 except Exception: sys.exit(0)
@@ -66,7 +76,7 @@ print(len(bad))' 2>/dev/null || echo 0)

  # 4 and 5. are the Beats actually writing?
  for beat in filebeat metricbeat; do
-    count=$(curl -s -m 8 -H 'Content-Type: application/json' \
+    count=$(es_curl -m 8 -H 'Content-Type: application/json' \
              "$ES/${beat}-*/_count" \
              -d "{\"query\":{\"range\":{\"@timestamp\":{\"gte\":\"now-${STALE_MAX_S}s\"}}}}" | jq_get count)
    if [[ -z ${count:-} ]]; then
@@ -81,12 +91,31 @@ fi
 level=$(curl -s -m 8 "$KIBANA/api/status" | jq_get status.overall.level)
 [[ ${level:-} == available ]] || note "Kibana is not available (${level:-no answer}) — the only UI is down"

-# 6. the heartbeat a Kibana rule posts here. Missing right after a restart is not yet a fault; age is.
-if [[ -f $HEARTBEAT ]]; then
-  age=$(( $(date +%s) - $(stat -c %Y "$HEARTBEAT") ))
-  (( age <= HEARTBEAT_MAX_S )) || note "no alert heartbeat for ${age}s (the Kibana rule → webhook → dumpd path is broken)"
-else
-  note "no alert heartbeat has ever arrived (is the Kibana heartbeat rule created? README 5h)"
+# 6. the heartbeat. A Kibana rule writes a kind=heartbeat document into the alert index every minute (see
+# observability/kibana-setup.py). Its absence is the ONLY thing that catches the alerting pipeline itself
+# dying: if Kibana, its rule engine or the index connector stops, no ordinary alert can report it, because
+# the thing that would report it is what broke. Checked in Elasticsearch, not as a file: a basic licence has
+# no webhook connector, so nothing posts to dumpd any more.
+if [[ -n $health ]]; then
+  beats=$(es_curl -m 8 -H 'Content-Type: application/json' "$ES/$ALERT_INDEX/_count" \
+            -d "{\"query\":{\"bool\":{\"filter\":[{\"term\":{\"kind\":\"heartbeat\"}},{\"range\":{\"@timestamp\":{\"gte\":\"now-${HEARTBEAT_MAX_S}s\"}}}]}}}")
+  if grep -q 'index_not_found_exception' <<<"${beats:-}"; then
+    note "the alert index $ALERT_INDEX does not exist — the Kibana connector and heartbeat rule have not been created (observability/kibana-setup.py)"
+  else
+    hb=$(printf '%s' "${beats:-}" | jq_get count)
+    if [[ -z ${hb:-} ]]; then
+      note "cannot read the alert heartbeat (query against $ALERT_INDEX failed)"
+    elif (( hb == 0 )); then
+      note "no alert heartbeat in the last $((HEARTBEAT_MAX_S/60)) minutes — Kibana's rule engine or the index connector has stopped, so NO alert can reach Discord"
+    fi
+  fi
+fi
+
+# 7. is the delivery side running? The heartbeat proves Kibana still writes alerts; this proves something
+# still takes them to Discord. Both halves have to be alive for an alarm to arrive.
+if [[ -f $ALERT_STATE ]]; then
+  age=$(( $(date +%s) - $(stat -c %Y "$ALERT_STATE") ))
+  (( age <= ALERT_STATE_MAX_S )) || note "vanguard-alerts.timer has not run for ${age}s — alerts are being written but not delivered"
 fi

 # the disks, independently of any agent
--- a/bin/vgdeband.sh
+++ b/bin/vgdeband.sh
@@ -11,6 +11,13 @@ ROOT="${MEDIA_ROOT:-/srv/media}"
 vglog() { local lvl="$1"; shift; printf 'ts=%s level=%s component=vgdeband %s\n' "$(date -Is)" "$lvl" "$*"; }
 log() { vglog info "msg=\"$*\""; }
 dur() { ffprobe -v error -show_entries format=duration -of csv=p=0 "$1" 2>/dev/null | cut -d. -f1; }
+# Metrics (LOGGING.md): the same counters the sweep keeps, under component=vgdeband. No-ops if not installed.
+if [[ -r "$(dirname "$0")/vgmetrics.sh" ]]; then
+  VG_MET_COMPONENT=vgdeband
+  . "$(dirname "$0")/vgmetrics.sh" && vgmet_init
+else
+  vgmet_add() { :; }; vgmet_set() { :; }; vgmet_time() { echo 0; }; vgmet_since() { echo 0; }
+fi
 if (( $# )); then files=("$@"); else mapfile -t files < <(find "$ROOT/streams" -mindepth 2 -maxdepth 2 -type f -name '*.mp4' | sort); fi
 for f in "${files[@]}"; do
  side="${f%.*}.json"
@@ -26,19 +33,22 @@ print(d.get("source","none"), 1 if d.get("band_removed") else 0, d.get("timecode
  tmp="$f.part"
  tcargs=(); [[ -n $tc ]] && tcargs=(-timecode "$tc" -write_tmcd 1); [[ -n $ctime ]] && tcargs+=(-metadata "creation_time=$ctime")
  vglog info "event=deband file=\"$f\" width=$w band_rows=$band timecode=\"$tc\""
+  job_t0=$(vgmet_time)
  nice -n 15 ionice -c 3 ffmpeg -nostdin -hide_banner -loglevel error -y -i "$f" \
    -map 0:v:0 -map '0:a?' \
    -c:v libx264 -preset veryfast -crf 18 -maxrate 30M -bufsize 60M -pix_fmt yuv420p -profile:v high -threads "${ENCODE_THREADS:-2}" \
    -vf "crop=iw:ih-${band}:0:0" -fps_mode passthrough -c:a copy \
    "${tcargs[@]}" -movflags +faststart -f mp4 "$tmp"
-  rc=$?; d1=$(dur "$f"); d2=$(dur "$tmp")
+  rc=$?; job_secs=$(vgmet_since "$job_t0"); d1=$(dur "$f"); d2=$(dur "$tmp")
+  vgmet_add "seconds:encode:stream" "$job_secs"
  if (( rc == 0 )) && [[ -n $d1 && -n $d2 ]] && (( d2 >= d1 - 2 && d2 <= d1 + 2 )); then
+    vgmet_add "jobs:encode:done:stream"; vgmet_add deband 1
    mv -f "$tmp" "$f"; chown 1000:1000 "$f" 2>/dev/null
    python3 -c '
 import json,sys; p=sys.argv[1]; d=json.load(open(p)); d["band_rows"]=int(sys.argv[2]); d["band_removed"]=True
 json.dump(d, open(p,"w"))' "$side" "$band"; chown 1000:1000 "$side" 2>/dev/null
    vglog info "event=done file=\"$f\""
  else
-    rm -f "$tmp"; vglog error "event=failed file=\"$f\" rc=$rc src_dur=${d1:-?} out_dur=${d2:-?}" >&2
+    rm -f "$tmp"; vgmet_add "jobs:encode:failed:stream"; vglog error "event=failed file=\"$f\" rc=$rc src_dur=${d1:-?} out_dur=${d2:-?}" >&2
  fi
 done
--- a/bin/vgmetrics.sh
+++ b/bin/vgmetrics.sh
@@ -0,0 +1,63 @@
+# Shared metric counters for the media scripts. Sourced, never executed.
+#
+#   VG_MET_COMPONENT=sweep                      # a plain assignment, NOT a prefix on the next line:
+#   . /usr/local/bin/vgmetrics.sh && vgmet_init  # bash discards "VAR=x . file" assignments on return
+#   vgmet_add "jobs:encode:done:stream"      # +1
+#   vgmet_add "seconds:encode:stream" 41.2   # +41.2
+#
+# Counters accumulate across runs in MARK/metrics/<component>.counters as plain "key value" lines, and a
+# <component>.pid file exists only while the script is executing. dumpd turns both into Prometheus text on
+# its /metrics; nothing here renders or serves anything, so the scripts stay simple and there is one place
+# that knows the metric names (bin/dumpd.py COUNTER_MAP).
+#
+# Keys are "name:label:label..." — the name picks the metric, the rest fill its labels in order.
+# Writes are atomic (temp file + mv) and every failure is silent: metrics must never break a sweep.
+
+VG_MET_DIR="${VG_MET_DIR:-${MEDIA_ROOT:-/srv/media}/.markers/metrics}"
+VG_MET_COMPONENT="${VG_MET_COMPONENT:-$(basename "${0%.sh}")}"
+VG_MET_START=0
+declare -A VGMET=()
+
+vgmet_init() {
+  : "${VG_MET_DIR:=${MEDIA_ROOT:-/srv/media}/.markers/metrics}"
+  : "${VG_MET_COMPONENT:=unknown}"
+  VG_MET_START=$(date +%s)
+  mkdir -p "$VG_MET_DIR" 2>/dev/null || return 0
+  local f="$VG_MET_DIR/$VG_MET_COMPONENT.counters" k v
+  if [[ -r $f ]]; then
+    while read -r k v; do [[ -n ${k:-} && -n ${v:-} ]] && VGMET["$k"]="$v"; done < "$f"
+  fi
+  vgmet_add runs 1
+  printf '%s\n' "$$" > "$VG_MET_DIR/$VG_MET_COMPONENT.pid" 2>/dev/null
+  trap vgmet_flush EXIT
+}
+
+# vgmet_add <key> [amount]   — float-safe accumulate
+vgmet_add() {
+  local k="$1" v="${2:-1}"
+  VGMET["$k"]=$(awk -v a="${VGMET[$k]:-0}" -v b="$v" 'BEGIN{ printf "%.3f", a+b }')
+}
+
+vgmet_set() { VGMET["$1"]="$2"; }
+
+vgmet_flush() {
+  : "${VG_MET_DIR:=${MEDIA_ROOT:-/srv/media}/.markers/metrics}"
+  : "${VG_MET_COMPONENT:=unknown}"
+  local f="$VG_MET_DIR/$VG_MET_COMPONENT.counters" tmp now k
+  rm -f "$VG_MET_DIR/$VG_MET_COMPONENT.pid" 2>/dev/null
+  [[ -d $VG_MET_DIR ]] || return 0
+  now=$(date +%s)
+  vgmet_set last_finish "$now"
+  vgmet_set last_duration "$(( now - VG_MET_START ))"
+  tmp=$(mktemp "$VG_MET_DIR/.$VG_MET_COMPONENT.XXXXXX" 2>/dev/null) || return 0
+  for k in "${!VGMET[@]}"; do printf '%s %s\n' "$k" "${VGMET[$k]}"; done > "$tmp" 2>/dev/null
+  chmod 0644 "$tmp" 2>/dev/null
+  mv -f "$tmp" "$f" 2>/dev/null || rm -f "$tmp" 2>/dev/null
+}
+
+# start/elapsed for one job
+vgmet_time()  { date +%s.%N; }
+vgmet_since() { awk -v s="$1" -v e="$(date +%s.%N)" 'BEGIN{ printf "%.3f", e-s }'; }
+
+# No-op fallbacks are defined by the callers if this file is missing, so a half-finished deploy cannot
+# break normalising. Keep every function above failure-tolerant.