atlasbot: include node taints in snapshot

ariadne/services/cluster_state.py

@@ -147,6 +147,7 @@ def _node_details(payload: dict[str, Any]) -> list[dict[str, Any]]:
     details: list[dict[str, Any]] = []
     for node in _items(payload):
         metadata = node.get("metadata") if isinstance(node.get("metadata"), dict) else {}
+        spec = node.get("spec") if isinstance(node.get("spec"), dict) else {}
         status = node.get("status") if isinstance(node.get("status"), dict) else {}
         node_info = status.get("nodeInfo") if isinstance(status.get("nodeInfo"), dict) else {}
         labels = metadata.get("labels") if isinstance(metadata.get("labels"), dict) else {}
@@ -156,6 +157,7 @@ def _node_details(payload: dict[str, Any]) -> list[dict[str, Any]]:
         roles = _node_roles(labels)
         conditions = _node_pressure_conditions(status.get("conditions"))
         created_at = metadata.get("creationTimestamp") if isinstance(metadata.get("creationTimestamp"), str) else ""
+        taints = _node_taints(spec.get("taints"))
         details.append(
             {
                 "name": name,
@@ -172,6 +174,8 @@ def _node_details(payload: dict[str, Any]) -> list[dict[str, Any]]:
                 "addresses": _node_addresses(status),
                 "created_at": created_at,
                 "age_hours": _age_hours(created_at),
+                "taints": taints,
+                "unschedulable": bool(spec.get("unschedulable")),
                 "capacity": _node_capacity(status.get("capacity")),
                 "allocatable": _node_capacity(status.get("allocatable")),
                 "pressure": conditions,
@@ -191,6 +195,27 @@ def _age_hours(timestamp: str) -> float | None:
     return round((datetime.now(timezone.utc) - parsed).total_seconds() / 3600, 1)
 
 
+def _node_taints(raw: Any) -> list[dict[str, str]]:
+    if not isinstance(raw, list):
+        return []
+    taints: list[dict[str, str]] = []
+    for entry in raw:
+        if not isinstance(entry, dict):
+            continue
+        key = entry.get("key")
+        effect = entry.get("effect")
+        value = entry.get("value")
+        if isinstance(key, str) and isinstance(effect, str):
+            taints.append(
+                {
+                    "key": key,
+                    "value": value if isinstance(value, str) else "",
+                    "effect": effect,
+                }
+            )
+    return taints
+
+
 def _summarize_inventory(details: list[dict[str, Any]]) -> dict[str, Any]:
     summary = {
         "total": 0,

tests/test_cluster_state.py

@@ -36,6 +36,11 @@ def test_collect_cluster_state(monkeypatch) -> None:
                             "labels": {"kubernetes.io/arch": "amd64"},
                             "creationTimestamp": "2026-01-01T00:00:00Z",
                         },
+                        "spec": {
+                            "taints": [
+                                {"key": "node-role.kubernetes.io/control-plane", "effect": "NoSchedule"}
+                            ]
+                        },
                         "status": {
                             "conditions": [{"type": "Ready", "status": "False"}],
                             "nodeInfo": {"architecture": "amd64"},
@@ -116,6 +121,7 @@ def test_collect_cluster_state(monkeypatch) -> None:
     assert "pressure_nodes" in snapshot["nodes_summary"]
     assert snapshot["nodes_detail"]
     assert snapshot["nodes_detail"][1]["age_hours"] is not None
+    assert snapshot["nodes_detail"][1]["taints"]
     assert snapshot["workloads"]
     assert snapshot["namespace_pods"]
     assert snapshot["namespace_pods"][0]["namespace"] == "media"