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

fix: correct answerer overrides for pods/pvc

Browse Source
This commit is contained in:
Brad Stein 2026-02-02 15:14:02 -03:00
parent d2144bd17f
commit 400be36093
1 changed files with 86 additions and 0 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

86
atlasbot/engine/answerer.py
View File

@ -671,6 +671,92 @@ class AnswerEngine:
reply = f"From the latest snapshot: {hw_line}; {hw_nodes_line}." reply = f"From the latest snapshot: {hw_line}; {hw_nodes_line}."
else: else:
reply = f"From the latest snapshot: {hw_line}." reply = f"From the latest snapshot: {hw_line}."
if "node" in lowered_q and any(tok in lowered_q for tok in ("how many", "count", "total")):
nodes_line = next((line for line in summary_lines if line.startswith("nodes:")), None)
if nodes_line:
reply = f"From the latest snapshot: {nodes_line}."
if "pod" in lowered_q and "node" in lowered_q and any(tok in lowered_q for tok in ("most", "highest", "max")):
pods_line = next(
(line for line in summary_lines if line.startswith("node_pods_max:") or line.startswith("node_pods_top:")),
None,
)
if pods_line:
reply = f"From the latest snapshot: {pods_line}."
if "load_index" in lowered_q or "load index" in lowered_q:
top = None
if isinstance(snapshot_used, dict):
summary = snapshot_used.get("summary") if isinstance(snapshot_used.get("summary"), dict) else {}
if summary:
top_list = summary.get("top", {}).get("node_load") if isinstance(summary.get("top"), dict) else None
if isinstance(top_list, list) and top_list:
top = top_list[0]
if isinstance(top, dict):
node = top.get("node")
load = top.get("load_index")
if node and load is not None:
reply = f"From the latest snapshot: hottest_load_index_node: {node} load_index={load}."
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):
summary = snapshot_used.get("summary") if isinstance(snapshot_used.get("summary"), dict) else {}
if summary:
top = summary.get("top", {}).get("workload_not_ready") if isinstance(summary.get("top"), dict) else None
if isinstance(top, list):
if top:
entries = []
for item in top[:3]:
if isinstance(item, dict):
entries.append(
f"{item.get('namespace','?')}/{item.get('name','?')} {item.get('kind','?')} ready={item.get('ready')} desired={item.get('desired')}"
)
if entries:
reply = f"From the latest snapshot: workloads_not_ready: {', '.join(entries)}."
else:
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:
waiting = None
if isinstance(snapshot_used, dict):
summary = snapshot_used.get("summary") if isinstance(snapshot_used.get("summary"), dict) else {}
if summary:
waiting = summary.get("pod_reason_totals", {}).get("waiting") if isinstance(summary.get("pod_reason_totals"), dict) else None
if isinstance(waiting, dict):
items = []
for reason, window in waiting.items():
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)}."
else:
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")):
top = None
if isinstance(snapshot_used, dict):
summary = snapshot_used.get("summary") if isinstance(snapshot_used.get("summary"), dict) else {}
if summary:
top = summary.get("top", {}).get("pvc_usage") if isinstance(summary.get("top"), dict) else None
if isinstance(top, list):
over = []
for item in top:
if not isinstance(item, dict):
continue
used = item.get("used_percent")
if used is not None and used >= 80:
over.append(f"{item.get('namespace','?')}/{item.get('pvc','?')}={used:.2f}%")
if over:
reply = f"From the latest snapshot: pvc_usage>=80%: {', '.join(over)}."
else:
reply = "From the latest snapshot: pvc_usage>=80%: none."
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)