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

fix: use summary data for workload/pod/pvc overrides

Browse Source
This commit is contained in:
Brad Stein 2026-02-02 17:07:26 -03:00
parent ea11460c92
commit 76eac2d6a4
1 changed files with 65 additions and 49 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

114
atlasbot/engine/answerer.py
View File

@ -690,80 +690,96 @@ class AnswerEngine:
if pods_line: if pods_line:
reply = f"From the latest snapshot: {pods_line}." reply = f"From the latest snapshot: {pods_line}."
if "load_index" in lowered_q or "load index" in lowered_q: if "load_index" in lowered_q or "load index" in lowered_q:
top = None
if isinstance(snapshot_used, dict): if isinstance(snapshot_used, dict):
summary = snapshot_used.get("summary") if isinstance(snapshot_used.get("summary"), dict) else {} summary = snapshot_used.get("summary") if isinstance(snapshot_used.get("summary"), dict) else {}
if summary: node_load = summary.get("node_load") if isinstance(summary.get("node_load"), list) else []
top_list = summary.get("top", {}).get("node_load") if isinstance(summary.get("top"), dict) else None best = None
if isinstance(top_list, list) and top_list: for item in node_load:
top = top_list[0] if not isinstance(item, dict):
if isinstance(top, dict): continue
node = top.get("node") node = item.get("node")
load = top.get("load_index") load = item.get("load_index")
if node and load is not None: try:
reply = f"From the latest snapshot: hottest_load_index_node: {node} load_index={load}." numeric = float(load)
except (TypeError, ValueError):
continue
if best is None or numeric > best["load_index"]:
best = {"node": node, "load_index": numeric}
if best and best.get("node") is not None:
reply = (
"From the latest snapshot: hottest_load_index_node: "
f"{best['node']} load_index={best['load_index']}."
)
if "workload" in lowered_q and any(tok in lowered_q for tok in ("not ready", "not-ready", "unready")): if "workload" in lowered_q and any(tok in lowered_q for tok in ("not ready", "not-ready", "unready")):
top = None
if isinstance(snapshot_used, dict): if isinstance(snapshot_used, dict):
summary = snapshot_used.get("summary") if isinstance(snapshot_used.get("summary"), dict) else {} summary = snapshot_used.get("summary") if isinstance(snapshot_used.get("summary"), dict) else {}
if summary: health = summary.get("workloads_health") if isinstance(summary.get("workloads_health"), dict) else {}
top = summary.get("top", {}).get("workload_not_ready") if isinstance(summary.get("top"), dict) else None items: list[dict[str, Any]] = []
if isinstance(top, list): for key in ("deployments", "statefulsets", "daemonsets"):
if top: entry = health.get(key) if isinstance(health.get(key), dict) else {}
entries = [] for item in entry.get("items") or []:
for item in top[:3]: if not isinstance(item, dict):
if isinstance(item, dict): continue
entries.append( items.append(
f"{item.get('namespace','?')}/{item.get('name','?')} {item.get('kind','?')} ready={item.get('ready')} desired={item.get('desired')}" {
) "kind": key[:-1],
"namespace": item.get("namespace") or "",
"name": item.get("name") or "",
"desired": item.get("desired"),
"ready": item.get("ready"),
}
)
if items:
items = sorted(items, key=lambda item: (item.get("namespace") or "", item.get("name") or ""))
entries = [
f"{item.get('namespace','?')}/{item.get('name','?')} {item.get('kind','?')} ready={item.get('ready')} desired={item.get('desired')}"
for item in items[:3]
]
if entries: if entries:
reply = f"From the latest snapshot: workloads_not_ready: {', '.join(entries)}." reply = f"From the latest snapshot: workloads_not_ready: {', '.join(entries)}."
else: else:
reply = "From the latest snapshot: workloads_not_ready: none." reply = "From the latest snapshot: workloads_not_ready: none."
if "pod" in lowered_q and ("waiting" in lowered_q or "wait" in lowered_q) and "reason" in lowered_q: if "pod" in lowered_q and ("waiting" in lowered_q or "wait" in lowered_q) and "reason" in lowered_q:
waiting = None
if isinstance(snapshot_used, dict): if isinstance(snapshot_used, dict):
summary = snapshot_used.get("summary") if isinstance(snapshot_used.get("summary"), dict) else {} summary = snapshot_used.get("summary") if isinstance(snapshot_used.get("summary"), dict) else {}
if summary: pod_issues = summary.get("pod_issues") if isinstance(summary.get("pod_issues"), dict) else {}
waiting = summary.get("pod_reason_totals", {}).get("waiting") if isinstance(summary.get("pod_reason_totals"), dict) else None waiting = pod_issues.get("waiting_reasons") if isinstance(pod_issues.get("waiting_reasons"), dict) else {}
if isinstance(waiting, dict): if waiting:
items = [] top = sorted(waiting.items(), key=lambda item: (-item[1], item[0]))[:3]
for reason, window in waiting.items(): items = [f"{reason}={count}" for reason, count in top]
if not isinstance(window, dict):
continue
# prefer 1h max if available, else 6h, else 24h
val = None
for key in ("1h", "6h", "24h"):
entry = window.get(key)
if isinstance(entry, dict):
val = entry.get("max")
if val is None:
val = entry.get("avg")
if val:
items.append(f"{reason}={val}")
break
if items:
reply = f"From the latest snapshot: pod_waiting_reasons: {'; '.join(items)}." reply = f"From the latest snapshot: pod_waiting_reasons: {'; '.join(items)}."
else: else:
reply = "From the latest snapshot: pod_waiting_reasons: none." reply = "From the latest snapshot: pod_waiting_reasons: none."
if "pvc" in lowered_q and any(tok in lowered_q for tok in ("usage", "used", "percent", "80")): if "pvc" in lowered_q and any(tok in lowered_q for tok in ("usage", "used", "percent", "80", "full")):
top = None
if isinstance(snapshot_used, dict): if isinstance(snapshot_used, dict):
summary = snapshot_used.get("summary") if isinstance(snapshot_used.get("summary"), dict) else {} summary = snapshot_used.get("summary") if isinstance(snapshot_used.get("summary"), dict) else {}
if summary: pvc_usage = summary.get("pvc_usage_top") if isinstance(summary.get("pvc_usage_top"), list) else []
top = summary.get("top", {}).get("pvc_usage") if isinstance(summary.get("top"), dict) else None
if isinstance(top, list):
over = [] over = []
for item in top: for item in pvc_usage:
if not isinstance(item, dict): if not isinstance(item, dict):
continue continue
used = item.get("used_percent") used = item.get("used_percent")
if used is not None and used >= 80: if used is None:
over.append(f"{item.get('namespace','?')}/{item.get('pvc','?')}={used:.2f}%") continue
try:
used_val = float(used)
except (TypeError, ValueError):
continue
if used_val >= 80:
name = item.get("name") or item.get("pvc")
if name:
over.append(f"{name}={used_val:.2f}%")
if over: if over:
reply = f"From the latest snapshot: pvc_usage>=80%: {', '.join(over)}." reply = f"From the latest snapshot: pvc_usage>=80%: {', '.join(over)}."
else: else:
reply = "From the latest snapshot: pvc_usage>=80%: none." reply = "From the latest snapshot: pvc_usage>=80%: none."
if "pressure" in lowered_q:
if isinstance(snapshot_used, dict):
summary = snapshot_used.get("summary") if isinstance(snapshot_used.get("summary"), dict) else {}
pressure_nodes = summary.get("pressure_nodes") if isinstance(summary.get("pressure_nodes"), dict) else {}
entries = [f"{key}={value}" for key, value in pressure_nodes.items() if value]
if entries:
reply = f"From the latest snapshot: pressure_nodes: {', '.join(entries)}."
if _has_token(lowered_q, "io") and ("node" in lowered_q or "nodes" in lowered_q): if _has_token(lowered_q, "io") and ("node" in lowered_q or "nodes" in lowered_q):
io_facts = _extract_hottest_facts(summary_lines, lowered_q) io_facts = _extract_hottest_facts(summary_lines, lowered_q)
io_line = next((fact for fact in io_facts if fact.startswith("hottest_io_node")), None) io_line = next((fact for fact in io_facts if fact.startswith("hottest_io_node")), None)