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

atlasbot: surface topology in snapshot summary

Browse Source
This commit is contained in:
Brad Stein 2026-01-30 17:06:29 -03:00
parent e3c68af679
commit 5f8b807cfa
1 changed files with 43 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

43
atlasbot/snapshot/builder.py
View File

@ -114,12 +114,15 @@ def _merge_cluster_summary(snapshot: dict[str, Any], summary: dict[str, Any]) ->
signals = cluster_summary.get("signals")
profiles = cluster_summary.get("profiles")
inventory = cluster_summary.get("inventory")
topology = cluster_summary.get("topology")
if isinstance(signals, list):
summary["signals"] = signals
if isinstance(profiles, dict):
summary["profiles"] = profiles
if isinstance(inventory, dict):
summary["inventory"] = inventory
if isinstance(topology, dict):
summary["topology"] = topology
def _nodes_detail(snapshot: dict[str, Any]) -> list[dict[str, Any]]:
@ -1323,6 +1326,45 @@ def _append_workloads(lines: list[str], summary: dict[str, Any]) -> None:
lines.append("workloads_top: " + "; ".join(parts))
def _append_topology(lines: list[str], summary: dict[str, Any]) -> None:
topology = summary.get("topology") if isinstance(summary.get("topology"), dict) else {}
if not topology:
return
nodes = topology.get("nodes") if isinstance(topology.get("nodes"), list) else []
workloads = topology.get("workloads") if isinstance(topology.get("workloads"), list) else []
if nodes:
parts = []
for entry in nodes[:5]:
if not isinstance(entry, dict):
continue
node = entry.get("node")
top = entry.get("workloads_top") if isinstance(entry.get("workloads_top"), list) else []
if not node or not top:
continue
items = ", ".join([f"{name}({count})" for name, count in top if name and count is not None])
if items:
parts.append(f"{node}={items}")
if parts:
lines.append("node_workloads_top: " + "; ".join(parts))
if workloads:
parts = []
for entry in workloads[:5]:
if not isinstance(entry, dict):
continue
namespace = entry.get("namespace")
name = entry.get("workload")
nodes_top = entry.get("nodes_top") if isinstance(entry.get("nodes_top"), list) else []
if not namespace or not name:
continue
nodes_label = ", ".join([f"{node}:{count}" for node, count in nodes_top if node])
label = f"{namespace}/{name}"
if nodes_label:
label = f"{label} [{nodes_label}]"
parts.append(label)
if parts:
lines.append("workload_nodes_top: " + "; ".join(parts))
def _append_flux(lines: list[str], summary: dict[str, Any]) -> None:
flux = summary.get("flux") if isinstance(summary.get("flux"), dict) else {}
if not flux:
@ -1656,6 +1698,7 @@ def summary_text(snapshot: dict[str, Any] | None) -> str:
_append_namespace_capacity_summary(lines, summary)
_append_longhorn(lines, summary)
_append_workloads(lines, summary)
_append_topology(lines, summary)
_append_workloads_by_namespace(lines, summary)
_append_node_load_summary(lines, summary)
_append_cluster_watchlist(lines, summary)