62 lines
1.9 KiB
Python
62 lines
1.9 KiB
Python
"""运维告警 API(spec §20.15.4): 只读列表+summary+人工 ack; 写入方=监控脚本(单写者).
|
|
|
|
alert 库路径: env SANGUO_ALERTS_DB > 各机缺省 data/alerts.db(相对 cwd)——
|
|
与监控脚本 wrapper(Set-Location C:\\sanguo_vnpy_v2)解析到同一文件。
|
|
"""
|
|
from fastapi import APIRouter, Depends, HTTPException
|
|
|
|
from sanguo_api.routes import verify_token
|
|
from sanguo_data import alerts_store as store
|
|
|
|
router = APIRouter(dependencies=[Depends(verify_token)])
|
|
|
|
_db = {"path": None}
|
|
|
|
|
|
def set_alerts_db_path(p):
|
|
"""create_app 注入(同 routes_paper.set_db_path 模式); None=缺省解析."""
|
|
_db["path"] = p
|
|
|
|
|
|
def _conn():
|
|
return store.connect(_db["path"] or store.default_alerts_db_path())
|
|
|
|
|
|
@router.get("/alerts")
|
|
def list_alerts(source: str = None, severity: str = None, status: str = None,
|
|
host: str = None, limit: int = 500, offset: int = 0) -> dict:
|
|
limit = max(1, min(limit, 2000))
|
|
offset = max(0, offset)
|
|
conn = _conn()
|
|
try:
|
|
items, total = store.list_alerts(conn, source=source, severity=severity,
|
|
status=status, host=host,
|
|
limit=limit, offset=offset)
|
|
finally:
|
|
conn.close()
|
|
return {"items": items, "total": total}
|
|
|
|
|
|
@router.get("/alerts/summary")
|
|
def alerts_summary() -> dict:
|
|
conn = _conn()
|
|
try:
|
|
return store.summary(conn)
|
|
finally:
|
|
conn.close()
|
|
|
|
|
|
@router.post("/alerts/{alert_id}/ack")
|
|
def ack_alert(alert_id: str) -> dict:
|
|
conn = _conn()
|
|
try:
|
|
try:
|
|
status = store.ack_alert(conn, alert_id, acked_by="user")
|
|
except KeyError:
|
|
raise HTTPException(404, f"alert 不存在: {alert_id}")
|
|
finally:
|
|
conn.close()
|
|
if status is None:
|
|
raise HTTPException(400, "已 resolved 的告警无需 ack")
|
|
return {"alert_id": alert_id, "status": status}
|