atlasbot: surface node taints

atlasbot/snapshot/builder.py

@@ -75,6 +75,7 @@ def build_summary(snapshot: dict[str, Any] | None) -> dict[str, Any]:
     summary.update(_build_pressure(snapshot))
     summary.update(_build_hardware(nodes_detail))
     summary.update(_build_node_ages(nodes_detail))
+    summary.update(_build_node_taints(nodes_detail))
     summary.update(_build_capacity(metrics))
     summary.update(_build_pods(metrics))
     summary.update(_build_namespace_pods(snapshot))
@@ -149,6 +150,28 @@ def _build_node_ages(nodes_detail: list[dict[str, Any]]) -> dict[str, Any]:
     return {"node_ages": ages[:5]} if ages else {}
 
 
+def _build_node_taints(nodes_detail: list[dict[str, Any]]) -> dict[str, Any]:
+    taints: dict[str, list[str]] = {}
+    for node in nodes_detail or []:
+        if not isinstance(node, dict):
+            continue
+        name = node.get("name")
+        if not name:
+            continue
+        entries = node.get("taints") if isinstance(node.get("taints"), list) else []
+        for entry in entries:
+            if not isinstance(entry, dict):
+                continue
+            key = entry.get("key")
+            effect = entry.get("effect")
+            if isinstance(key, str) and isinstance(effect, str):
+                label = f"{key}:{effect}"
+                taints.setdefault(label, []).append(name)
+    if not taints:
+        return {}
+    return {"node_taints": {key: sorted(names) for key, names in taints.items()}}
+
+
 def _build_pods(metrics: dict[str, Any]) -> dict[str, Any]:
     pods = {
         "running": metrics.get("pods_running"),
@@ -385,6 +408,20 @@ def _append_node_ages(lines: list[str], summary: dict[str, Any]) -> None:
         lines.append("node_age_top: " + "; ".join(parts))
 
 
+def _append_node_taints(lines: list[str], summary: dict[str, Any]) -> None:
+    taints = summary.get("node_taints") if isinstance(summary.get("node_taints"), dict) else {}
+    if not taints:
+        return
+    parts = []
+    for key, names in taints.items():
+        if not isinstance(names, list):
+            continue
+        name_list = _format_names([str(name) for name in names if name])
+        parts.append(f"{key}={len(names)} ({name_list})" if name_list else f"{key}={len(names)}")
+    if parts:
+        lines.append("node_taints: " + "; ".join(sorted(parts)))
+
+
 def _append_pressure(lines: list[str], summary: dict[str, Any]) -> None:
     pressure = summary.get("pressure_nodes")
     if not isinstance(pressure, dict) or not pressure:
@@ -824,6 +861,7 @@ def summary_text(snapshot: dict[str, Any] | None) -> str:
     _append_pressure(lines, summary)
     _append_hardware(lines, summary)
     _append_node_ages(lines, summary)
+    _append_node_taints(lines, summary)
     _append_capacity(lines, summary)
     _append_pods(lines, summary)
     _append_namespace_pods(lines, summary)