88 lines
2.6 KiB
Python
88 lines
2.6 KiB
Python
|
"""Attach alert_logs to fleet metric sample windows (10min flush alignment)."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
from datetime import datetime, timedelta
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
from server.utils.display_time import to_naive_utc
|
||
|
|
|
||
|
|
ALERT_TYPE_LABELS = {
|
||
|
|
"cpu": "CPU",
|
||
|
|
"mem": "内存",
|
||
|
|
"disk": "磁盘",
|
||
|
|
"time_drift": "时钟漂移",
|
||
|
|
"system": "系统",
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def _parse_naive_utc(value: datetime | str | None) -> datetime | None:
|
||
|
|
if value is None:
|
||
|
|
return None
|
||
|
|
if isinstance(value, datetime):
|
||
|
|
return to_naive_utc(value)
|
||
|
|
raw = str(value).strip()
|
||
|
|
if not raw:
|
||
|
|
return None
|
||
|
|
normalized = raw.replace(" ", "T") if " " in raw and "T" not in raw else raw
|
||
|
|
if normalized.endswith("Z"):
|
||
|
|
normalized = normalized[:-1] + "+00:00"
|
||
|
|
try:
|
||
|
|
dt = datetime.fromisoformat(normalized)
|
||
|
|
except ValueError:
|
||
|
|
return None
|
||
|
|
return to_naive_utc(dt)
|
||
|
|
|
||
|
|
|
||
|
|
def _alert_to_dict(alert: Any) -> dict[str, Any]:
|
||
|
|
atype = str(getattr(alert, "alert_type", "") or "")
|
||
|
|
return {
|
||
|
|
"server_name": getattr(alert, "server_name", None) or "",
|
||
|
|
"server_id": getattr(alert, "server_id", None),
|
||
|
|
"alert_type": atype,
|
||
|
|
"alert_type_label": ALERT_TYPE_LABELS.get(atype, atype),
|
||
|
|
"value": getattr(alert, "value", None) or "",
|
||
|
|
"is_recovery": bool(getattr(alert, "is_recovery", False)),
|
||
|
|
"created_at": str(getattr(alert, "created_at", "") or ""),
|
||
|
|
}
|
||
|
|
|
||
|
|
|
||
|
|
def attach_alerts_to_fleet_points(
|
||
|
|
points: list[dict[str, Any]],
|
||
|
|
alerts: list[Any],
|
||
|
|
*,
|
||
|
|
interval_minutes: int = 10,
|
||
|
|
) -> list[dict[str, Any]]:
|
||
|
|
"""Map each sample to alerts in (previous_sample, current_sample] window."""
|
||
|
|
if not points:
|
||
|
|
return []
|
||
|
|
|
||
|
|
parsed: list[tuple[datetime | None, dict[str, Any]]] = []
|
||
|
|
for p in points:
|
||
|
|
parsed.append((_parse_naive_utc(p.get("recorded_at")), p))
|
||
|
|
|
||
|
|
alert_rows: list[tuple[datetime, dict[str, Any]]] = []
|
||
|
|
for a in alerts:
|
||
|
|
adt = _parse_naive_utc(getattr(a, "created_at", None))
|
||
|
|
if adt is not None:
|
||
|
|
alert_rows.append((adt, _alert_to_dict(a)))
|
||
|
|
|
||
|
|
window = timedelta(minutes=max(1, interval_minutes))
|
||
|
|
enriched: list[dict[str, Any]] = []
|
||
|
|
|
||
|
|
for i, (pt_dt, p) in enumerate(parsed):
|
||
|
|
if pt_dt is None:
|
||
|
|
enriched.append({**p, "alerts": [], "alert_count": 0})
|
||
|
|
continue
|
||
|
|
|
||
|
|
prev_dt = parsed[i - 1][0] if i > 0 else None
|
||
|
|
start = prev_dt if prev_dt is not None else pt_dt - window
|
||
|
|
bucket = [
|
||
|
|
item
|
||
|
|
for adt, item in alert_rows
|
||
|
|
if start < adt <= pt_dt
|
||
|
|
]
|
||
|
|
enriched.append({**p, "alerts": bucket, "alert_count": len(bucket)})
|
||
|
|
|
||
|
|
return enriched
|