169 lines
6.8 KiB
Python
169 lines
6.8 KiB
Python
"""alerts 告警事件存储(spec §20.15.4)——监控脚本单写、sanguo_api 只读+ack、NAS 侧经 inbox 摄取.
|
|
|
|
事件模型: 确定性 alert_id 幂等去重(重 Sight=同 id 再 upsert → count++ 即「第 N 晚」升级;
|
|
severity 跟随最新评估——同日黄→红升级即时落表, AL-1 10-05);
|
|
check_key(去日期维度)驱动转绿自动 resolved;人工只 ack 不 resolved。
|
|
保留期: resolved 30 天 / open 黄按 last_seen 28 天(MON-202610-04 收口 10-05:
|
|
活着不删) / open 红不删, 每班 purge。
|
|
"""
|
|
import json
|
|
import os
|
|
import sqlite3
|
|
from datetime import datetime, timedelta
|
|
|
|
_SCHEMA = """
|
|
CREATE TABLE IF NOT EXISTS alerts (
|
|
alert_id TEXT PRIMARY KEY,
|
|
check_key TEXT NOT NULL,
|
|
source TEXT NOT NULL,
|
|
host TEXT NOT NULL,
|
|
severity TEXT NOT NULL,
|
|
title TEXT NOT NULL,
|
|
detail TEXT NOT NULL DEFAULT '',
|
|
evidence TEXT NOT NULL DEFAULT '{}',
|
|
first_seen TEXT NOT NULL,
|
|
last_seen TEXT NOT NULL,
|
|
count INTEGER NOT NULL DEFAULT 1,
|
|
status TEXT NOT NULL DEFAULT 'open',
|
|
acked_at TEXT,
|
|
acked_by TEXT,
|
|
resolved_at TEXT
|
|
);
|
|
CREATE INDEX IF NOT EXISTS idx_alerts_status_severity ON alerts(status, severity);
|
|
CREATE INDEX IF NOT EXISTS idx_alerts_check_key ON alerts(check_key);
|
|
"""
|
|
|
|
_FIELDS = ("alert_id check_key source host severity title detail evidence "
|
|
"first_seen last_seen count status acked_at acked_by resolved_at").split()
|
|
|
|
|
|
def default_alerts_db_path() -> str:
|
|
"""env SANGUO_ALERTS_DB 覆盖; 缺省 data/alerts.db(相对 cwd=各机数据根)."""
|
|
return os.environ.get("SANGUO_ALERTS_DB", os.path.join("data", "alerts.db"))
|
|
|
|
|
|
def connect(path: str) -> sqlite3.Connection:
|
|
d = os.path.dirname(os.path.abspath(path))
|
|
os.makedirs(d, exist_ok=True)
|
|
conn = sqlite3.connect(path, timeout=30)
|
|
conn.execute("PRAGMA busy_timeout=30000")
|
|
conn.row_factory = sqlite3.Row
|
|
conn.executescript(_SCHEMA)
|
|
return conn
|
|
|
|
|
|
def _now() -> str:
|
|
return datetime.now().isoformat(timespec="seconds")
|
|
|
|
|
|
def upsert_alert(conn: sqlite3.Connection, event: dict) -> str:
|
|
"""幂等写入一条告警事件; 重 Sight(同 alert_id 仍 open/acked)→count++/
|
|
last_seen 刷新/severity 跟随最新评估(AL-1 10-05: 同日黄→红升级不再丢,
|
|
旧版 UPDATE 不动 severity=次日新 id 才纠正, open_red 角标少计一晚)."""
|
|
now = _now()
|
|
alert_id = event["alert_id"]
|
|
cur = conn.execute(
|
|
"UPDATE alerts SET count=count+1, last_seen=?, detail=?, evidence=?, "
|
|
"title=?, severity=? "
|
|
"WHERE alert_id=? AND status IN ('open','acked')",
|
|
(now, event.get("detail", ""),
|
|
json.dumps(event.get("evidence", {}), ensure_ascii=False),
|
|
event["title"], event["severity"], alert_id))
|
|
if cur.rowcount == 0:
|
|
exists = conn.execute(
|
|
"SELECT 1 FROM alerts WHERE alert_id=?", (alert_id,)).fetchone()
|
|
if exists: # 已 resolved 的同日 id 不复活(当日转绿又告=罕见, 次日新 id)
|
|
return alert_id
|
|
conn.execute(
|
|
"INSERT INTO alerts (alert_id, check_key, source, host, severity, title,"
|
|
" detail, evidence, first_seen, last_seen, count, status)"
|
|
" VALUES (?,?,?,?,?,?,?,?,?,?,1,'open')",
|
|
(alert_id, event["check_key"], event["source"], event["host"],
|
|
event["severity"], event["title"], event.get("detail", ""),
|
|
json.dumps(event.get("evidence", {}), ensure_ascii=False), now, now))
|
|
conn.commit()
|
|
return alert_id
|
|
|
|
|
|
def resolve_passed(conn: sqlite3.Connection, passed_keys, now: str = None) -> int:
|
|
"""本班评估为绿的 check_key → 其 open/acked 告警全部自动 resolved."""
|
|
now = now or _now()
|
|
n = 0
|
|
for key in passed_keys:
|
|
cur = conn.execute(
|
|
"UPDATE alerts SET status='resolved', resolved_at=? "
|
|
"WHERE check_key=? AND status IN ('open','acked')", (now, key))
|
|
n += cur.rowcount
|
|
conn.commit()
|
|
return n
|
|
|
|
|
|
def purge_expired(conn: sqlite3.Connection, now: str = None) -> int:
|
|
"""resolved 留 30 天 / open 黄按 last_seen 留 28 天(MON-202610-04 收口
|
|
10-05: 旧 first_seen 14d 硬删把每日重 Sight 的持续黄无声从值班页抹掉+
|
|
count 升级链断——last_seen 每次重 Sight 刷新=「活着不删」, 僵死 28d 才清;
|
|
resolved 黄统一走 resolved_at 30d, open 红永不删)."""
|
|
now = now or _now()
|
|
t = datetime.fromisoformat(now)
|
|
cut_resolved = (t - timedelta(days=30)).isoformat(timespec="seconds")
|
|
cut_open_yellow = (t - timedelta(days=28)).isoformat(timespec="seconds")
|
|
n = conn.execute(
|
|
"DELETE FROM alerts WHERE (status='resolved' AND resolved_at<?)"
|
|
" OR (status IN ('open','acked') AND severity='yellow'"
|
|
" AND last_seen<?)", (cut_resolved, cut_open_yellow)
|
|
).rowcount
|
|
conn.commit()
|
|
return n
|
|
|
|
|
|
def list_alerts(conn, source=None, severity=None, status=None, host=None,
|
|
limit=500, offset=0):
|
|
where, args = [], []
|
|
for col, val in (("source", source), ("severity", severity),
|
|
("status", status), ("host", host)):
|
|
if val:
|
|
where.append(f"{col}=?")
|
|
args.append(val)
|
|
cond = (" WHERE " + " AND ".join(where)) if where else ""
|
|
total = conn.execute("SELECT COUNT(*) FROM alerts" + cond, args).fetchone()[0]
|
|
rows = conn.execute(
|
|
f"SELECT * FROM alerts{cond} ORDER BY last_seen DESC LIMIT ? OFFSET ?",
|
|
args + [limit, offset]).fetchall()
|
|
out = []
|
|
for r in rows:
|
|
d = {k: r[k] for k in _FIELDS}
|
|
try:
|
|
d["evidence"] = json.loads(d["evidence"] or "{}")
|
|
except (ValueError, TypeError):
|
|
pass
|
|
out.append(d)
|
|
return out, total
|
|
|
|
|
|
def ack_alert(conn, alert_id: str, acked_by: str, now: str = None):
|
|
"""人工只 ack(open→acked); resolved 再 ack 返回 None(调用方 400)."""
|
|
now = now or _now()
|
|
cur = conn.execute(
|
|
"UPDATE alerts SET status='acked', acked_at=?, acked_by=? "
|
|
"WHERE alert_id=? AND status='open'", (now, acked_by, alert_id))
|
|
conn.commit()
|
|
if cur.rowcount:
|
|
return "acked"
|
|
if conn.execute("SELECT 1 FROM alerts WHERE alert_id=?",
|
|
(alert_id,)).fetchone() is None:
|
|
raise KeyError(alert_id)
|
|
return None
|
|
|
|
|
|
def summary(conn) -> dict:
|
|
"""导航角标数据: open_red=红项 open/acked 数; by_source=四类在案数."""
|
|
open_red = conn.execute(
|
|
"SELECT COUNT(*) FROM alerts WHERE status IN ('open','acked')"
|
|
" AND severity='red'").fetchone()[0]
|
|
by_source = {}
|
|
for s in ("data", "strategy", "factor", "infra"):
|
|
by_source[s] = conn.execute(
|
|
"SELECT COUNT(*) FROM alerts WHERE status IN ('open','acked')"
|
|
" AND source=?", (s,)).fetchone()[0]
|
|
return {"open_red": open_red, "by_source": by_source}
|