whqreza
b44816aa98
- tools/monitor.py: weekly SQL digest over bot_traces/bot_trace_events — pipeline funnel, Jev health (failures/latency/tokens/model drift vs pinned jev-1.13.0), user-choice quality proxy (Jev↔user A/B), difflib fallback volume, STT provider health, and the plan-§4.5 archive guard (>5GB or >100k events/month -> «archive, do not delete» session); real-traffic only by default (calibration/test traces excluded, --include-calibration shows all); writes markdown to data/reports/; --check exits 1 on warnings - docs/DEPLOY-AND-MONITORING.md: deploy checklist — server env vars (TypeSafe key via Coolify env or bot/.env), recreate-not-restart for new mounts, single-Bale-poller rule, smoke tests per plan §7.12, one-env rollback (JEV_ENABLED/OBS_ENABLED=false); bot server deployment deliberately stays an owner decision (no service added to the Coolify compose)
293 lines
14 KiB
Python
293 lines
14 KiB
Python
"""Phase-7 weekly monitoring digest (plan §7) — all from the observability tables.
|
||
|
||
docker compose exec bot python tools/monitor.py [--days 7] [--check] [--include-calibration]
|
||
|
||
- Prints a Persian digest and writes the same content as markdown to
|
||
`data/reports/monitoring_<YYYYMMDD_HHMM>.md`.
|
||
- No dashboard by design (owner decision): raw tables + SELECTs are the
|
||
source of truth; this script is the weekly read of them.
|
||
- Default window covers REAL traffic only: traces with source='calibration'
|
||
and the test traces (chat_id=0) are excluded from the stats (pass
|
||
`--include-calibration` to see everything; the volume section always shows
|
||
the full breakdown).
|
||
- `--check` exits 1 when a warning trips (space/volume growth per plan §4.5 —
|
||
>5GB or >100k events/month → the «archive, do not delete» session is due —
|
||
plus Jev failure ratio and model drift) so it can be wired to a scheduler.
|
||
- Nothing is ever deleted here (no-deletion policy).
|
||
"""
|
||
import argparse
|
||
import asyncio
|
||
import sys
|
||
from datetime import datetime, timezone
|
||
from pathlib import Path
|
||
|
||
sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
|
||
|
||
from sqlalchemy import text
|
||
|
||
from app.config import settings
|
||
from app.db import SessionLocal
|
||
|
||
GB = 1024 ** 3
|
||
TABLE_SIZE_WARN = 5 * GB # plan §4.5: >5GB of observability storage
|
||
MONTHLY_EVENTS_WARN = 100_000 # plan §4.5: >100k events / month
|
||
JEV_FAIL_RATIO_WARN = 0.20 # of jev.result events in the window (n>=5)
|
||
|
||
# window predicate reused everywhere: --days 0 → all time
|
||
WIN = "(:days = 0 or created_at >= now() - (:days * interval '1 day'))"
|
||
WIN_E = "(:days = 0 or e.created_at >= now() - (:days * interval '1 day'))"
|
||
# real-traffic filter: keep only traces of genuine chats (no calibration, no chat_id=0 tests)
|
||
REAL = ("trace_id in (select id from bot_traces where coalesce(source,'bale') <> 'calibration' "
|
||
"and coalesce(chat_id, 0) <> 0)")
|
||
REAL_E = ("e.trace_id in (select id from bot_traces where coalesce(source,'bale') <> 'calibration' "
|
||
"and coalesce(chat_id, 0) <> 0)")
|
||
|
||
|
||
def _n(x) -> str:
|
||
try:
|
||
return f"{int(x):,}"
|
||
except (TypeError, ValueError):
|
||
return "—"
|
||
|
||
|
||
def _human(nbytes: int) -> str:
|
||
if not nbytes:
|
||
return "0 B"
|
||
step = 1024.0
|
||
v = float(nbytes)
|
||
for unit in ("B", "KB", "MB", "GB", "TB"):
|
||
if v < step or unit == "TB":
|
||
return f"{v:,.1f} {unit}"
|
||
v /= step
|
||
return f"{v:,.1f} TB"
|
||
|
||
|
||
def _join(rows, fmt) -> str:
|
||
return " · ".join(fmt(r) for r in rows) or "—"
|
||
|
||
|
||
async def build(days: int, include_cal: bool = False) -> tuple[list[str], list[str]]:
|
||
"""Return (digest lines, warnings)."""
|
||
flt = "" if include_cal else f" and {REAL}"
|
||
flt_e = "" if include_cal else f" and {REAL_E}"
|
||
warnings: list[str] = []
|
||
L: list[str] = []
|
||
|
||
async with SessionLocal() as s:
|
||
async def rows(sql, **params):
|
||
return (await s.execute(text(sql), {"days": days, **params})).all()
|
||
|
||
async def one(sql, **params):
|
||
return (await s.execute(text(sql), {"days": days, **params})).first()
|
||
|
||
# ---------------------------------------------------------- volume (always all traces)
|
||
by_src = await rows("""
|
||
select coalesce(source,'?') src, count(*) n from bot_traces
|
||
where (:days = 0 or created_at >= now() - (:days * interval '1 day'))
|
||
group by 1 order by n desc
|
||
""")
|
||
per_day = await rows(f"""
|
||
select to_char(date_trunc('day', created_at), 'MM-DD') d, count(*) n
|
||
from bot_trace_events where {WIN}{flt}
|
||
group by 1 order by min(created_at)
|
||
""")
|
||
total_events = await one(f"select count(*) n from bot_trace_events where {WIN}{flt}")
|
||
|
||
L += [f"پنجره: {'از ابتدا' if days == 0 else f'{days} روز اخیر'}"
|
||
+ ("" if include_cal else " · فقط ترافیک واقعی (بدون calibration/test)"), ""]
|
||
L.append("- منابع روندها: " + _join(by_src, lambda r: f"{r.src}: {_n(r.n)}"))
|
||
if per_day:
|
||
L.append(f"- رخداد{'' if include_cal else ' واقعی'} در روز: "
|
||
+ _join(per_day, lambda r: f"{r.d}={_n(r.n)}")
|
||
+ f" (کل پنجره: {_n(total_events.n)})")
|
||
|
||
# ---------------------------------------------------------- stages / latency
|
||
stages = await rows(f"""
|
||
select stage, count(*) n,
|
||
count(*) filter (where not ok) fails,
|
||
round(percentile_cont(0.5) within group (order by latency_ms)) p50,
|
||
round(percentile_cont(0.95) within group (order by latency_ms)) p95
|
||
from bot_trace_events where {WIN}{flt}
|
||
group by 1 order by n desc
|
||
""")
|
||
L += ["", "### مرحلهها (stage · تعداد · خطا · p50/p95 ms)"]
|
||
if not stages:
|
||
L.append("- —")
|
||
for r in stages:
|
||
lat = f" · {r.p50}/{r.p95}" if r.p50 is not None else ""
|
||
bad = f" · ⚠️ {_n(r.fails)} خطا" if r.fails else ""
|
||
L.append(f"- `{r.stage}`: {_n(r.n)}{bad}{lat}")
|
||
fails = [(r.stage, r.fails) for r in stages if r.fails]
|
||
if fails:
|
||
warnings.append("خطا در مراحل: " + ", ".join(f"{a}×{b}" for a, b in fails))
|
||
|
||
# ---------------------------------------------------------- funnel
|
||
nlu = await one(f"""
|
||
select count(*) n, count(*) filter (where ok) ok from bot_trace_events
|
||
where stage='nlu.result' and {WIN}{flt}
|
||
""")
|
||
mentions = await one(f"""
|
||
select coalesce(sum(jsonb_array_length(payload->'entity_mentions')), 0) n
|
||
from bot_trace_events
|
||
where stage='entities.extracted' and jsonb_typeof(payload->'entity_mentions')='array'
|
||
and {WIN}{flt}
|
||
""")
|
||
jev = await one(f"""
|
||
select count(*) n, count(*) filter (where ok) ok, count(*) filter (where not ok) fail
|
||
from bot_trace_events where stage='jev.result' and {WIN}{flt}
|
||
""")
|
||
route_src = await rows(f"""
|
||
select coalesce(payload->>'source','?') src, count(*) n from bot_trace_events
|
||
where stage='routing.decision' and {WIN}{flt} group by 1 order by n desc
|
||
""")
|
||
jev_status = await rows(f"""
|
||
select coalesce(m->>'status','?') st, count(*) n
|
||
from bot_trace_events e, jsonb_array_elements(e.payload->'mentions') m
|
||
where e.stage='routing.decision' and e.payload->>'source'='jev' and {WIN_E}{flt_e}
|
||
group by 1 order by n desc
|
||
""")
|
||
bands = await rows(f"""
|
||
select case
|
||
when (m->>'confidence')::float >= :ex then 'exact'
|
||
when (m->>'confidence')::float >= :fl then 'ambiguity'
|
||
else 'low' end band, count(*) n
|
||
from bot_trace_events e, jsonb_array_elements(e.payload->'mentions') m
|
||
where e.stage='routing.decision' and e.payload->>'source'='jev' and {WIN_E}{flt_e}
|
||
group by 1 order by n desc
|
||
""", ex=settings.JEV_CONF_EXACT, fl=settings.JEV_CONF_FLOOR)
|
||
difflib = await rows(f"""
|
||
select coalesce(payload->>'status','?') st, count(*) n from bot_trace_events
|
||
where stage='routing.decision' and payload->>'source'='difflib' and {WIN}{flt}
|
||
group by 1 order by n desc
|
||
""")
|
||
commits = await one(f"""
|
||
select count(*) filter (where stage='draft.commit') committed,
|
||
count(*) filter (where stage='draft.cancel') cancelled,
|
||
count(*) filter (where stage='draft.render') rendered
|
||
from bot_trace_events where {WIN}{flt}
|
||
""")
|
||
|
||
L += ["", "### قیف پردازش"]
|
||
L.append(f"- NLU (nlu.result): {_n(nlu.n)} — mentionهای استخراجشده: {_n(mentions.n)}")
|
||
L.append(f"- Jev (jev.result): {_n(jev.n)} درخواست/فیلد — خطا: {_n(jev.fail)}")
|
||
L.append(f"- مسیریابی: " + _join(route_src, lambda r: f"{r.src}: {_n(r.n)}"))
|
||
L.append(f"- وضعیت mentionهای Jev: " + _join(jev_status, lambda r: f"{r.st}: {_n(r.n)}"))
|
||
L.append(f"- باندهای confidence (EXACT={settings.JEV_CONF_EXACT} / FLOOR={settings.JEV_CONF_FLOOR}): "
|
||
+ _join(bands, lambda r: f"{r.band}: {_n(r.n)}"))
|
||
L.append(f"- difflib (fallback): " + _join(difflib, lambda r: f"{r.st}: {_n(r.n)}"))
|
||
L.append(f"- پیشنویس: نمایش {_n(commits.rendered)} · ثبت نهایی {_n(commits.committed)} · لغو {_n(commits.cancelled)}")
|
||
|
||
if jev.n and jev.n >= 5 and jev.fail / jev.n > JEV_FAIL_RATIO_WARN:
|
||
warnings.append(f"نسبت خطای Jev بالا: {jev.fail}/{jev.n}")
|
||
|
||
# ---------------------------------------------------------- user choices (quality proxy)
|
||
user_amb = await rows(f"""
|
||
select coalesce(payload->>'action','?') act, coalesce(payload->>'arg','-') arg, count(*) n
|
||
from bot_trace_events where stage='ambiguity.user_choice' and {WIN}{flt}
|
||
group by 1,2 order by n desc
|
||
""")
|
||
picks = await rows(f"""
|
||
select case when (payload->>'idx')::int = 0 then 'top' else 'other' end k, count(*) n
|
||
from bot_trace_events where stage='user_choice.pick' and {WIN}{flt}
|
||
group by 1 order by n desc
|
||
""")
|
||
L += ["", "### انتخاب کاربر (پروکسی کیفیت — A/B واقعی Jev↔difflib↔کاربر)"]
|
||
L.append("- ابهامها: " + (_join(user_amb, lambda r: f"{r.act}({r.arg}): {_n(r.n)}")
|
||
if user_amb else "— هنوز کاربری در ابهام انتخاب نکرده"))
|
||
top = sum(r.n for r in picks if r.k == "top")
|
||
other = sum(r.n for r in picks if r.k == "other")
|
||
if picks:
|
||
L.append(f"- دکمههای گزینه: گزینه اول (پیشنهاد مدل) {_n(top)} · گزینه دیگر {_n(other)}")
|
||
new_rej = sum(r.n for r in user_amb if r.act == "new")
|
||
if new_rej:
|
||
L.append(f"- «هیچکدام/جدید» در ابهام: {_n(new_rej)} (پیشنهاد مدل رد شد)")
|
||
|
||
# ---------------------------------------------------------- jev health
|
||
tokens = await one(f"""
|
||
select coalesce(sum((payload->'usage'->>'input_tokens')::bigint),0) tin,
|
||
coalesce(sum((payload->'usage'->>'output_tokens')::bigint),0) tout
|
||
from bot_trace_events
|
||
where stage='jev.result' and payload->'usage' is not null and {WIN}{flt}
|
||
""")
|
||
models = await rows(f"""
|
||
select coalesce(model,'?') m, count(*) n from bot_trace_events
|
||
where stage='jev.result' and {WIN}{flt} group by 1 order by n desc
|
||
""")
|
||
L += ["", "### سلامت Jev"]
|
||
L.append(f"- توکن مصرفی (از usage): ورودی {_n(tokens.tin)} · خروجی {_n(tokens.tout)}")
|
||
L.append(f"- مدلها: " + _join(models, lambda r: f"`{r.m}`×{_n(r.n)}"))
|
||
drift = [r.m for r in models if r.m not in ("?", settings.JEV_MODEL)]
|
||
if drift:
|
||
warnings.append(f"انحراف نسخه مدل Jev: {drift} ≠ پین {settings.JEV_MODEL}")
|
||
|
||
# ---------------------------------------------------------- stt health
|
||
stt = await rows(f"""
|
||
select coalesce(payload->>'provider','?') p, count(*) n, count(*) filter (where ok) ok,
|
||
round(percentile_cont(0.5) within group (order by latency_ms)) p50
|
||
from bot_trace_events where stage='stt.result' and {WIN}{flt} group by 1 order by n desc
|
||
""")
|
||
if stt:
|
||
L += ["", "### STT"]
|
||
for r in stt:
|
||
L.append(f"- {r.p}: {_n(r.ok)}/{_n(r.n)} موفق · p50 {r.p50 if r.p50 is not None else '—'} ms")
|
||
|
||
# ---------------------------------------------------------- space / archive guard (always all data)
|
||
sizes = await one("""
|
||
select pg_total_relation_size('bot_trace_events') ev_bytes,
|
||
pg_total_relation_size('bot_traces') tr_bytes,
|
||
pg_total_relation_size('bot_messages') msg_bytes
|
||
""")
|
||
month_events = await one("""
|
||
select count(*) n from bot_trace_events
|
||
where created_at >= date_trunc('month', now())
|
||
""")
|
||
voices_dir = Path(settings.OBS_VOICE_DIR)
|
||
v_files = v_bytes = 0
|
||
if voices_dir.exists():
|
||
for p in voices_dir.rglob("*"):
|
||
if p.is_file():
|
||
v_files += 1
|
||
v_bytes += p.stat().st_size
|
||
obs_bytes = (sizes.ev_bytes or 0) + (sizes.tr_bytes or 0) + v_bytes
|
||
|
||
L += ["", "### فضا و آرشیو (سیاست: آرشیو، نه حذف — پلن §4.5)"]
|
||
L.append(f"- bot_trace_events: {_human(sizes.ev_bytes or 0)} · bot_traces: {_human(sizes.tr_bytes or 0)} "
|
||
f"· bot_messages: {_human(sizes.msg_bytes or 0)} · صدای خام: {_human(v_bytes)} ({_n(v_files)} فایل)")
|
||
L.append(f"- رخدادهای این ماه تقویمی: {_n(month_events.n)}")
|
||
if obs_bytes > TABLE_SIZE_WARN:
|
||
warnings.append(f"حجم Observability از ۵GB گذشت: {_human(obs_bytes)} → جلسه آرشیو لازم است")
|
||
if month_events.n > MONTHLY_EVENTS_WARN:
|
||
warnings.append(f"رخداد این ماه از ۱۰۰هزار گذشت: {_n(month_events.n)} → جلسه آرشیو لازم است")
|
||
|
||
return L, warnings
|
||
|
||
|
||
async def main() -> None:
|
||
ap = argparse.ArgumentParser(description="Phase-7 monitoring digest")
|
||
ap.add_argument("--days", type=int, default=7, help="0 = از ابتدا (default: 7)")
|
||
ap.add_argument("--check", action="store_true", help="exit 1 when a warning trips")
|
||
ap.add_argument("--include-calibration", action="store_true",
|
||
help="شامل ترافیک calibration/test در آمار (پیشفرض: فقط ترافیک واقعی)")
|
||
args = ap.parse_args()
|
||
|
||
ts = datetime.now(timezone.utc).strftime("%Y%m%d_%H%M")
|
||
lines, warnings = await build(args.days, args.include_calibration)
|
||
|
||
header = [f"# پایش هفتگی بات — {ts} UTC", ""]
|
||
if warnings:
|
||
header += ["## ⚠️ هشدارها", *[f"- {w}" for w in warnings], ""]
|
||
|
||
out_dir = Path(settings.OBS_VOICE_DIR).parent / "reports"
|
||
out_dir.mkdir(parents=True, exist_ok=True)
|
||
report = out_dir / f"monitoring_{ts}.md"
|
||
report.write_text("\n".join(header + lines), encoding="utf-8")
|
||
|
||
print("\n".join(header + lines))
|
||
print(f"\nگزارش: {report}")
|
||
if args.check and warnings:
|
||
sys.exit(1)
|
||
|
||
|
||
if __name__ == "__main__":
|
||
asyncio.run(main())
|