408 lines
14 KiB
Python
408 lines
14 KiB
Python
from __future__ import annotations
|
|
|
|
import re
|
|
from typing import Any
|
|
|
|
from atlasbot.engine.intent_router import IntentMatch
|
|
from atlasbot.snapshot.builder import summary_text
|
|
|
|
from ._base import *
|
|
|
|
|
|
def _join_context(parts: list[str]) -> str:
|
|
text = "\n".join([part for part in parts if part])
|
|
return text.strip()
|
|
|
|
|
|
def _format_metric_value(value: Any) -> str:
|
|
if isinstance(value, bool):
|
|
return str(value).lower()
|
|
if isinstance(value, int):
|
|
return str(value)
|
|
if isinstance(value, float):
|
|
return f"{value:.1f}".rstrip("0").rstrip(".")
|
|
return str(value)
|
|
|
|
|
|
def _format_history(history: list[dict[str, str]] | None) -> str:
|
|
if not history:
|
|
return ""
|
|
lines = ["Recent conversation (non-authoritative):"]
|
|
for entry in history[-4:]:
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
question = entry.get("q")
|
|
answer = entry.get("a")
|
|
role = entry.get("role")
|
|
content = entry.get("content")
|
|
if question:
|
|
lines.append(f"Q: {question}")
|
|
if answer:
|
|
lines.append(f"A: {answer}")
|
|
if role and content:
|
|
prefix = "Q" if role == "user" else "A"
|
|
lines.append(f"{prefix}: {content}")
|
|
return "\n".join(lines)
|
|
|
|
|
|
def _summary_lines(snapshot: dict[str, Any] | None) -> list[str]:
|
|
text = summary_text(snapshot)
|
|
if not text:
|
|
return []
|
|
return [line for line in text.splitlines() if line.strip()]
|
|
|
|
|
|
def _line_starting_with(lines: list[str], prefix: str) -> str | None:
|
|
if not lines:
|
|
return None
|
|
for line in lines:
|
|
if line.lower().startswith(prefix.lower()):
|
|
return line
|
|
return None
|
|
|
|
|
|
def _spine_lines(lines: list[str]) -> dict[str, str]:
|
|
spine: dict[str, str] = {}
|
|
_spine_nodes(lines, spine)
|
|
_spine_hardware(lines, spine)
|
|
_spine_hottest(lines, spine)
|
|
_spine_postgres(lines, spine)
|
|
_spine_namespaces(lines, spine)
|
|
_spine_pressure(lines, spine)
|
|
return spine
|
|
|
|
|
|
_NODES_PREFIX = "nodes:"
|
|
|
|
|
|
def _spine_nodes(lines: list[str], spine: dict[str, str]) -> None:
|
|
nodes_line = _line_starting_with(lines, _NODES_PREFIX)
|
|
if nodes_line:
|
|
spine["nodes_count"] = nodes_line
|
|
spine["nodes_ready"] = nodes_line
|
|
return
|
|
nodes_total = _line_starting_with(lines, "nodes_total:")
|
|
nodes_ready = _line_starting_with(lines, "nodes_ready:")
|
|
if nodes_total:
|
|
spine["nodes_count"] = nodes_total
|
|
if nodes_ready:
|
|
spine["nodes_ready"] = nodes_ready
|
|
|
|
|
|
def _spine_hardware(lines: list[str], spine: dict[str, str]) -> None:
|
|
hardware_line = _line_starting_with(lines, "hardware_nodes:")
|
|
if not hardware_line:
|
|
hardware_line = _line_starting_with(lines, "hardware:")
|
|
if hardware_line:
|
|
spine["nodes_non_rpi"] = hardware_line
|
|
|
|
|
|
def _spine_hottest(lines: list[str], spine: dict[str, str]) -> None:
|
|
hottest_line = _line_starting_with(lines, "hottest:")
|
|
if not hottest_line:
|
|
return
|
|
for key in ("hottest_cpu", "hottest_ram", "hottest_net", "hottest_io", "hottest_disk"):
|
|
spine[key] = hottest_line
|
|
|
|
|
|
def _spine_postgres(lines: list[str], spine: dict[str, str]) -> None:
|
|
postgres_total = _line_starting_with(lines, "postgres_connections_total:")
|
|
if postgres_total:
|
|
spine["postgres_connections"] = postgres_total
|
|
postgres_line = _line_starting_with(lines, "postgres:")
|
|
if postgres_line:
|
|
spine["postgres_hottest"] = postgres_line
|
|
|
|
|
|
def _spine_namespaces(lines: list[str], spine: dict[str, str]) -> None:
|
|
namespaces_top = _line_starting_with(lines, "namespaces_top:")
|
|
if namespaces_top:
|
|
spine["namespace_most_pods"] = namespaces_top
|
|
|
|
|
|
def _spine_pressure(lines: list[str], spine: dict[str, str]) -> None:
|
|
pressure_line = _line_starting_with(lines, "pressure_nodes:")
|
|
if pressure_line:
|
|
spine["pressure_summary"] = pressure_line
|
|
return
|
|
load_line = _line_starting_with(lines, "node_load_top:")
|
|
if load_line:
|
|
spine["pressure_summary"] = load_line
|
|
|
|
|
|
def _parse_group_line(line: str) -> dict[str, list[str]]:
|
|
groups: dict[str, list[str]] = {}
|
|
if not line:
|
|
return groups
|
|
payload = line.split(":", 1)[1] if ":" in line else line
|
|
for part in payload.split(";"):
|
|
part = part.strip()
|
|
if not part or "=" not in part:
|
|
continue
|
|
key, value = part.split("=", 1)
|
|
value = value.strip()
|
|
nodes: list[str] = []
|
|
if "(" in value and ")" in value:
|
|
inner = value[value.find("(") + 1 : value.rfind(")")]
|
|
nodes = [item.strip() for item in inner.split(",") if item.strip()]
|
|
if not nodes:
|
|
cleaned = re.sub(r"^[0-9]+", "", value).strip()
|
|
nodes = [item.strip() for item in cleaned.split(",") if item.strip()]
|
|
groups[key.strip()] = nodes
|
|
return groups
|
|
|
|
|
|
def _parse_hottest(line: str, metric: str) -> str | None:
|
|
if not line:
|
|
return None
|
|
payload = line.split(":", 1)[1] if ":" in line else line
|
|
for part in payload.split(";"):
|
|
part = part.strip()
|
|
if part.startswith(f"{metric}="):
|
|
return part
|
|
return None
|
|
|
|
|
|
def _spine_answer(intent: IntentMatch, spine_line: str | None) -> str | None:
|
|
if not spine_line:
|
|
return None
|
|
handlers = {
|
|
"nodes_count": _spine_nodes_answer,
|
|
"nodes_ready": _spine_nodes_answer,
|
|
"nodes_non_rpi": _spine_non_rpi_answer,
|
|
"hardware_mix": _spine_hardware_answer,
|
|
"postgres_connections": _spine_postgres_answer,
|
|
"postgres_hottest": _spine_postgres_answer,
|
|
"namespace_most_pods": _spine_namespace_answer,
|
|
"pressure_summary": _spine_pressure_answer,
|
|
}
|
|
kind = intent.kind
|
|
if kind.startswith("hottest_"):
|
|
return _spine_hottest_answer(kind, spine_line)
|
|
handler = handlers.get(kind)
|
|
if handler:
|
|
return handler(spine_line)
|
|
return spine_line
|
|
|
|
|
|
def _spine_nodes_answer(line: str) -> str:
|
|
return line
|
|
|
|
|
|
def _spine_non_rpi_answer(line: str) -> str:
|
|
groups = _parse_group_line(line)
|
|
non_rpi: list[str] = []
|
|
for key, nodes in groups.items():
|
|
if key.lower().startswith("rpi"):
|
|
continue
|
|
non_rpi.extend(nodes)
|
|
if non_rpi:
|
|
return "Non-Raspberry Pi nodes: " + ", ".join(non_rpi) + "."
|
|
return line
|
|
|
|
|
|
def _spine_hardware_answer(line: str) -> str:
|
|
return line
|
|
|
|
|
|
def _spine_hottest_answer(kind: str, line: str) -> str:
|
|
metric = kind.split("_", 1)[1]
|
|
hottest = _parse_hottest(line, metric)
|
|
if hottest:
|
|
return hottest
|
|
return line
|
|
|
|
|
|
def _spine_postgres_answer(line: str) -> str:
|
|
return line
|
|
|
|
|
|
def _spine_namespace_answer(line: str) -> str:
|
|
payload = line.split(":", 1)[1] if ":" in line else line
|
|
top = payload.split(";")[0].strip()
|
|
if top:
|
|
return f"Namespace with most pods: {top}."
|
|
return line
|
|
|
|
|
|
def _spine_pressure_answer(line: str) -> str:
|
|
return line
|
|
|
|
|
|
def _spine_from_summary(summary: dict[str, Any]) -> dict[str, str]:
|
|
if not isinstance(summary, dict) or not summary:
|
|
return {}
|
|
spine: dict[str, str] = {}
|
|
spine.update(_spine_from_counts(summary))
|
|
spine.update(_spine_from_hardware(summary))
|
|
spine.update(_spine_from_hottest(summary))
|
|
spine.update(_spine_from_postgres(summary))
|
|
spine.update(_spine_from_namespace_pods(summary))
|
|
spine.update(_spine_from_pressure(summary))
|
|
return spine
|
|
|
|
|
|
def _spine_from_counts(summary: dict[str, Any]) -> dict[str, str]:
|
|
counts = summary.get("counts") if isinstance(summary.get("counts"), dict) else {}
|
|
inventory = summary.get("inventory") if isinstance(summary.get("inventory"), dict) else {}
|
|
nodes = summary.get("nodes") if isinstance(summary.get("nodes"), dict) else {}
|
|
workers = inventory.get("workers") if isinstance(inventory.get("workers"), dict) else {}
|
|
total = nodes.get("total")
|
|
ready = nodes.get("ready")
|
|
not_ready = nodes.get("not_ready")
|
|
if total is None:
|
|
total = counts.get("nodes_total")
|
|
if ready is None:
|
|
ready = counts.get("nodes_ready")
|
|
if not_ready is None and isinstance(inventory.get("not_ready_names"), list):
|
|
not_ready = len(inventory.get("not_ready_names") or [])
|
|
workers_ready = workers.get("ready")
|
|
workers_total = workers.get("total")
|
|
if total is None and ready is None and not_ready is None:
|
|
return {}
|
|
parts = []
|
|
if total is not None:
|
|
parts.append(f"total={int(total)}")
|
|
if ready is not None:
|
|
parts.append(f"ready={int(ready)}")
|
|
if not_ready is not None:
|
|
parts.append(f"not_ready={int(not_ready)}")
|
|
if workers_total is not None and workers_ready is not None:
|
|
parts.append(f"workers_ready={int(workers_ready)}/{int(workers_total)}")
|
|
line = "nodes: " + ", ".join(parts)
|
|
return {"nodes_count": line, "nodes_ready": line}
|
|
|
|
|
|
def _spine_from_hardware(summary: dict[str, Any]) -> dict[str, str]:
|
|
hardware = summary.get("hardware") if isinstance(summary.get("hardware"), dict) else {}
|
|
if not hardware:
|
|
return {}
|
|
parts = []
|
|
for key, nodes in hardware.items():
|
|
if not isinstance(nodes, list):
|
|
continue
|
|
node_list = ", ".join(str(n) for n in nodes if n)
|
|
if node_list:
|
|
parts.append(f"{key}=({node_list})")
|
|
if not parts:
|
|
return {}
|
|
return {"nodes_non_rpi": "hardware: " + "; ".join(parts)}
|
|
|
|
|
|
def _spine_from_hottest(summary: dict[str, Any]) -> dict[str, str]:
|
|
hottest = summary.get("hottest") if isinstance(summary.get("hottest"), dict) else {}
|
|
top = summary.get("top") if isinstance(summary.get("top"), dict) else {}
|
|
top_hottest = top.get("node_hottest") if isinstance(top.get("node_hottest"), dict) else {}
|
|
if not hottest and top_hottest:
|
|
hottest = top_hottest
|
|
elif top_hottest:
|
|
for key, value in top_hottest.items():
|
|
if key not in hottest and value is not None:
|
|
hottest[key] = value
|
|
if not hottest:
|
|
return {}
|
|
mapping = {}
|
|
for key in ("cpu", "ram", "net", "io", "disk"):
|
|
entry = hottest.get(key)
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
node = entry.get("node") or entry.get("label") or ""
|
|
value = entry.get("value")
|
|
if node:
|
|
mapping[f"hottest_{key}"] = f"{key}={node} ({_format_metric_value(value)})"
|
|
if not mapping:
|
|
return {}
|
|
return mapping
|
|
|
|
|
|
def _spine_from_postgres(summary: dict[str, Any]) -> dict[str, str]:
|
|
postgres = summary.get("postgres") if isinstance(summary.get("postgres"), dict) else {}
|
|
if not postgres:
|
|
top = summary.get("top") if isinstance(summary.get("top"), dict) else {}
|
|
postgres = top.get("postgres") if isinstance(top.get("postgres"), dict) else {}
|
|
if not postgres:
|
|
return {}
|
|
used = postgres.get("used")
|
|
max_conn = postgres.get("max")
|
|
hottest = postgres.get("hottest_db") if isinstance(postgres.get("hottest_db"), dict) else {}
|
|
hottest_label = hottest.get("label") or ""
|
|
facts: dict[str, str] = {}
|
|
if used is not None and max_conn is not None:
|
|
facts["postgres_connections"] = f"postgres_connections_total: used={int(used)}, max={int(max_conn)}"
|
|
if hottest_label:
|
|
facts["postgres_hottest"] = f"postgres_hottest_db: {hottest_label}"
|
|
return facts
|
|
|
|
|
|
def _spine_from_namespace_pods(summary: dict[str, Any]) -> dict[str, str]:
|
|
pods = summary.get("namespace_pods") if isinstance(summary.get("namespace_pods"), list) else []
|
|
if not pods:
|
|
top = summary.get("top") if isinstance(summary.get("top"), dict) else {}
|
|
pods = top.get("namespace_pods") if isinstance(top.get("namespace_pods"), list) else []
|
|
if not pods:
|
|
return {}
|
|
best_name = ""
|
|
best_value = None
|
|
for entry in pods:
|
|
if not isinstance(entry, dict):
|
|
continue
|
|
name = entry.get("namespace") or entry.get("name") or entry.get("label") or ""
|
|
value = entry.get("pods")
|
|
if value is None:
|
|
value = entry.get("pods_total")
|
|
if value is None:
|
|
value = entry.get("value")
|
|
try:
|
|
numeric = float(value)
|
|
except (TypeError, ValueError):
|
|
numeric = None
|
|
if name and numeric is not None and (best_value is None or numeric > best_value):
|
|
best_name = name
|
|
best_value = numeric
|
|
if best_name:
|
|
return {"namespace_most_pods": f"namespace_most_pods: {best_name} ({int(best_value or 0)} pods)"}
|
|
return {}
|
|
|
|
|
|
def _spine_from_pressure(summary: dict[str, Any]) -> dict[str, str]:
|
|
pressure = summary.get("pressure_summary") if isinstance(summary.get("pressure_summary"), dict) else {}
|
|
if not pressure:
|
|
pressure = summary.get("pressure_nodes") if isinstance(summary.get("pressure_nodes"), dict) else {}
|
|
if not pressure:
|
|
return {}
|
|
total = pressure.get("total")
|
|
unsched = pressure.get("unschedulable")
|
|
names = pressure.get("names") if isinstance(pressure.get("names"), list) else []
|
|
parts = []
|
|
if total is None and names:
|
|
total = len([name for name in names if name])
|
|
if total is not None:
|
|
parts.append(f"total={int(total)}")
|
|
if unsched is not None:
|
|
parts.append(f"unschedulable={int(unsched)}")
|
|
if parts:
|
|
return {"pressure_summary": "pressure_nodes: " + ", ".join(parts)}
|
|
return {}
|
|
|
|
|
|
def _spine_fallback(intent: IntentMatch, lines: list[str]) -> str | None:
|
|
if not lines:
|
|
return None
|
|
keywords = {
|
|
"nodes_count": ("nodes:", "nodes_total:"),
|
|
"nodes_ready": ("nodes:", "nodes_ready:"),
|
|
"postgres_hottest": ("postgres_hottest", "hottest_db", "postgres"),
|
|
"namespace_most_pods": ("namespace", "pods", "namespaces_top"),
|
|
"pressure_summary": ("pressure", "node_load_top"),
|
|
}
|
|
for token in keywords.get(intent.kind, ("",)):
|
|
if not token:
|
|
continue
|
|
for line in lines:
|
|
if token in line:
|
|
return line
|
|
return None
|
|
|
|
|
|
__all__ = [name for name in globals() if name.startswith("_") and not name.startswith("__")]
|