76 lines
2.2 KiB
Python
76 lines
2.2 KiB
Python
|
"""Fleet-wide metric aggregation from Agent heartbeat payloads."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
from dataclasses import dataclass
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
|
||
|
|
def _safe_pct(value: Any) -> float | None:
|
||
|
|
if value is None:
|
||
|
|
return None
|
||
|
|
try:
|
||
|
|
num = float(value)
|
||
|
|
except (TypeError, ValueError):
|
||
|
|
return None
|
||
|
|
if num < 0 or num > 100:
|
||
|
|
return None
|
||
|
|
return num
|
||
|
|
|
||
|
|
|
||
|
|
def _metric_from_system_info(system_info: dict[str, Any], primary: str, fallback: str) -> float | None:
|
||
|
|
return _safe_pct(system_info.get(primary) if system_info.get(primary) is not None else system_info.get(fallback))
|
||
|
|
|
||
|
|
|
||
|
|
@dataclass(frozen=True)
|
||
|
|
class FleetMetricAggregate:
|
||
|
|
online_count: int
|
||
|
|
sample_count: int
|
||
|
|
cpu_avg: int | None
|
||
|
|
mem_avg: int | None
|
||
|
|
disk_avg: int | None
|
||
|
|
|
||
|
|
|
||
|
|
def aggregate_fleet_metrics(heartbeats: list[dict[str, Any]]) -> FleetMetricAggregate:
|
||
|
|
"""Aggregate Redis heartbeat hash rows (is_online + system_info JSON string)."""
|
||
|
|
online_count = 0
|
||
|
|
cpu_vals: list[float] = []
|
||
|
|
mem_vals: list[float] = []
|
||
|
|
disk_vals: list[float] = []
|
||
|
|
|
||
|
|
for row in heartbeats:
|
||
|
|
if row.get("is_online") in (True, "True"):
|
||
|
|
online_count += 1
|
||
|
|
raw = row.get("system_info") or "{}"
|
||
|
|
try:
|
||
|
|
info = json.loads(raw) if isinstance(raw, str) else (raw or {})
|
||
|
|
except json.JSONDecodeError:
|
||
|
|
info = {}
|
||
|
|
if not isinstance(info, dict):
|
||
|
|
continue
|
||
|
|
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")
|
||
|
|
if cpu is not None:
|
||
|
|
cpu_vals.append(cpu)
|
||
|
|
if mem is not None:
|
||
|
|
mem_vals.append(mem)
|
||
|
|
if disk is not None:
|
||
|
|
disk_vals.append(disk)
|
||
|
|
|
||
|
|
sample_count = max(len(cpu_vals), len(mem_vals), len(disk_vals))
|
||
|
|
|
||
|
|
def _avg(vals: list[float]) -> int | None:
|
||
|
|
if not vals:
|
||
|
|
return None
|
||
|
|
return round(sum(vals) / len(vals))
|
||
|
|
|
||
|
|
return FleetMetricAggregate(
|
||
|
|
online_count=online_count,
|
||
|
|
sample_count=sample_count,
|
||
|
|
cpu_avg=_avg(cpu_vals),
|
||
|
|
mem_avg=_avg(mem_vals),
|
||
|
|
disk_avg=_avg(disk_vals),
|
||
|
|
)
|