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

snapshot: surface load and namespace summaries

Browse Source
This commit is contained in:
Brad Stein 2026-01-29 13:32:57 -03:00
parent 9f2ccf4f31
commit 2cd53bf36f
1 changed files with 18 additions and 2 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

20
atlasbot/snapshot/builder.py
View File

@ -94,9 +94,11 @@ def build_summary(snapshot: dict[str, Any] | None) -> dict[str, Any]:
summary.update(_build_hottest(metrics))
summary.update(_build_pvc(metrics))
summary.update(_build_namespace_capacity(metrics))
summary.update(_build_namespace_capacity_summary(metrics))
summary.update(_build_longhorn(snapshot))
summary.update(_build_root_disk_headroom(metrics))
summary.update(_build_node_load(metrics))
summary.update(_build_node_load_summary(metrics))
summary.update(_build_workloads(snapshot))
summary.update(_build_flux(snapshot))
return summary
@ -402,6 +404,20 @@ def _build_namespace_capacity(metrics: dict[str, Any]) -> dict[str, Any]:
return {"namespace_capacity": capacity}
def _build_namespace_capacity_summary(metrics: dict[str, Any]) -> dict[str, Any]:
summary = metrics.get("namespace_capacity_summary")
if not isinstance(summary, dict) or not summary:
return {}
return {"namespace_capacity_summary": summary}
def _build_node_load_summary(metrics: dict[str, Any]) -> dict[str, Any]:
summary = metrics.get("node_load_summary")
if not isinstance(summary, dict) or not summary:
return {}
return {"node_load_summary": summary}
def _build_workloads(snapshot: dict[str, Any]) -> dict[str, Any]:
workloads = snapshot.get("workloads") if isinstance(snapshot.get("workloads"), list) else []
return {"workloads": workloads}
@ -851,7 +867,7 @@ def _append_root_disk_headroom(lines: list[str], summary: dict[str, Any]) -> Non
lines.append("root_disk_low_headroom: " + "; ".join(parts))
def _append_longhorn(lines: list[str], summary: dict[str, Any]) -> None:
def _append_longhorn(lines: list[str], summary: dict[str, Any]) -> None: # noqa: C901
longhorn = summary.get("longhorn") if isinstance(summary.get("longhorn"), dict) else {}
if not longhorn:
return
@ -979,7 +995,7 @@ def _append_namespace_io_net(lines: list[str], summary: dict[str, Any]) -> None:
lines.append("namespace_io_top: " + "; ".join(parts))
def _append_pod_usage(lines: list[str], summary: dict[str, Any]) -> None:
def _append_pod_usage(lines: list[str], summary: dict[str, Any]) -> None: # noqa: C901, PLR0912
metrics = summary.get("metrics") if isinstance(summary.get("metrics"), dict) else {}
cpu_top = metrics.get("pod_cpu_top") if isinstance(metrics.get("pod_cpu_top"), list) else []
cpu_top_node = (