bstein/atlasbot
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

atlasbot: add hardware usage top summary

Browse Source
This commit is contained in:
Brad Stein 2026-02-01 17:55:07 -03:00
parent 67488a8898
commit 1ef8ae4d3d
1 changed files with 20 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
atlasbot/snapshot/builder.py
View File

@ -1580,6 +1580,7 @@ def _append_hardware_usage(lines: list[str], summary: dict[str, Any]) -> None:
if not isinstance(usage, list) or not usage: if not isinstance(usage, list) or not usage:
return return
parts = [] parts = []
tops: dict[str, tuple[str, float]] = {}
for entry in usage[:5]: for entry in usage[:5]:
if not isinstance(entry, dict): if not isinstance(entry, dict):
continue continue
@ -1595,8 +1596,27 @@ def _append_hardware_usage(lines: list[str], summary: dict[str, Any]) -> None:
label += f" cpu={_format_float(cpu)} ram={_format_float(ram)}" label += f" cpu={_format_float(cpu)} ram={_format_float(ram)}"
label += f" io={_format_rate_bytes(io)} net={_format_rate_bytes(net)}" label += f" io={_format_rate_bytes(io)} net={_format_rate_bytes(net)}"
parts.append(label) parts.append(label)
for metric, value in (("cpu", cpu), ("ram", ram), ("io", io), ("net", net), ("load", load)):
if isinstance(value, (int, float)):
current = tops.get(metric)
if current is None or float(value) > current[1]:
tops[metric] = (hardware, float(value))
if parts: if parts:
lines.append("hardware_usage_avg: " + "; ".join(parts)) lines.append("hardware_usage_avg: " + "; ".join(parts))
if tops:
top_parts = []
for metric in ("cpu", "ram", "io", "net", "load"):
entry = tops.get(metric)
if not entry:
continue
hardware, value = entry
if metric in {"io", "net"}:
rendered = _format_rate_bytes(value)
else:
rendered = _format_float(value)
top_parts.append(f"{metric}={hardware} ({rendered})")
if top_parts:
lines.append("hardware_usage_top: " + "; ".join(top_parts))
def _append_cluster_watchlist(lines: list[str], summary: dict[str, Any]) -> None: def _append_cluster_watchlist(lines: list[str], summary: dict[str, Any]) -> None: