570 lines
17 KiB
Python
Executable File
570 lines
17 KiB
Python
Executable File
#!/usr/bin/env python3
|
|
"""Plan and apply guarded self-healing actions from Jeannie status findings."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import argparse
|
|
import json
|
|
import os
|
|
import re
|
|
import subprocess
|
|
import sys
|
|
import time
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
REPO_ROOT = Path(__file__).resolve().parents[1]
|
|
DEFAULT_KUBECONFIG = Path(
|
|
os.environ.get(
|
|
"KUBECONFIG", os.environ.get("LAB_KUBECONFIG_PATH", "/home/jv/.kube/config")
|
|
)
|
|
)
|
|
STATE_DIR = Path(
|
|
os.environ.get(
|
|
"HOMELAB_STATE_DIR",
|
|
Path(os.environ.get("XDG_DATA_HOME", Path.home() / ".local/share")) / "homelab",
|
|
)
|
|
)
|
|
HEAL_STATE_FILE = Path(
|
|
os.environ.get("LAB_HEAL_STATE_FILE", STATE_DIR / "heal-state.json")
|
|
)
|
|
HEAL_COOLDOWN_SECONDS = int(os.environ.get("LAB_HEAL_COOLDOWN_SECONDS", "1800"))
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class HealRule:
|
|
check: str
|
|
command: str
|
|
risk: str
|
|
reason: str
|
|
auto: bool
|
|
priority: int
|
|
|
|
|
|
RULES = [
|
|
HealRule(
|
|
"Kubernetes API",
|
|
"./jeannie start-cluster",
|
|
"low",
|
|
"API is down; start-cluster starts kubelet/containerd and worker VMs without destroying state.",
|
|
True,
|
|
100,
|
|
),
|
|
HealRule(
|
|
"Pimox workers running",
|
|
"./jeannie start-cluster",
|
|
"low",
|
|
"Worker VMs are expected cluster capacity and start-cluster is idempotent.",
|
|
True,
|
|
95,
|
|
),
|
|
HealRule(
|
|
"Kubernetes nodes Ready",
|
|
"./jeannie start-cluster",
|
|
"low",
|
|
"NotReady worker nodes commonly recover by starting the saved cluster runtime.",
|
|
True,
|
|
90,
|
|
),
|
|
HealRule(
|
|
"Pi-hole DNS",
|
|
"./jeannie rpi-services",
|
|
"medium",
|
|
"Reapplies RPi DNS services; safe for the lab but can briefly disrupt DNS.",
|
|
True,
|
|
80,
|
|
),
|
|
HealRule(
|
|
"RPi Docker root state",
|
|
"./jeannie rpi-services",
|
|
"medium",
|
|
"May repair service runtime but should be reviewed if storage is failing.",
|
|
True,
|
|
78,
|
|
),
|
|
HealRule(
|
|
"Uptime Kuma HTTP",
|
|
"./jeannie rpi-services",
|
|
"medium",
|
|
"Reapplies RPi service Compose stack.",
|
|
True,
|
|
70,
|
|
),
|
|
HealRule(
|
|
"Gitea container",
|
|
"./jeannie deploy-gitea",
|
|
"medium",
|
|
"Reapplies the Debian-hosted Gitea Compose service.",
|
|
True,
|
|
75,
|
|
),
|
|
HealRule(
|
|
"Gitea local HTTP",
|
|
"./jeannie deploy-gitea",
|
|
"medium",
|
|
"Local Gitea HTTP is down; redeploying Compose is usually safe but still mutates Git service runtime.",
|
|
True,
|
|
74,
|
|
),
|
|
HealRule(
|
|
"Traefik deployment",
|
|
"./jeannie promote validate",
|
|
"diagnostic",
|
|
"Validate gates first; deployment rollout failures are usually downstream of cluster/node health.",
|
|
False,
|
|
65,
|
|
),
|
|
HealRule(
|
|
"Website deployment",
|
|
"./jeannie promote validate",
|
|
"diagnostic",
|
|
"Validate gates first; deployment rollout failures are usually downstream of cluster/node health.",
|
|
False,
|
|
64,
|
|
),
|
|
HealRule(
|
|
"Traefik LoadBalancer HTTP",
|
|
"./jeannie doctor-edge",
|
|
"diagnostic",
|
|
"Edge/LB failures need diagnosis after cluster nodes are healthy.",
|
|
False,
|
|
60,
|
|
),
|
|
HealRule(
|
|
"Website public URL",
|
|
"./jeannie doctor-edge",
|
|
"diagnostic",
|
|
"Public URL failures need edge and cluster diagnosis after local services recover.",
|
|
False,
|
|
58,
|
|
),
|
|
HealRule(
|
|
"Gitea public route",
|
|
"./jeannie doctor-gitea",
|
|
"diagnostic",
|
|
"Public Gitea route needs Gitea and edge diagnosis.",
|
|
False,
|
|
55,
|
|
),
|
|
HealRule(
|
|
"No problem pods",
|
|
"./jeannie explain status",
|
|
"diagnostic",
|
|
"Pod states need root-cause context; avoid blind deletes.",
|
|
False,
|
|
50,
|
|
),
|
|
HealRule(
|
|
"Recent deployments healthy",
|
|
"./jeannie explain status",
|
|
"diagnostic",
|
|
"Deployment drift can be a symptom of node health, image pulls, or scheduling.",
|
|
False,
|
|
45,
|
|
),
|
|
HealRule(
|
|
"Pod restart pressure",
|
|
"./jeannie explain status",
|
|
"diagnostic",
|
|
"Restart pressure needs workload-specific diagnosis before mutation.",
|
|
False,
|
|
40,
|
|
),
|
|
HealRule(
|
|
"Traefik 5xx/404 signals",
|
|
"./jeannie doctor-edge",
|
|
"diagnostic",
|
|
"Use edge logs and route checks before changing config.",
|
|
False,
|
|
35,
|
|
),
|
|
]
|
|
|
|
|
|
def load_status(path: Path | None) -> dict[str, object]:
|
|
if path:
|
|
return json.loads(path.read_text(encoding="utf-8"))
|
|
|
|
process = subprocess.run(
|
|
[str(REPO_ROOT / "jeannie"), "status", "--json"],
|
|
cwd=REPO_ROOT,
|
|
capture_output=True,
|
|
text=True,
|
|
check=False,
|
|
)
|
|
if not process.stdout.strip():
|
|
print(process.stderr.strip() or "status produced no JSON", file=sys.stderr)
|
|
raise SystemExit(1)
|
|
return json.loads(process.stdout)
|
|
|
|
|
|
def status_rows(status: dict[str, object]) -> list[dict[str, str]]:
|
|
rows = status.get("rows", [])
|
|
if not isinstance(rows, list):
|
|
return []
|
|
return [
|
|
row
|
|
for row in rows
|
|
if isinstance(row, dict) and row.get("status") in {"fail", "warn"}
|
|
]
|
|
|
|
|
|
def rule_for(row: dict[str, str]) -> HealRule | None:
|
|
check = str(row.get("check", ""))
|
|
for rule in RULES:
|
|
if rule.check == check:
|
|
return rule
|
|
return None
|
|
|
|
|
|
def pimox_worker_index(row: dict[str, str]) -> int | None:
|
|
text = " ".join(str(row.get(key, "")) for key in ("summary", "detail", "check"))
|
|
match = re.search(r"\bpimox-worker-(\d{1,2})\b", text)
|
|
if not match:
|
|
return None
|
|
return int(match.group(1))
|
|
|
|
|
|
def command_for(row: dict[str, str], rule: HealRule) -> tuple[str, str, int]:
|
|
if rule.check == "Kubernetes nodes Ready":
|
|
index = pimox_worker_index(row)
|
|
if index is not None:
|
|
return (
|
|
f"./jeannie workers restart {index}",
|
|
f"Restart Pimox VM for pimox-worker-{index:02d}; a running VM with a NotReady kubelet can be wedged.",
|
|
98,
|
|
)
|
|
return rule.command, rule.reason, rule.priority
|
|
|
|
|
|
def build_plan(status: dict[str, object]) -> list[dict[str, object]]:
|
|
planned: list[dict[str, object]] = []
|
|
seen_commands: set[str] = set()
|
|
blocked = recent_failed_commands()
|
|
for row in status_rows(status):
|
|
rule = rule_for(row)
|
|
if rule is None:
|
|
planned.append(
|
|
{
|
|
"check": row.get("check", ""),
|
|
"area": row.get("area", ""),
|
|
"status": row.get("status", ""),
|
|
"command": "./jeannie explain status",
|
|
"risk": "diagnostic",
|
|
"auto": False,
|
|
"priority": 1,
|
|
"reason": "No specific heal rule exists yet.",
|
|
"summary": row.get("summary", ""),
|
|
}
|
|
)
|
|
continue
|
|
command, reason, priority = command_for(row, rule)
|
|
auto = rule.auto
|
|
if command in blocked:
|
|
failed_finding = blocked[command]
|
|
auto = False
|
|
command = "./jeannie doctor-cluster --details"
|
|
reason = f"Previous heal action did not clear this finding recently; escalate to diagnostics before retrying. Last failure: {failed_finding}"
|
|
priority = max(priority - 20, 1)
|
|
if command in seen_commands and auto:
|
|
continue
|
|
seen_commands.add(command)
|
|
planned.append(
|
|
{
|
|
"check": row.get("check", ""),
|
|
"area": row.get("area", ""),
|
|
"status": row.get("status", ""),
|
|
"command": command,
|
|
"risk": rule.risk,
|
|
"auto": auto,
|
|
"priority": priority,
|
|
"reason": reason,
|
|
"summary": row.get("summary", ""),
|
|
}
|
|
)
|
|
return planned
|
|
|
|
|
|
def impact_sort_key(item: dict[str, object]) -> tuple[int, int, str]:
|
|
status_score = 10 if item.get("status") == "fail" else 0
|
|
return (
|
|
int(item.get("priority") or 0) + status_score,
|
|
1 if item.get("auto") else 0,
|
|
str(item.get("check") or ""),
|
|
)
|
|
|
|
|
|
def highest_impact(
|
|
plan: list[dict[str, object]], *, auto_only: bool
|
|
) -> dict[str, object] | None:
|
|
candidates = (
|
|
[item for item in plan if item.get("auto")] if auto_only else list(plan)
|
|
)
|
|
if not candidates:
|
|
return None
|
|
return max(candidates, key=impact_sort_key)
|
|
|
|
|
|
def print_plan(plan: list[dict[str, object]], ai: bool) -> None:
|
|
target = highest_impact(plan, auto_only=True)
|
|
top_overall = highest_impact(plan, auto_only=False)
|
|
deferred = [
|
|
item
|
|
for item in sorted(plan, key=impact_sort_key, reverse=True)
|
|
if item is not target
|
|
]
|
|
print("Jeannie Heal Plan")
|
|
print("=================")
|
|
print("mode: one-at-a-time")
|
|
print(f"findings={len(plan)}")
|
|
print()
|
|
if target:
|
|
print("Next heal target:")
|
|
print(f" command: {target['command']}")
|
|
print(f" finding: {target['area']} / {target['check']} ({target['status']})")
|
|
print(f" impact: {target['priority']}")
|
|
print(f" why: {target['reason']}")
|
|
else:
|
|
print("Next heal target: none")
|
|
if top_overall:
|
|
print(
|
|
f"highest finding is diagnostic-only: {top_overall['area']} / {top_overall['check']}"
|
|
)
|
|
print(f"next diagnostic command: {top_overall['command']}")
|
|
print()
|
|
if deferred:
|
|
print("Deferred until next status/heal cycle:")
|
|
for item in deferred[:8]:
|
|
auto_marker = "auto" if item.get("auto") else "diagnostic"
|
|
print(
|
|
f" - {item['area']} / {item['check']} ({auto_marker}, impact={item['priority']})"
|
|
)
|
|
if ai:
|
|
print_ai_context([target] if target else ([top_overall] if top_overall else []))
|
|
|
|
|
|
def print_ai_context(plan: list[dict[str, object]]) -> None:
|
|
query = " ".join(str(item["check"]) for item in plan[:5])
|
|
if not query:
|
|
return
|
|
process = subprocess.run(
|
|
[
|
|
str(REPO_ROOT / "scripts" / "query-homelab-ai-index"),
|
|
"--citations",
|
|
"--limit",
|
|
"3",
|
|
query,
|
|
],
|
|
cwd=REPO_ROOT,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.DEVNULL,
|
|
check=False,
|
|
)
|
|
if process.returncode == 0 and process.stdout.strip():
|
|
print()
|
|
print("AI/RAG Context")
|
|
print("--------------")
|
|
print(process.stdout.strip())
|
|
|
|
|
|
def apply_plan(plan: list[dict[str, object]], yes: bool) -> int:
|
|
target = highest_impact(plan, auto_only=True)
|
|
if not target:
|
|
print("No auto-eligible heal actions.")
|
|
return 0
|
|
|
|
print("Jeannie Heal Apply")
|
|
print("==================")
|
|
command = str(target["command"])
|
|
sandbox = subprocess.run(
|
|
[str(REPO_ROOT / "scripts" / "agent-sandbox"), "check", command, "--json"],
|
|
cwd=REPO_ROOT,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.STDOUT,
|
|
check=False,
|
|
)
|
|
try:
|
|
decision = json.loads(sandbox.stdout).get("decision")
|
|
except json.JSONDecodeError:
|
|
decision = "approval"
|
|
if decision != "allow":
|
|
print(f"skip: {command} blocked by agent sandbox decision={decision}")
|
|
return 1
|
|
print(f"target: {target['area']} / {target['check']}")
|
|
if not yes:
|
|
print(f"would run: {command}")
|
|
else:
|
|
print(f"run: {command}")
|
|
process = subprocess.run(command.split(), cwd=REPO_ROOT, text=True, check=False)
|
|
if process.returncode != 0:
|
|
record_failed_action(command, target)
|
|
return process.returncode
|
|
if verify_healed(target):
|
|
print(f"healed: {target['area']} / {target['check']}")
|
|
clear_failed_action(command)
|
|
else:
|
|
print(f"still failing: {target['area']} / {target['check']}")
|
|
print("Run ./jeannie status --details for the current evidence.")
|
|
record_failed_action(command, target)
|
|
return 1
|
|
if not yes:
|
|
print()
|
|
print("Dry run only. Re-run with: ./jeannie heal apply --yes")
|
|
print(
|
|
"After it runs, use ./jeannie status again before healing the next finding."
|
|
)
|
|
else:
|
|
print("Run ./jeannie status again before healing the next finding.")
|
|
return 0
|
|
|
|
|
|
def load_heal_state() -> dict[str, object]:
|
|
if not HEAL_STATE_FILE.is_file():
|
|
return {"failed_actions": {}}
|
|
try:
|
|
document = json.loads(HEAL_STATE_FILE.read_text(encoding="utf-8"))
|
|
except (OSError, json.JSONDecodeError):
|
|
return {"failed_actions": {}}
|
|
if not isinstance(document, dict):
|
|
return {"failed_actions": {}}
|
|
if not isinstance(document.get("failed_actions"), dict):
|
|
document["failed_actions"] = {}
|
|
return document
|
|
|
|
|
|
def write_heal_state(document: dict[str, object]) -> None:
|
|
HEAL_STATE_FILE.parent.mkdir(parents=True, exist_ok=True)
|
|
HEAL_STATE_FILE.write_text(
|
|
json.dumps(document, indent=2, sort_keys=True) + "\n", encoding="utf-8"
|
|
)
|
|
|
|
|
|
def recent_failed_commands() -> dict[str, str]:
|
|
now = int(time.time())
|
|
document = load_heal_state()
|
|
failed = document.get("failed_actions", {})
|
|
blocked: dict[str, str] = {}
|
|
changed = False
|
|
for command, info in list(failed.items()):
|
|
if not isinstance(info, dict):
|
|
failed.pop(command, None)
|
|
changed = True
|
|
continue
|
|
failed_at = int(info.get("failed_at", 0) or 0)
|
|
if now - failed_at > HEAL_COOLDOWN_SECONDS:
|
|
failed.pop(command, None)
|
|
changed = True
|
|
continue
|
|
blocked[command] = str(info.get("finding") or "unknown")
|
|
if changed:
|
|
document["failed_actions"] = failed
|
|
write_heal_state(document)
|
|
return blocked
|
|
|
|
|
|
def record_failed_action(command: str, target: dict[str, object]) -> None:
|
|
document = load_heal_state()
|
|
failed = document.setdefault("failed_actions", {})
|
|
if isinstance(failed, dict):
|
|
failed[command] = {
|
|
"failed_at": int(time.time()),
|
|
"finding": f"{target.get('area')} / {target.get('check')}",
|
|
"summary": target.get("summary", ""),
|
|
}
|
|
write_heal_state(document)
|
|
print(f"recorded failed heal action cooldown: {command}")
|
|
|
|
|
|
def clear_failed_action(command: str) -> None:
|
|
document = load_heal_state()
|
|
failed = document.get("failed_actions", {})
|
|
if isinstance(failed, dict) and command in failed:
|
|
failed.pop(command, None)
|
|
write_heal_state(document)
|
|
|
|
|
|
def verify_healed(target: dict[str, object]) -> bool:
|
|
timeout = int(os.environ.get("LAB_HEAL_VERIFY_TIMEOUT_SECONDS", "180"))
|
|
interval = int(os.environ.get("LAB_HEAL_VERIFY_INTERVAL_SECONDS", "10"))
|
|
deadline = time.monotonic() + timeout
|
|
while True:
|
|
if target_is_healed(target):
|
|
return True
|
|
if time.monotonic() >= deadline:
|
|
return False
|
|
remaining = int(deadline - time.monotonic())
|
|
print(f"waiting for heal verification ({max(remaining, 0)}s left)...")
|
|
time.sleep(interval)
|
|
|
|
|
|
def target_is_healed(target: dict[str, object]) -> bool:
|
|
command = str(target.get("command") or "")
|
|
worker_match = re.search(r"\bworkers\s+restart\s+(\d+)\b", command)
|
|
if worker_match:
|
|
return pimox_worker_ready(int(worker_match.group(1)))
|
|
|
|
try:
|
|
status = load_status(None)
|
|
except SystemExit:
|
|
return False
|
|
area = str(target.get("area") or "")
|
|
check = str(target.get("check") or "")
|
|
for row in status_rows(status):
|
|
if str(row.get("area") or "") == area and str(row.get("check") or "") == check:
|
|
return False
|
|
return True
|
|
|
|
|
|
def pimox_worker_ready(index: int) -> bool:
|
|
node_name = f"pimox-worker-{index:02d}"
|
|
process = subprocess.run(
|
|
[
|
|
"kubectl",
|
|
"--kubeconfig",
|
|
str(DEFAULT_KUBECONFIG),
|
|
"get",
|
|
"node",
|
|
node_name,
|
|
"--no-headers",
|
|
],
|
|
cwd=REPO_ROOT,
|
|
text=True,
|
|
stdout=subprocess.PIPE,
|
|
stderr=subprocess.DEVNULL,
|
|
check=False,
|
|
)
|
|
if process.returncode != 0:
|
|
return False
|
|
parts = process.stdout.split()
|
|
return len(parts) >= 2 and parts[1] == "Ready"
|
|
|
|
|
|
def main() -> int:
|
|
parser = argparse.ArgumentParser(description=__doc__)
|
|
subparsers = parser.add_subparsers(dest="command", required=True)
|
|
plan_parser = subparsers.add_parser("plan")
|
|
plan_parser.add_argument("--status-json", type=Path)
|
|
plan_parser.add_argument("--ai", action="store_true")
|
|
apply_parser = subparsers.add_parser("apply")
|
|
apply_parser.add_argument("--status-json", type=Path)
|
|
apply_parser.add_argument("--yes", action="store_true")
|
|
apply_parser.add_argument("--ai", action="store_true")
|
|
args = parser.parse_args()
|
|
|
|
status = load_status(getattr(args, "status_json", None))
|
|
plan = build_plan(status)
|
|
if args.command == "plan":
|
|
print_plan(plan, args.ai)
|
|
return 0
|
|
if args.command == "apply":
|
|
print_plan(plan, args.ai)
|
|
print()
|
|
return apply_plan(plan, args.yes)
|
|
return 2
|
|
|
|
|
|
if __name__ == "__main__":
|
|
raise SystemExit(main())
|