Hermes Agent
4a1d949e41
完整 Web 管理平台,涵盖: P0 生产加固: - 会话超时(空闲+绝对) + CSRF 防护 + 登录失败限流 - HTTPS/TLS 证书管理(自签生成/上传/删除/过期提醒) - SSL Bump (HTTPS 透明代理) 可视化配置 - 配置文件 diff 预览 + 二次确认才写入 - 服务控制二次确认 + 操作期间按钮锁定 P1 运维能力: - 日志轮转管理(logrotate) + 日志状态监控 - 告警规则(5xx/命中率/磁盘/客户端流量) - 多 Squid 实例管理 + 一键切换 - squidclient mgr 实时性能指标 - 流量异常检测(突增/大文件/高频小包/新客户端) P2 日志分析增强: - 日志持久化到 SQLite + 历史时间范围查询 - 导出 CSV/JSON(日志/KPI/异常/审计) - GeoIP 地图(Leaflet + ip-api.com 离线可选 GeoLite2) - 每客户端 24h 趋势图 - 自定义 Squid logformat 解析器 P3 体验锦上添花: - 暗/亮主题切换(cookie 持久化) - WebSSH 终端(xterm.js + paramiko) - 完整审计日志 + 用户管理 技术栈: Flask 3.0 + SQLAlchemy 2.0 + SQLite + Chart.js + Leaflet 总代码量: ~16500 行 (15 Python 模块 + 34 模板 + CSS) 路由数: 73
436 lines
15 KiB
Python
436 lines
15 KiB
Python
"""alerts.py - Alert rule engine for Squid Web Manager (P1-2).
|
|
|
|
Provides:
|
|
- AlertRule dataclass - configurable metric threshold rule
|
|
- evaluate_rules(entries, rules, now=None) - run all enabled rules against
|
|
the current access.log stats and return a list of triggered alerts.
|
|
- save_rules / load_rules - persist rule list as JSON
|
|
- default_rules() - opinionated starting set that covers the common SRE cases
|
|
|
|
All metric computation uses only Python stdlib + existing log_parser statistics
|
|
- no new dependencies. Audit writes use a lazy import of `app` so this module
|
|
can be imported without forcing a circular dependency on app.py at startup.
|
|
"""
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import os
|
|
import shutil
|
|
import time
|
|
from dataclasses import asdict, dataclass, field
|
|
from typing import Any, Iterable
|
|
|
|
# --- Supported metrics ----------------------------------------------------
|
|
# Each entry documents the operator semantics. Values are floats unless noted.
|
|
METRICS: dict[str, str] = {
|
|
"5xx_rate": "HTTP 5xx 错误占比 (0.0 - 1.0)",
|
|
"4xx_rate": "HTTP 4xx 错误占比 (0.0 - 1.0)",
|
|
"hit_ratio": "缓存命中率 (0.0 - 1.0)",
|
|
"denied_rate": "ACL 拒绝请求占比 (0.0 - 1.0)",
|
|
"disk_usage_pct": "缓存目录磁盘使用率 (0.0 - 100.0)",
|
|
"client_bytes_per_min": "Top 客户端每分钟流量 (字节/分钟)",
|
|
}
|
|
|
|
OPERATORS: dict[str, str] = {
|
|
"gt": ">",
|
|
"lt": "<",
|
|
"gte": ">=",
|
|
"lte": "<=",
|
|
}
|
|
|
|
SEVERITIES = ("info", "warning", "critical")
|
|
|
|
|
|
# --- Rule dataclass -------------------------------------------------------
|
|
|
|
@dataclass
|
|
class AlertRule:
|
|
"""One alert rule. Persisted as JSON via save_rules / load_rules."""
|
|
|
|
name: str
|
|
metric: str
|
|
operator: str
|
|
threshold: float
|
|
window_minutes: int = 5
|
|
enabled: bool = True
|
|
cooldown_minutes: int = 30
|
|
last_triggered: float = 0.0
|
|
severity: str = "warning"
|
|
notify_webhook: str = ""
|
|
|
|
def to_dict(self) -> dict:
|
|
d = asdict(self)
|
|
# JSON doesn't love bare floats like 0.0 - keep it simple.
|
|
return d
|
|
|
|
@classmethod
|
|
def from_dict(cls, d: dict) -> "AlertRule":
|
|
"""Build from a dict (e.g. one loaded from JSON). Missing keys fall
|
|
back to dataclass defaults so old config files keep working."""
|
|
if not isinstance(d, dict):
|
|
raise ValueError("AlertRule.from_dict expects a dict")
|
|
# Only forward the keys we know about; ignore extras.
|
|
allowed = {f for f in cls.__dataclass_fields__.keys()} # type: ignore[attr-defined]
|
|
kwargs = {k: d[k] for k in d.keys() & allowed}
|
|
# Coerce numerics - if the file is hand-edited we don't want to crash.
|
|
for key in ("threshold", "last_triggered"):
|
|
if key in kwargs:
|
|
try:
|
|
kwargs[key] = float(kwargs[key])
|
|
except (TypeError, ValueError):
|
|
kwargs[key] = 0.0
|
|
for key in ("window_minutes", "cooldown_minutes"):
|
|
if key in kwargs:
|
|
try:
|
|
kwargs[key] = int(kwargs[key])
|
|
except (TypeError, ValueError):
|
|
kwargs[key] = 5 if key == "window_minutes" else 30
|
|
kwargs["enabled"] = bool(kwargs.get("enabled", True))
|
|
# Severity / operator / metric - fall back if hand-edited to junk.
|
|
sev = kwargs.get("severity", "warning")
|
|
if sev not in SEVERITIES:
|
|
sev = "warning"
|
|
kwargs["severity"] = sev
|
|
op = kwargs.get("operator", "gt")
|
|
if op not in OPERATORS:
|
|
op = "gt"
|
|
kwargs["operator"] = op
|
|
return cls(**kwargs)
|
|
|
|
|
|
# --- Metric evaluation ---------------------------------------------------
|
|
|
|
def _apply_operator(value: float, op: str, threshold: float) -> bool:
|
|
if op == "gt":
|
|
return value > threshold
|
|
if op == "lt":
|
|
return value < threshold
|
|
if op == "gte":
|
|
return value >= threshold
|
|
if op == "lte":
|
|
return value <= threshold
|
|
return False
|
|
|
|
|
|
def _compute_metric(
|
|
metric: str,
|
|
stats: dict,
|
|
cache_dir: str | None,
|
|
window_minutes: int,
|
|
) -> float:
|
|
"""Compute the current value of a metric from pre-aggregated stats.
|
|
|
|
For most metrics this is just `stats[metric]`. For client_bytes_per_min we
|
|
fold in the time window (which is configurable on the rule, so we can't
|
|
pre-bake it into log_parser). disk_usage_pct needs shutil.disk_usage.
|
|
"""
|
|
if metric == "5xx_rate":
|
|
return float(stats.get("errors_5xx_rate") or 0.0)
|
|
if metric == "4xx_rate":
|
|
return float(stats.get("errors_4xx_rate") or 0.0)
|
|
if metric == "hit_ratio":
|
|
return float(stats.get("hit_ratio") or 0.0)
|
|
if metric == "denied_rate":
|
|
return float(stats.get("denied_rate") or 0.0)
|
|
if metric == "disk_usage_pct":
|
|
target = cache_dir or "/var/spool/squid"
|
|
try:
|
|
total, used, free = shutil.disk_usage(target)
|
|
except (FileNotFoundError, PermissionError, OSError):
|
|
return 0.0
|
|
if not total:
|
|
return 0.0
|
|
return round(used * 100.0 / total, 2)
|
|
if metric == "client_bytes_per_min":
|
|
tops = stats.get("top_clients") or []
|
|
if not tops:
|
|
return 0.0
|
|
max_bytes = max((c.get("bytes") or 0) for c in tops)
|
|
# Approximate bytes-per-minute for the top client over the rule window.
|
|
# If the time span is < window_minutes we still divide by window so
|
|
# the operator gets a comparable value (long window = low rate).
|
|
w = max(1, int(window_minutes))
|
|
return float(max_bytes) * 60.0 / w
|
|
return 0.0
|
|
|
|
|
|
def _humanise_metric(metric: str, value: float) -> str:
|
|
if metric.endswith("_rate"):
|
|
return f"{value * 100:.2f}%"
|
|
if metric == "hit_ratio":
|
|
return f"{value * 100:.1f}%"
|
|
if metric == "disk_usage_pct":
|
|
return f"{value:.1f}%"
|
|
if metric == "client_bytes_per_min":
|
|
# Display in KB/MB/GB - keep parity with log_parser.format_bytes shape.
|
|
n = float(value)
|
|
for unit in ("B/s", "KB/s", "MB/s", "GB/s"):
|
|
if abs(n) < 1024.0:
|
|
if unit == "B/s":
|
|
return f"{n:.0f} {unit}"
|
|
return f"{n:.1f} {unit}"
|
|
n /= 1024.0
|
|
return f"{n:.2f} TB/s"
|
|
return f"{value:.4f}"
|
|
|
|
|
|
def _format_message(rule: AlertRule, value: float, window_minutes: int) -> str:
|
|
op = OPERATORS.get(rule.operator, rule.operator)
|
|
human = _humanise_metric(rule.metric, value)
|
|
metric_desc = METRICS.get(rule.metric, rule.metric)
|
|
return (
|
|
f"{rule.name}: {metric_desc} 当前值 {human} {op} 阈值 "
|
|
f"{_humanise_metric(rule.metric, rule.threshold)} "
|
|
f"(窗口 {window_minutes}m, 严重度 {rule.severity})"
|
|
)
|
|
|
|
|
|
# --- Audit (lazy import to avoid circular reference) ---------------------
|
|
|
|
def _audit_trigger(rule: AlertRule, value: float, message: str):
|
|
"""Try to write a row into the audit_log. Silently swallow on failure
|
|
so a missing app context never breaks evaluation."""
|
|
try:
|
|
from app import audit, db # type: ignore
|
|
audit(
|
|
"alert_triggered",
|
|
f"name={rule.name} metric={rule.metric} "
|
|
f"value={value:.6f} threshold={rule.threshold} "
|
|
f"severity={rule.severity} msg={message[:300]}",
|
|
)
|
|
db.session.commit()
|
|
except Exception:
|
|
# Never let an audit failure mask a real alert.
|
|
try:
|
|
from app import db # type: ignore
|
|
db.session.rollback()
|
|
except Exception:
|
|
pass
|
|
|
|
|
|
def _post_webhook(rule: AlertRule, payload: dict):
|
|
"""Fire-and-forget webhook POST. We import urllib lazily; if the URL is
|
|
bad or the import fails we just log silently - alerts mustn't crash
|
|
the calling request."""
|
|
url = (rule.notify_webhook or "").strip()
|
|
if not url:
|
|
return
|
|
try:
|
|
import urllib.request
|
|
import urllib.error
|
|
body = json.dumps(payload).encode("utf-8")
|
|
req = urllib.request.Request(
|
|
url,
|
|
data=body,
|
|
headers={"Content-Type": "application/json"},
|
|
method="POST",
|
|
)
|
|
urllib.request.urlopen(req, timeout=5).read()
|
|
except Exception:
|
|
# We deliberately don't surface webhook errors - delivery is
|
|
# best-effort and the audit row is the source of truth.
|
|
pass
|
|
|
|
|
|
# --- Main entry point ----------------------------------------------------
|
|
|
|
def evaluate_rules(
|
|
entries: list | None,
|
|
rules: Iterable[AlertRule],
|
|
*,
|
|
cache_dir: str | None = None,
|
|
now: float | None = None,
|
|
) -> list[dict]:
|
|
"""Evaluate all enabled rules against the current access.log parsing.
|
|
|
|
Parameters
|
|
----------
|
|
entries : list of log_parser.LogEntry (may be empty / None).
|
|
rules : iterable of AlertRule to check.
|
|
cache_dir : path to the Squid cache dir. Used for disk_usage_pct. If
|
|
None, defaults to /var/spool/squid.
|
|
now : epoch seconds to treat as "now". Defaults to time.time().
|
|
|
|
Returns
|
|
-------
|
|
list of dict:
|
|
{rule, value, severity, message, ts, triggered}
|
|
Each dict contains:
|
|
- rule : the AlertRule instance that matched
|
|
- value : the float value that triggered the rule
|
|
- severity : copy of rule.severity
|
|
- message : human-readable summary
|
|
- ts : epoch seconds when this was evaluated
|
|
- triggered : bool - True if outside cooldown, False if suppressed
|
|
The list always contains one entry per *evaluated* rule; rows with
|
|
triggered=False are useful for the UI to show "currently OK / cooldown".
|
|
"""
|
|
ts = float(now if now is not None else time.time())
|
|
stats = _safe_stats(entries)
|
|
out: list[dict] = []
|
|
rules_list = list(rules)
|
|
for rule in rules_list:
|
|
if not getattr(rule, "enabled", True):
|
|
continue
|
|
metric = getattr(rule, "metric", "")
|
|
op = getattr(rule, "operator", "gt")
|
|
threshold = float(getattr(rule, "threshold", 0.0) or 0.0)
|
|
window = max(1, int(getattr(rule, "window_minutes", 5) or 5))
|
|
try:
|
|
value = _compute_metric(metric, stats, cache_dir, window)
|
|
except Exception:
|
|
value = 0.0
|
|
triggered = _apply_operator(value, op, threshold)
|
|
in_cooldown = False
|
|
if triggered:
|
|
last = float(getattr(rule, "last_triggered", 0.0) or 0.0)
|
|
cooldown = max(0, int(getattr(rule, "cooldown_minutes", 30) or 30))
|
|
if last and (ts - last) < cooldown * 60:
|
|
# Suppress but still report so the UI can show "cooling down".
|
|
triggered = False
|
|
in_cooldown = True
|
|
else:
|
|
rule.last_triggered = ts
|
|
msg = _format_message(rule, value, window)
|
|
if triggered:
|
|
_audit_trigger(rule, value, msg)
|
|
_post_webhook(
|
|
rule,
|
|
{
|
|
"name": rule.name,
|
|
"metric": rule.metric,
|
|
"value": value,
|
|
"threshold": threshold,
|
|
"operator": op,
|
|
"severity": rule.severity,
|
|
"message": msg,
|
|
"ts": ts,
|
|
},
|
|
)
|
|
out.append({
|
|
"rule": rule,
|
|
"value": value,
|
|
"threshold": threshold,
|
|
"operator": op,
|
|
"severity": rule.severity,
|
|
"message": msg,
|
|
"ts": ts,
|
|
"triggered": triggered,
|
|
"in_cooldown": in_cooldown,
|
|
})
|
|
return out
|
|
|
|
|
|
def _safe_stats(entries: list | None) -> dict:
|
|
"""Run log_parser.aggregate_stats if available, otherwise return {}.
|
|
Import log_parser lazily so unit tests / minimal contexts don't crash."""
|
|
if not entries:
|
|
return {}
|
|
try:
|
|
import log_parser # type: ignore
|
|
return log_parser.aggregate_stats(entries) or {}
|
|
except Exception:
|
|
return {}
|
|
|
|
|
|
# --- Persistence --------------------------------------------------------
|
|
|
|
def save_rules(rules: list[AlertRule], path: str) -> None:
|
|
"""Write the rule list to `path` as pretty-printed JSON."""
|
|
payload = [r.to_dict() for r in (rules or [])]
|
|
os.makedirs(os.path.dirname(path) or ".", exist_ok=True)
|
|
tmp = path + ".tmp"
|
|
with open(tmp, "w", encoding="utf-8") as f:
|
|
json.dump(payload, f, ensure_ascii=False, indent=2)
|
|
f.flush()
|
|
os.fsync(f.fileno())
|
|
os.replace(tmp, path)
|
|
|
|
|
|
def load_rules(path: str) -> list[AlertRule]:
|
|
"""Load rules from a JSON file. Missing / corrupt files yield the
|
|
`default_rules()` set; that way the UI always has something to show."""
|
|
try:
|
|
with open(path, "r", encoding="utf-8") as f:
|
|
data = json.load(f)
|
|
except FileNotFoundError:
|
|
return default_rules()
|
|
except (json.JSONDecodeError, OSError, ValueError):
|
|
return default_rules()
|
|
if not isinstance(data, list):
|
|
return default_rules()
|
|
out: list[AlertRule] = []
|
|
for item in data:
|
|
try:
|
|
out.append(AlertRule.from_dict(item))
|
|
except Exception:
|
|
# Skip individual bad rows instead of nuking the whole file.
|
|
continue
|
|
return out or default_rules()
|
|
|
|
|
|
def default_rules() -> list[AlertRule]:
|
|
"""A balanced starter set: catches the common SRE cases without being
|
|
noisy. Operators can tune thresholds or disable rules from the UI."""
|
|
return [
|
|
AlertRule(
|
|
name="5xx 错误率突增",
|
|
metric="5xx_rate",
|
|
operator="gt",
|
|
threshold=0.05, # > 5% 5xx
|
|
window_minutes=5,
|
|
cooldown_minutes=30,
|
|
severity="critical",
|
|
),
|
|
AlertRule(
|
|
name="缓存命中率下跌",
|
|
metric="hit_ratio",
|
|
operator="lt",
|
|
threshold=0.20, # < 20%
|
|
window_minutes=5,
|
|
cooldown_minutes=60,
|
|
severity="warning",
|
|
),
|
|
AlertRule(
|
|
name="磁盘空间不足",
|
|
metric="disk_usage_pct",
|
|
operator="gte",
|
|
threshold=90.0, # >= 90%
|
|
window_minutes=5,
|
|
cooldown_minutes=60,
|
|
severity="critical",
|
|
),
|
|
AlertRule(
|
|
name="客户端流量异常",
|
|
metric="client_bytes_per_min",
|
|
operator="gt",
|
|
threshold=200 * 1024 * 1024, # > 200 MB/min
|
|
window_minutes=5,
|
|
cooldown_minutes=15,
|
|
severity="warning",
|
|
),
|
|
AlertRule(
|
|
name="拒绝率过高",
|
|
metric="denied_rate",
|
|
operator="gt",
|
|
threshold=0.50, # > 50%
|
|
window_minutes=5,
|
|
cooldown_minutes=30,
|
|
severity="warning",
|
|
),
|
|
]
|
|
|
|
|
|
# --- CLI smoke test ------------------------------------------------------
|
|
|
|
if __name__ == "__main__": # pragma: no cover - manual sanity check
|
|
import log_parser as _lp
|
|
sample = os.path.join(os.path.dirname(__file__), "sample_access.log")
|
|
text = open(sample, "r", encoding="utf-8").read() if os.path.exists(sample) else ""
|
|
entries = _lp.parse_lines(text) if text else []
|
|
triggered = evaluate_rules(entries, default_rules())
|
|
print(f"triggered: {len(triggered)}")
|
|
for t in triggered:
|
|
marker = "*" if t["triggered"] else " "
|
|
print(f" {marker} {t['rule'].name}: value={t['value']:.4f} threshold={t['threshold']} ({t['message']})")
|