254 lines
8.4 KiB
Python
254 lines
8.4 KiB
Python
|
#!/usr/bin/env python3
|
||
|
|
"""placement-audit.py -- readable audit of sidechat placement / pre-send gate
|
||
|
|
activity in dm-log.jsonl.
|
||
|
|
|
||
|
|
The gate and navigation emit machine-readable events (pre_send_assert_failed,
|
||
|
|
placement_failed, placement_mismatch, sidechat_uuid_capture_failed, nav_failed,
|
||
|
|
...). This CLI turns them into human-readable summaries so the logs stay
|
||
|
|
useful for troubleshooting sidechat reliability.
|
||
|
|
|
||
|
|
Usage:
|
||
|
|
placement-audit.py --since 24h # summary over the last 24 hours
|
||
|
|
placement-audit.py --since 7d # ... 7 days
|
||
|
|
placement-audit.py --since 60m # ... 60 minutes
|
||
|
|
placement-audit.py --watch # live tail of placement-relevant events
|
||
|
|
|
||
|
|
Read-only: never writes, no network, no browser.
|
||
|
|
"""
|
||
|
|
|
||
|
|
import argparse
|
||
|
|
import json
|
||
|
|
import re
|
||
|
|
import sys
|
||
|
|
import time
|
||
|
|
from collections import Counter, defaultdict
|
||
|
|
from datetime import datetime, timedelta, timezone
|
||
|
|
|
||
|
|
DM_LOG = "/home/super/Projects/NetVM/dm-log.jsonl"
|
||
|
|
JOB_LOG = "/home/super/Projects/NetVM/job-log.jsonl"
|
||
|
|
|
||
|
|
# Event types that count as placement/navigation failures.
|
||
|
|
FAILURE_TYPES = {
|
||
|
|
"sidechat_uuid_capture_failed",
|
||
|
|
"placement_mismatch",
|
||
|
|
"placement_failed",
|
||
|
|
"pre_send_assert_failed",
|
||
|
|
"nav_failed",
|
||
|
|
"send_failed",
|
||
|
|
"failed",
|
||
|
|
}
|
||
|
|
|
||
|
|
# Event types worth streaming in --watch (dm-log).
|
||
|
|
WATCH_TYPES_DM = FAILURE_TYPES | {
|
||
|
|
"verified",
|
||
|
|
"sent",
|
||
|
|
"nav_ok",
|
||
|
|
"sidechat_autoprovisioned",
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def parse_since(s):
|
||
|
|
m = re.fullmatch(r"(\d+)(m|h|d)", (s or "").strip().lower())
|
||
|
|
if not m:
|
||
|
|
raise ValueError(f"bad --since value {s!r}; use like 60m, 24h, 7d")
|
||
|
|
n, unit = int(m.group(1)), m.group(2)
|
||
|
|
return timedelta(minutes=n) if unit == "m" else timedelta(hours=n) if unit == "h" else timedelta(days=n)
|
||
|
|
|
||
|
|
|
||
|
|
def parse_ts(ts):
|
||
|
|
if not ts:
|
||
|
|
return None
|
||
|
|
try:
|
||
|
|
dt = datetime.fromisoformat(str(ts).replace("Z", "+00:00"))
|
||
|
|
except ValueError:
|
||
|
|
return None
|
||
|
|
if dt.tzinfo is None:
|
||
|
|
dt = dt.replace(tzinfo=timezone.utc)
|
||
|
|
return dt
|
||
|
|
|
||
|
|
|
||
|
|
def short_uuid(u):
|
||
|
|
u = str(u or "")
|
||
|
|
return u[:8] + "..." if len(u) > 12 else u
|
||
|
|
|
||
|
|
|
||
|
|
def fmt_ts(e):
|
||
|
|
dt = parse_ts(e.get("ts"))
|
||
|
|
return dt.strftime("%m-%d %H:%M:%S") if dt else "?"
|
||
|
|
|
||
|
|
|
||
|
|
def expected_actual(e):
|
||
|
|
"""Return (expected, actual) strings for a failure event."""
|
||
|
|
t = e.get("type")
|
||
|
|
if t in ("pre_send_assert_failed", "placement_failed"):
|
||
|
|
return e.get("expected_uuid"), e.get("actual_url")
|
||
|
|
if t == "placement_mismatch":
|
||
|
|
return e.get("expected_uuid"), e.get("actual_uuid")
|
||
|
|
if t == "sidechat_uuid_capture_failed":
|
||
|
|
return "(uuid capture)", e.get("out")
|
||
|
|
if t == "nav_failed":
|
||
|
|
return e.get("expected") or e.get("reason"), e.get("got") or e.get("out")
|
||
|
|
if t in ("send_failed", "failed"):
|
||
|
|
return e.get("reason"), (e.get("send_out") or e.get("err") or "")[:120]
|
||
|
|
return None, None
|
||
|
|
|
||
|
|
|
||
|
|
def one_line(e):
|
||
|
|
"""Compact one-line summary of an event."""
|
||
|
|
t = e.get("type", "?")
|
||
|
|
who = f"{e.get('agent', '?')}->{e.get('to', '?')}/{e.get('target', '?')}"
|
||
|
|
base = f"{fmt_ts(e)} {t:28s} {str(e.get('id', ''))[:8]:8s} {who}"
|
||
|
|
if t in FAILURE_TYPES:
|
||
|
|
reason = e.get("reason") or ""
|
||
|
|
exp, act = expected_actual(e)
|
||
|
|
extra = f" reason={reason}" if reason else ""
|
||
|
|
if exp or act:
|
||
|
|
extra += f" expected={short_uuid(exp) if exp and len(str(exp)) > 20 else exp} actual={act}"
|
||
|
|
return base + extra
|
||
|
|
if t == "sent":
|
||
|
|
return base + f" verified={e.get('verified')}"
|
||
|
|
if t == "verified":
|
||
|
|
return base + f" placement={e.get('placement')} thread={short_uuid(e.get('thread_uuid'))}"
|
||
|
|
if t == "nav_ok":
|
||
|
|
return base + f" status={e.get('status')} url={e.get('browser_url')}"
|
||
|
|
if t == "sidechat_autoprovisioned":
|
||
|
|
return base + f" thread={short_uuid(e.get('thread_uuid'))}"
|
||
|
|
return base
|
||
|
|
|
||
|
|
|
||
|
|
def iter_log(path, cutoff=None):
|
||
|
|
"""Yield parsed events from a jsonl file, optionally filtered by ts."""
|
||
|
|
try:
|
||
|
|
f = open(path, encoding="utf-8")
|
||
|
|
except OSError as ex:
|
||
|
|
print(f"warning: cannot open {path}: {ex}", file=sys.stderr)
|
||
|
|
return
|
||
|
|
with f:
|
||
|
|
for line in f:
|
||
|
|
line = line.strip()
|
||
|
|
if not line:
|
||
|
|
continue
|
||
|
|
try:
|
||
|
|
e = json.loads(line)
|
||
|
|
except json.JSONDecodeError:
|
||
|
|
continue
|
||
|
|
if cutoff is not None:
|
||
|
|
dt = parse_ts(e.get("ts"))
|
||
|
|
if dt is None or dt < cutoff:
|
||
|
|
continue
|
||
|
|
yield e
|
||
|
|
|
||
|
|
|
||
|
|
def cmd_since(args):
|
||
|
|
try:
|
||
|
|
delta = parse_since(args.since)
|
||
|
|
except ValueError as ex:
|
||
|
|
print(str(ex), file=sys.stderr)
|
||
|
|
return 1
|
||
|
|
cutoff = datetime.now(timezone.utc) - delta
|
||
|
|
|
||
|
|
sends = Counter() # (agent, target) -> sent events
|
||
|
|
sent_ok = Counter() # (agent, target) -> sent with verified=True
|
||
|
|
taxonomy = Counter() # failure label -> count
|
||
|
|
failures = [] # failure events, for the recent list
|
||
|
|
total = 0
|
||
|
|
|
||
|
|
for e in iter_log(DM_LOG, cutoff):
|
||
|
|
total += 1
|
||
|
|
t = e.get("type")
|
||
|
|
key = (e.get("agent") or "?", e.get("target") or "?")
|
||
|
|
if t == "sent":
|
||
|
|
sends[key] += 1
|
||
|
|
if e.get("verified") is True:
|
||
|
|
sent_ok[key] += 1
|
||
|
|
if t in FAILURE_TYPES:
|
||
|
|
reason = e.get("reason") or ""
|
||
|
|
label = f"{t}" + (f":{reason}" if reason else "")
|
||
|
|
taxonomy[label] += 1
|
||
|
|
failures.append(e)
|
||
|
|
|
||
|
|
# followup nudge failures live in job-log.jsonl
|
||
|
|
for e in iter_log(JOB_LOG, cutoff):
|
||
|
|
if e.get("type") == "followup_nudge_failed":
|
||
|
|
taxonomy["followup_nudge_failed"] += 1
|
||
|
|
failures.append({"type": "followup_nudge_failed",
|
||
|
|
"ts": e.get("ts"), "id": e.get("dm_id"),
|
||
|
|
"agent": "sweeper", "to": e.get("recipient"),
|
||
|
|
"target": e.get("target"),
|
||
|
|
"reason": (e.get("error") or "")[:100]})
|
||
|
|
|
||
|
|
print(f"== placement audit: last {args.since} (since {cutoff.strftime('%Y-%m-%d %H:%M UTC')}) ==")
|
||
|
|
print(f"dm-log events scanned: {total}")
|
||
|
|
print()
|
||
|
|
print("-- per-target sends --")
|
||
|
|
print(f"{'agent':10s} {'target':28s} {'sent':>5s} {'verified':>8s} {'unver':>6s}")
|
||
|
|
for (agent, target), n in sorted(sends.items(), key=lambda kv: -kv[1]):
|
||
|
|
ok = sent_ok.get((agent, target), 0)
|
||
|
|
print(f"{agent:10s} {target:28s} {n:5d} {ok:8d} {n - ok:6d}")
|
||
|
|
if not sends:
|
||
|
|
print("(no sends in window)")
|
||
|
|
print()
|
||
|
|
print("-- failure taxonomy --")
|
||
|
|
if taxonomy:
|
||
|
|
for label, n in taxonomy.most_common():
|
||
|
|
print(f"{n:5d} {label}")
|
||
|
|
else:
|
||
|
|
print("(no placement failures in window)")
|
||
|
|
print()
|
||
|
|
print("-- 10 most recent failures --")
|
||
|
|
failures.sort(key=lambda e: parse_ts(e.get("ts")) or datetime.min.replace(tzinfo=timezone.utc),
|
||
|
|
reverse=True)
|
||
|
|
for e in failures[:10]:
|
||
|
|
print(one_line(e))
|
||
|
|
if not failures:
|
||
|
|
print("(none)")
|
||
|
|
return 0
|
||
|
|
|
||
|
|
|
||
|
|
def cmd_watch(_args):
|
||
|
|
print("watching dm-log.jsonl for placement events (Ctrl-C to stop)...", file=sys.stderr)
|
||
|
|
try:
|
||
|
|
f = open(DM_LOG, encoding="utf-8")
|
||
|
|
except OSError as ex:
|
||
|
|
print(f"cannot open {DM_LOG}: {ex}", file=sys.stderr)
|
||
|
|
return 1
|
||
|
|
with f:
|
||
|
|
f.seek(0, 2) # start at end: live view only
|
||
|
|
try:
|
||
|
|
while True:
|
||
|
|
line = f.readline()
|
||
|
|
if not line:
|
||
|
|
time.sleep(2)
|
||
|
|
continue
|
||
|
|
line = line.strip()
|
||
|
|
if not line:
|
||
|
|
continue
|
||
|
|
try:
|
||
|
|
e = json.loads(line)
|
||
|
|
except json.JSONDecodeError:
|
||
|
|
continue
|
||
|
|
if e.get("type") in WATCH_TYPES_DM:
|
||
|
|
print(one_line(e), flush=True)
|
||
|
|
except KeyboardInterrupt:
|
||
|
|
print("\nstopped.", file=sys.stderr)
|
||
|
|
return 0
|
||
|
|
|
||
|
|
|
||
|
|
def main(argv=None):
|
||
|
|
ap = argparse.ArgumentParser(description="Audit sidechat placement / gate events in dm-log.jsonl")
|
||
|
|
ap.add_argument("--since", metavar="60m|24h|7d",
|
||
|
|
help="summarize events newer than this (e.g. 60m, 24h, 7d)")
|
||
|
|
ap.add_argument("--watch", action="store_true",
|
||
|
|
help="live tail of placement-relevant events")
|
||
|
|
args = ap.parse_args(argv)
|
||
|
|
if args.watch:
|
||
|
|
return cmd_watch(args)
|
||
|
|
if args.since:
|
||
|
|
return cmd_since(args)
|
||
|
|
ap.print_help()
|
||
|
|
return 2
|
||
|
|
|
||
|
|
|
||
|
|
if __name__ == "__main__":
|
||
|
|
sys.exit(main())
|