Files
Nexus/server/utils/watch_metrics.py
2026-07-08 22:31:31 +08:00

563 lines
20 KiB
Python
Raw Permalink Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
"""Watch slot probe — rate calculation and metric normalization."""
from __future__ import annotations
import json
from dataclasses import dataclass
from typing import Any
from server.utils.fleet_metrics import _metric_from_system_info, _safe_pct
WATCH_MAX_SLOTS = 4
WATCH_PIN_TTL_DEFAULT_MINUTES = 30
WATCH_PIN_TTL_ALLOWED_MINUTES = frozenset({30, 60, 120, 480, 1440})
# 兼容旧 API / 列名
WATCH_PIN_TTL_HOURS = 8
WATCH_PIN_TTL_EXTENDED_HOURS = 24
def normalize_watch_ttl_minutes(minutes: int) -> int:
"""监测窗口:30分 / 1h / 2h / 8h / 24h(默认 30 分)。"""
if minutes in WATCH_PIN_TTL_ALLOWED_MINUTES:
return minutes
# 历史:按小时传入
legacy_map = {8: 480, 24: 1440, 1: 60, 2: 120}
if minutes in legacy_map:
return legacy_map[minutes]
return WATCH_PIN_TTL_DEFAULT_MINUTES
def normalize_watch_ttl_hours(hours: int) -> int:
"""Deprecated — use normalize_watch_ttl_minutes(hours * 60) for hour values."""
return normalize_watch_ttl_minutes(hours * 60 if hours in (1, 2, 8, 24) else hours)
def resolve_pin_ttl_minutes(*, ttl_minutes: int | None, ttl_hours: int | None) -> int:
if ttl_minutes and int(ttl_minutes) in WATCH_PIN_TTL_ALLOWED_MINUTES:
return int(ttl_minutes)
if ttl_hours in (8, 24):
return int(ttl_hours) * 60
return WATCH_PIN_TTL_DEFAULT_MINUTES
WATCH_PROBE_INTERVAL_SEC = 5
WATCH_REDIS_FRESH_SEC = 90
WATCH_AGENT_FRESH_SEC = 12 # 2× 5s watch interval + slack
WATCH_AGENT_INTERVAL_SEC = 5
WATCH_PROBE_TIMEOUT_SEC = 8
PROBE_OK = "ok"
PROBE_SSH_TIMEOUT = "ssh_timeout"
PROBE_SSH_AUTH = "ssh_auth"
PROBE_NO_CRED = "no_cred"
PROBE_REDIS_STALE = "redis_stale"
PROBE_PARSE_ERROR = "parse_error"
PROBE_OFFLINE = "offline"
@dataclass
class WatchProbeSample:
probe_status: str
source: str
is_online: bool = False
duration_ms: int = 0
error: str | None = None
cpu_pct: int | None = None
mem_pct: int | None = None
disk_pct: int | None = None
cpu_cores: int | None = None
cpu_cores_physical: int | None = None
cpu_model: str | None = None
memory_total_gb: float | None = None
memory_used_gb: float | None = None
memory_available_gb: float | None = None
disk_total_gb: float | None = None
disk_used_gb: float | None = None
disk_free_gb: float | None = None
swap_total_gb: float | None = None
swap_used_gb: float | None = None
swap_pct: float | None = None
os_release: str | None = None
uptime_seconds: int | None = None
disk_mount: str = "/"
load_1: float | None = None
load_5: float | None = None
load_15: float | None = None
process_running: int | None = None
process_total: int | None = None
per_cpu_pct: list[float] | None = None
mem_free_mb: int | None = None
mem_shared_mb: int | None = None
mem_buffers_mb: int | None = None
mem_cached_mb: int | None = None
net_bytes_sent: int | None = None
net_bytes_recv: int | None = None
disk_read_bytes: int | None = None
disk_write_bytes: int | None = None
net_up_bps: int | None = None
net_down_bps: int | None = None
disk_read_bps: int | None = None
disk_write_bps: int | None = None
processes_json: list[dict[str, Any]] | dict[str, Any] | None = None
_LIVE_HARDWARE_KEYS = (
"cpu_cores",
"cpu_cores_physical",
"cpu_model",
"memory_total_gb",
"memory_used_gb",
"memory_available_gb",
"disk_total_gb",
"disk_used_gb",
"disk_free_gb",
"disk_mount",
"swap_total_gb",
"swap_used_gb",
"swap_pct",
"os_release",
"uptime_seconds",
"process_running",
"process_total",
"per_cpu_pct",
"mem_free_mb",
"mem_shared_mb",
"mem_buffers_mb",
"mem_cached_mb",
)
def to_live_dict(self, *, server_id: int) -> dict[str, Any]:
out: dict[str, Any] = {
"server_id": server_id,
"probe_status": self.probe_status,
"source": self.source,
"is_online": self.is_online,
"cpu_pct": self.cpu_pct,
"mem_pct": self.mem_pct,
"disk_pct": self.disk_pct,
"load_1": self.load_1,
"load_5": self.load_5,
"load_15": self.load_15,
"error": self.error,
}
for key in self._LIVE_HARDWARE_KEYS:
val = getattr(self, key, None)
if val is not None and val != "":
out[key] = val
if self.processes_json:
out["processes"] = self.processes_json
return out
def redis_sample_has_basic_metrics(sample: WatchProbeSample | None) -> bool:
"""True when Agent heartbeat has CPU or memory — IO optional."""
if sample is None or sample.probe_status != PROBE_OK:
return False
return sample.cpu_pct is not None or sample.mem_pct is not None
def parse_heartbeat_system_info(heartbeat: dict[str, str]) -> dict[str, Any]:
raw = heartbeat.get("system_info") or "{}"
try:
info = json.loads(raw) if isinstance(raw, str) else (raw or {})
except json.JSONDecodeError:
return {}
return info if isinstance(info, dict) else {}
def heartbeat_watch_mode_active(heartbeat: dict[str, str]) -> bool:
"""True when Agent explicitly reports watch_mode in the last heartbeat payload."""
return bool(parse_heartbeat_system_info(heartbeat).get("watch_mode"))
def agent_watch_heartbeat_fresh(heartbeat: dict[str, str]) -> bool:
"""Stricter freshness for 5s Agent watch payloads (not 90s fleet heartbeat)."""
from datetime import datetime, timezone
if heartbeat.get("is_online") != "True":
return False
ts = heartbeat.get("last_heartbeat") or ""
if not ts:
return False
try:
agent_ts = datetime.fromisoformat(ts.replace("Z", "+00:00"))
if agent_ts.tzinfo is None:
agent_ts = agent_ts.replace(tzinfo=timezone.utc)
age = (datetime.now(timezone.utc) - agent_ts).total_seconds()
return age <= WATCH_AGENT_FRESH_SEC
except Exception:
return False
def agent_watch_sample_usable(
heartbeat: dict[str, str] | None,
sample: WatchProbeSample | None,
) -> bool:
"""Agent 5s path: fresh heartbeat + watch_mode + basic CPU/mem metrics."""
if not heartbeat or sample is None:
return False
if not agent_watch_heartbeat_fresh(heartbeat):
return False
if not heartbeat_watch_mode_active(heartbeat):
return False
return redis_sample_has_basic_metrics(sample)
def sanitize_processes(raw: Any) -> list[dict[str, Any]] | None:
if not isinstance(raw, list):
return None
out: list[dict[str, Any]] = []
for item in raw[:5]:
if not isinstance(item, dict):
continue
name = str(item.get("name") or "")[:80]
out.append(
{
"pid": item.get("pid"),
"name": name,
"cpu_pct": item.get("cpu_pct"),
"mem_pct": item.get("mem_pct"),
}
)
return out or None
def sanitize_process_bundle(raw: dict[str, Any] | None) -> dict[str, Any] | None:
if not isinstance(raw, dict):
return None
top_cpu = sanitize_processes(raw.get("top_cpu")) or []
top_mem = sanitize_processes(raw.get("top_mem")) or []
legacy = sanitize_processes(raw.get("top_processes"))
if legacy and not top_cpu:
top_cpu = legacy
if not top_cpu and not top_mem:
return None
return {"top_cpu": top_cpu, "top_mem": top_mem}
@dataclass
class _RateState:
net_bytes_sent: int | None = None
net_bytes_recv: int | None = None
disk_read_bytes: int | None = None
disk_write_bytes: int | None = None
recorded_at_ts: float | None = None
def server_can_be_watched(server: Any) -> tuple[bool, str]:
"""Return whether server has Agent or SSH credentials for watch probes."""
if getattr(server, "agent_version", None):
return True, ""
if getattr(server, "password", None) or getattr(server, "ssh_key_private", None):
return True, ""
if getattr(server, "ssh_key_configured", None) or getattr(server, "preset_id", None):
return True, ""
if getattr(server, "ssh_key_preset_id", None):
return True, ""
return False, "需要已安装 Agent 或配置 SSH 凭据"
def _int_pct(value: Any) -> int | None:
pct = _safe_pct(value)
return round(pct) if pct is not None else None
def _positive_int(value: Any) -> int | None:
if value is None:
return None
try:
num = int(value)
except (TypeError, ValueError):
return None
return num if num >= 0 else None
def _cpu_cores(value: Any) -> int | None:
n = _positive_int(value)
return n if n and n > 0 else None
def _float_gb(value: Any) -> float | None:
if value is None:
return None
try:
num = float(value)
except (TypeError, ValueError):
return None
if num < 0:
return None
return round(num, 1)
def _short_text(value: Any, max_len: int = 120) -> str | None:
if value is None:
return None
text = str(value).strip()
if not text:
return None
return text[:max_len]
def _parse_hardware_fields(data: dict[str, Any]) -> dict[str, Any]:
return {
"cpu_cores": _cpu_cores(data.get("cpu_cores")),
"cpu_cores_physical": _cpu_cores(data.get("cpu_cores_physical")),
"cpu_model": _short_text(data.get("cpu_model")),
"memory_total_gb": _float_gb(data.get("memory_total_gb")),
"memory_used_gb": _float_gb(data.get("memory_used_gb")),
"memory_available_gb": _float_gb(data.get("memory_available_gb")),
"disk_total_gb": _float_gb(data.get("disk_total_gb")),
"disk_used_gb": _float_gb(data.get("disk_used_gb")),
"disk_free_gb": _float_gb(data.get("disk_free_gb")),
"swap_total_gb": _float_gb(data.get("swap_total_gb")),
"swap_used_gb": _float_gb(data.get("swap_used_gb")),
"swap_pct": _float_gb(data.get("swap_pct")),
"os_release": _short_text(data.get("os_release")),
"uptime_seconds": _positive_int(data.get("uptime_seconds")),
"process_running": _positive_int(data.get("process_running")),
"process_total": _positive_int(data.get("process_total")),
"per_cpu_pct": _parse_per_cpu_pct(data.get("per_cpu_pct")),
"mem_free_mb": _positive_int(data.get("mem_free_mb")),
"mem_shared_mb": _positive_int(data.get("mem_shared_mb")),
"mem_buffers_mb": _positive_int(data.get("mem_buffers_mb")),
"mem_cached_mb": _positive_int(data.get("mem_cached_mb")),
}
def _parse_per_cpu_pct(raw: Any) -> list[float] | None:
if not isinstance(raw, (list, tuple)):
return None
out: list[float] = []
for item in raw[:128]:
try:
out.append(round(float(item), 1))
except (TypeError, ValueError):
continue
return out or None
def fill_hardware_gaps(sample: WatchProbeSample) -> None:
"""Derive used/free GB from totals + % when probe omitted absolute values."""
if sample.memory_used_gb is None and sample.memory_total_gb is not None and sample.mem_pct is not None:
sample.memory_used_gb = round(sample.memory_total_gb * sample.mem_pct / 100, 1)
if sample.memory_available_gb is None and sample.memory_total_gb is not None and sample.memory_used_gb is not None:
sample.memory_available_gb = round(max(0.0, sample.memory_total_gb - sample.memory_used_gb), 1)
if sample.disk_used_gb is None and sample.disk_total_gb is not None and sample.disk_pct is not None:
sample.disk_used_gb = round(sample.disk_total_gb * sample.disk_pct / 100, 1)
if sample.disk_free_gb is None and sample.disk_total_gb is not None and sample.disk_used_gb is not None:
sample.disk_free_gb = round(max(0.0, sample.disk_total_gb - sample.disk_used_gb), 1)
def _pick_hardware_field(ssh_sample: WatchProbeSample, redis_sample: WatchProbeSample, field: str) -> Any:
sv = getattr(ssh_sample, field, None)
rv = getattr(redis_sample, field, None)
return sv if sv is not None and sv != "" else rv
def compute_rates(
prev: _RateState | None,
*,
net_bytes_sent: int | None,
net_bytes_recv: int | None,
disk_read_bytes: int | None,
disk_write_bytes: int | None,
now_ts: float,
) -> tuple[int | None, int | None, int | None, int | None]:
if prev is None or prev.recorded_at_ts is None:
return None, None, None, None
dt = now_ts - prev.recorded_at_ts
if dt <= 0:
return None, None, None, None
def _rate(curr: int | None, old: int | None) -> int | None:
if curr is None or old is None:
return None
delta = curr - old
if delta < 0:
return None
return int(delta / dt)
return (
_rate(net_bytes_sent, prev.net_bytes_sent),
_rate(net_bytes_recv, prev.net_bytes_recv),
_rate(disk_read_bytes, prev.disk_read_bytes),
_rate(disk_write_bytes, prev.disk_write_bytes),
)
def apply_rates_to_sample(sample: WatchProbeSample, prev: _RateState | None, now_ts: float) -> _RateState:
up, down, dr, dw = compute_rates(
prev,
net_bytes_sent=sample.net_bytes_sent,
net_bytes_recv=sample.net_bytes_recv,
disk_read_bytes=sample.disk_read_bytes,
disk_write_bytes=sample.disk_write_bytes,
now_ts=now_ts,
)
sample.net_up_bps = up
sample.net_down_bps = down
sample.disk_read_bps = dr
sample.disk_write_bps = dw
return _RateState(
net_bytes_sent=sample.net_bytes_sent,
net_bytes_recv=sample.net_bytes_recv,
disk_read_bytes=sample.disk_read_bytes,
disk_write_bytes=sample.disk_write_bytes,
recorded_at_ts=now_ts,
)
def parse_ssh_watch_payload(data: dict[str, Any]) -> WatchProbeSample:
load = data.get("load_avg") or data.get("load") or []
if isinstance(load, (list, tuple)) and len(load) >= 3:
load_1, load_5, load_15 = float(load[0]), float(load[1]), float(load[2])
else:
load_1 = float(data["load_1"]) if data.get("load_1") is not None else None
load_5 = float(data["load_5"]) if data.get("load_5") is not None else None
load_15 = float(data["load_15"]) if data.get("load_15") is not None else None
processes = sanitize_process_bundle(data) if "top_cpu" in data or "top_mem" in data else None
if processes is None:
legacy = sanitize_processes(data.get("top_processes"))
if legacy:
processes = {"top_cpu": legacy, "top_mem": []}
sample = WatchProbeSample(
probe_status=PROBE_OK,
source="ssh",
is_online=True,
cpu_pct=_int_pct(data.get("cpu_usage")),
mem_pct=_int_pct(data.get("mem_usage")),
disk_pct=_int_pct(data.get("disk_usage")),
disk_mount=str(data.get("disk_mount") or "/"),
load_1=load_1,
load_5=load_5,
load_15=load_15,
processes_json=processes,
**_parse_hardware_fields(data),
)
fill_hardware_gaps(sample)
return sample
def sample_needs_load_supplement(sample: WatchProbeSample) -> bool:
"""True when 1/5/15 load averages are missing from Agent-only path."""
if sample.probe_status != PROBE_OK:
return False
return sample.load_1 is None or sample.load_5 is None or sample.load_15 is None
def apply_load_avg_to_sample(sample: WatchProbeSample, load_avg: Any) -> WatchProbeSample:
"""Fill load_1/5/15 on an existing sample (mutates in place)."""
if not isinstance(load_avg, (list, tuple)) or len(load_avg) < 3:
return sample
if sample.load_1 is None:
sample.load_1 = float(load_avg[0])
if sample.load_5 is None:
sample.load_5 = float(load_avg[1])
if sample.load_15 is None:
sample.load_15 = float(load_avg[2])
return sample
def parse_redis_watch_payload(heartbeat: dict[str, str]) -> WatchProbeSample | None:
if heartbeat.get("is_online") != "True":
return None
raw = heartbeat.get("system_info") or "{}"
try:
info = json.loads(raw) if isinstance(raw, str) else (raw or {})
except json.JSONDecodeError:
return None
if not isinstance(info, dict):
return None
cpu = _metric_from_system_info(info, "cpu_usage", "cpu")
mem = _metric_from_system_info(info, "mem_usage", "mem")
disk = _metric_from_system_info(info, "disk_usage", "disk")
load = info.get("load_avg") or []
load_1 = load_5 = load_15 = None
if isinstance(load, (list, tuple)) and len(load) >= 3:
load_1, load_5, load_15 = float(load[0]), float(load[1]), float(load[2])
sample = WatchProbeSample(
probe_status=PROBE_OK,
source="redis",
is_online=True,
cpu_pct=round(cpu) if cpu is not None else None,
mem_pct=round(mem) if mem is not None else None,
disk_pct=round(disk) if disk is not None else None,
load_1=load_1,
load_5=load_5,
load_15=load_15,
**_parse_hardware_fields(info),
)
if cpu is None and mem is None:
return None
fill_hardware_gaps(sample)
return sample
def merge_redis_and_ssh(redis_sample: WatchProbeSample | None, ssh_sample: WatchProbeSample) -> WatchProbeSample:
if redis_sample is None:
return ssh_sample
if ssh_sample.probe_status != PROBE_OK:
return redis_sample
hw = {field: _pick_hardware_field(ssh_sample, redis_sample, field) for field in WatchProbeSample._LIVE_HARDWARE_KEYS}
def _pick_ssh_first(ssh_val, redis_val):
return ssh_val if ssh_val is not None else redis_val
out = WatchProbeSample(
probe_status=PROBE_OK,
source="mixed",
is_online=True,
duration_ms=ssh_sample.duration_ms,
# 5s Agent watch 优先;SSH 仅兜底或无 watch_mode 时
cpu_pct=_pick_ssh_first(ssh_sample.cpu_pct, redis_sample.cpu_pct),
mem_pct=_pick_ssh_first(ssh_sample.mem_pct, redis_sample.mem_pct),
disk_pct=_pick_ssh_first(ssh_sample.disk_pct, redis_sample.disk_pct),
disk_mount=str(hw.get("disk_mount") or ssh_sample.disk_mount or redis_sample.disk_mount or "/"),
load_1=_pick_ssh_first(ssh_sample.load_1, redis_sample.load_1),
load_5=_pick_ssh_first(ssh_sample.load_5, redis_sample.load_5),
load_15=_pick_ssh_first(ssh_sample.load_15, redis_sample.load_15),
cpu_cores=hw.get("cpu_cores"),
cpu_cores_physical=hw.get("cpu_cores_physical"),
cpu_model=hw.get("cpu_model"),
memory_total_gb=hw.get("memory_total_gb"),
memory_used_gb=hw.get("memory_used_gb"),
memory_available_gb=hw.get("memory_available_gb"),
disk_total_gb=hw.get("disk_total_gb"),
disk_used_gb=hw.get("disk_used_gb"),
disk_free_gb=hw.get("disk_free_gb"),
swap_total_gb=hw.get("swap_total_gb"),
swap_used_gb=hw.get("swap_used_gb"),
swap_pct=hw.get("swap_pct"),
os_release=hw.get("os_release"),
uptime_seconds=hw.get("uptime_seconds"),
process_running=hw.get("process_running"),
process_total=hw.get("process_total"),
per_cpu_pct=hw.get("per_cpu_pct"),
mem_free_mb=hw.get("mem_free_mb"),
mem_shared_mb=hw.get("mem_shared_mb"),
mem_buffers_mb=hw.get("mem_buffers_mb"),
mem_cached_mb=hw.get("mem_cached_mb"),
processes_json=ssh_sample.processes_json or redis_sample.processes_json,
)
fill_hardware_gaps(out)
return out
def failure_sample(
*,
status: str,
source: str,
error: str,
duration_ms: int = 0,
is_online: bool = False,
) -> WatchProbeSample:
return WatchProbeSample(
probe_status=status,
source=source,
is_online=is_online,
duration_ms=duration_ms,
error=error[:255],
)