69 lines
2.3 KiB
Python
69 lines
2.3 KiB
Python
|
"""Per-server metric parsing and offline segment merging."""
|
||
|
|
|
||
|
|
from __future__ import annotations
|
||
|
|
|
||
|
|
import json
|
||
|
|
from typing import Any
|
||
|
|
|
||
|
|
from server.utils.fleet_metrics import _metric_from_system_info
|
||
|
|
|
||
|
|
|
||
|
|
def parse_system_info_metrics(system_info: dict[str, Any] | str | None) -> tuple[int | None, int | None]:
|
||
|
|
"""Extract CPU/mem integer percentages from Agent system_info."""
|
||
|
|
if system_info is None:
|
||
|
|
return None, None
|
||
|
|
info: dict[str, Any]
|
||
|
|
if isinstance(system_info, str):
|
||
|
|
try:
|
||
|
|
parsed = json.loads(system_info)
|
||
|
|
except json.JSONDecodeError:
|
||
|
|
return None, None
|
||
|
|
info = parsed if isinstance(parsed, dict) else {}
|
||
|
|
elif isinstance(system_info, dict):
|
||
|
|
info = system_info
|
||
|
|
else:
|
||
|
|
return None, None
|
||
|
|
|
||
|
|
cpu = _metric_from_system_info(info, "cpu_usage", "cpu")
|
||
|
|
mem = _metric_from_system_info(info, "mem_usage", "mem")
|
||
|
|
cpu_i = round(cpu) if cpu is not None else None
|
||
|
|
mem_i = round(mem) if mem is not None else None
|
||
|
|
return cpu_i, mem_i
|
||
|
|
|
||
|
|
|
||
|
|
def sample_from_redis_heartbeat(data: dict[str, Any]) -> tuple[bool, int | None, int | None]:
|
||
|
|
"""Map Redis heartbeat hash to (is_online, cpu_pct, mem_pct) for one flush sample."""
|
||
|
|
is_online = data.get("is_online") == "True"
|
||
|
|
if not is_online:
|
||
|
|
return False, None, None
|
||
|
|
cpu, mem = parse_system_info_metrics(data.get("system_info"))
|
||
|
|
return True, cpu, mem
|
||
|
|
|
||
|
|
|
||
|
|
def merge_offline_segments(points: list[dict[str, Any]]) -> list[dict[str, str]]:
|
||
|
|
"""Merge consecutive is_online=false points into {from, to} segments (recorded_at strings)."""
|
||
|
|
if not points:
|
||
|
|
return []
|
||
|
|
|
||
|
|
segments: list[dict[str, str]] = []
|
||
|
|
seg_start: str | None = None
|
||
|
|
prev_at: str | None = None
|
||
|
|
|
||
|
|
for point in points:
|
||
|
|
recorded_at = str(point.get("recorded_at") or "")
|
||
|
|
if not recorded_at:
|
||
|
|
continue
|
||
|
|
is_online = bool(point.get("is_online"))
|
||
|
|
if not is_online:
|
||
|
|
if seg_start is None:
|
||
|
|
seg_start = recorded_at
|
||
|
|
elif seg_start is not None and prev_at is not None:
|
||
|
|
segments.append({"from": seg_start, "to": prev_at})
|
||
|
|
seg_start = None
|
||
|
|
prev_at = recorded_at
|
||
|
|
|
||
|
|
if seg_start is not None and prev_at is not None:
|
||
|
|
segments.append({"from": seg_start, "to": prev_at})
|
||
|
|
|
||
|
|
return segments
|