bstein/titan-iac
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

atlasbot: generalize inventory answers

Browse Source
This commit is contained in:
Brad Stein 2026-01-26 19:34:19 -03:00
parent 2d09e7f965
commit 16d0a22163
1 changed files with 50 additions and 30 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

80
services/comms/scripts/atlasbot/bot.py
View File

@ -382,11 +382,18 @@ def _inventory_sets(inventory: list[dict[str, Any]]) -> dict[str, Any]:
ready = [node["name"] for node in inventory if node.get("ready") is True]
not_ready = [node["name"] for node in inventory if node.get("ready") is False]
groups = _group_nodes(inventory)
workers = [node for node in inventory if "worker" in (node.get("roles") or [])]
worker_names = [node["name"] for node in workers]
worker_ready = [node["name"] for node in workers if node.get("ready") is True]
worker_not_ready = [node["name"] for node in workers if node.get("ready") is False]
return {
"names": sorted(names),
"ready": sorted(ready),
"not_ready": sorted(not_ready),
"groups": groups,
"worker_names": sorted(worker_names),
"worker_ready": sorted(worker_ready),
"worker_not_ready": sorted(worker_not_ready),
}
def structured_answer(prompt: str, *, inventory: list[dict[str, Any]], metrics_summary: str) -> str:
@ -402,6 +409,9 @@ def structured_answer(prompt: str, *, inventory: list[dict[str, Any]], metrics_s
ready = sets["ready"]
not_ready = sets["not_ready"]
groups = sets["groups"]
worker_names = sets["worker_names"]
worker_ready = sets["worker_ready"]
worker_not_ready = sets["worker_not_ready"]
total = len(names)
for node in _extract_titan_nodes(q):
@ -410,31 +420,12 @@ def structured_answer(prompt: str, *, inventory: list[dict[str, Any]], metrics_s
return f"Yes. {node} is in the Atlas cluster."
return f"No. {node} is not in the Atlas cluster."
if any(word in q for word in ("how many", "count", "number")) and "node" in q and "worker" not in q:
return f"Atlas has {total} nodes; {len(ready)} ready, {len(not_ready)} not ready."
if "node names" in q or ("nodes" in q and "named" in q) or "naming" in q:
return "Atlas node names: " + ", ".join(names) + "."
if "ready" in q and "node" in q and "worker" in q:
if "not ready" in q or "unready" in q or "down" in q:
return "Worker nodes not ready: " + (", ".join(not_ready) if not_ready else "none") + "."
return "Ready worker nodes ({}): {}.".format(len(ready), ", ".join(ready))
if "worker" in q and any(word in q for word in ("missing", "expected", "should")):
expected_workers = expected_worker_nodes_from_metrics()
missing = sorted(set(expected_workers) - set(ready + not_ready)) if expected_workers else []
if "missing" in q and missing:
return "Missing worker nodes: " + ", ".join(missing) + "."
if expected_workers:
msg = f"Grafana inventory expects {len(expected_workers)} workers."
if missing:
msg += f" Missing: {', '.join(missing)}."
return msg
return "No expected worker inventory found; using live cluster state."
if "worker" in q and "node" in q and "ready" not in q and "missing" not in q:
return f"Worker nodes: {len(ready)} ready, {len(not_ready)} not ready."
if "non-raspberry" in q or "non raspberry" in q or "not raspberry" in q:
non_rpi = sorted(set(groups.get("jetson", [])) | set(groups.get("amd64", [])))
if "besides" in q:
amd = groups.get("amd64", [])
return f"Non‑Raspberry Pi nodes (excluding Jetson): {', '.join(amd)}." if amd else "No non‑Raspberry Pi nodes outside Jetson."
return f"Non‑Raspberry Pi nodes: {', '.join(non_rpi)}." if non_rpi else "No non‑Raspberry Pi nodes found."
if "jetson" in q:
jets = groups.get("jetson", [])
@ -446,24 +437,53 @@ def structured_answer(prompt: str, *, inventory: list[dict[str, Any]], metrics_s
if "rpi4" in q:
rpi4 = groups.get("rpi4", [])
if any(word in q for word in ("how many", "count", "number")):
return f"Atlas has {len(rpi4)} rpi4 nodes."
return f"rpi4 nodes: {', '.join(rpi4)}." if rpi4 else "No rpi4 nodes found."
if "rpi5" in q:
rpi5 = groups.get("rpi5", [])
if any(word in q for word in ("how many", "count", "number")):
return f"Atlas has {len(rpi5)} rpi5 nodes."
return f"rpi5 nodes: {', '.join(rpi5)}." if rpi5 else "No rpi5 nodes found."
if "raspberry" in q or "rpi" in q:
rpi = sorted(set(groups.get("rpi4", [])) | set(groups.get("rpi5", [])))
if any(word in q for word in ("how many", "count", "number")):
return f"Atlas has {len(rpi)} Raspberry Pi nodes."
return f"Raspberry Pi nodes: {', '.join(rpi)}." if rpi else "No Raspberry Pi nodes found."
if "non-raspberry" in q or "non raspberry" in q or "not raspberry" in q:
non_rpi = sorted(set(groups.get("jetson", [])) | set(groups.get("amd64", [])))
return f"Non‑Raspberry Pi nodes: {', '.join(non_rpi)}." if non_rpi else "No non‑Raspberry Pi nodes found."
if "arm64-unknown" in q or "unknown" in q:
if "arm64-unknown" in q or "unknown" in q or "no hardware" in q:
unknown = sorted(set(groups.get("arm64-unknown", [])) | set(groups.get("unknown", [])))
return f"Unknown hardware nodes: {', '.join(unknown)}." if unknown else "No unknown hardware labels."
if "worker" in q and "node" in q:
if any(word in q for word in ("missing", "expected", "should")):
expected_workers = expected_worker_nodes_from_metrics()
missing = sorted(set(expected_workers) - set(worker_ready + worker_not_ready)) if expected_workers else []
if "missing" in q and missing:
return "Missing worker nodes: " + ", ".join(missing) + "."
if expected_workers:
msg = f"Grafana inventory expects {len(expected_workers)} workers."
if missing:
msg += f" Missing: {', '.join(missing)}."
return msg
return "No expected worker inventory found; using live cluster state."
if "not ready" in q or "unready" in q or "down" in q:
return "Worker nodes not ready: " + (", ".join(worker_not_ready) if worker_not_ready else "none") + "."
if any(word in q for word in ("how many", "count", "number")):
return f"Worker nodes: {len(worker_ready)} ready, {len(worker_not_ready)} not ready."
return "Ready worker nodes ({}): {}.".format(len(worker_ready), ", ".join(worker_ready))
if any(word in q for word in ("how many", "count", "number")) and "node" in q:
return f"Atlas has {total} nodes; {len(ready)} ready, {len(not_ready)} not ready."
if "node names" in q or ("nodes" in q and "named" in q) or "naming" in q:
return "Atlas node names: " + ", ".join(names) + "."
if "ready" in q and "node" in q:
return f"Ready nodes ({len(ready)}): {', '.join(ready)}."
return ""
def _metric_tokens(entry: dict[str, Any]) -> str: