diff --git a/atlasbot/snapshot/builder.py b/atlasbot/snapshot/builder.py index 60809b0..7819d54 100644 --- a/atlasbot/snapshot/builder.py +++ b/atlasbot/snapshot/builder.py @@ -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 = (