193 lines
6.6 KiB
Python
193 lines
6.6 KiB
Python
#!/usr/bin/env python3
|
|
"""Generate troubleshooting runbooks from service documentation using Ollama LLM.
|
|
|
|
Scans docs/services/individual/*.md, finds matching compose files, and produces
|
|
structured runbooks in docs/runbooks/.
|
|
|
|
Cron: monthly or on-demand
|
|
0 9 1 * * cd /home/homelab/organized/repos/homelab && python3 scripts/runbook-generator.py --all
|
|
"""
|
|
|
|
import argparse
|
|
import hashlib
|
|
import logging
|
|
import sqlite3
|
|
import sys
|
|
from pathlib import Path
|
|
|
|
sys.path.insert(0, str(Path(__file__).parent))
|
|
from lib.ollama import ollama_generate, ollama_available, OllamaUnavailableError
|
|
|
|
log = logging.getLogger(__name__)
|
|
|
|
REPO_DIR = Path("/home/homelab/organized/repos/homelab")
|
|
DOCS_DIR = REPO_DIR / "docs" / "services" / "individual"
|
|
RUNBOOKS_DIR = REPO_DIR / "docs" / "runbooks"
|
|
HOSTS_DIR = REPO_DIR / "hosts"
|
|
DB_PATH = Path(__file__).parent / "runbook.db"
|
|
|
|
|
|
def init_db(db_path: Path) -> sqlite3.Connection:
|
|
conn = sqlite3.connect(str(db_path))
|
|
conn.execute(
|
|
"CREATE TABLE IF NOT EXISTS generated "
|
|
"(service TEXT PRIMARY KEY, doc_hash TEXT, generated_at TEXT)"
|
|
)
|
|
conn.commit()
|
|
return conn
|
|
|
|
|
|
def get_stored_hash(conn: sqlite3.Connection, service: str) -> str | None:
|
|
row = conn.execute(
|
|
"SELECT doc_hash FROM generated WHERE service = ?", (service,)
|
|
).fetchone()
|
|
return row[0] if row else None
|
|
|
|
|
|
def record_generation(conn: sqlite3.Connection, service: str, doc_hash: str) -> None:
|
|
from datetime import datetime
|
|
|
|
conn.execute(
|
|
"INSERT OR REPLACE INTO generated (service, doc_hash, generated_at) VALUES (?, ?, ?)",
|
|
(service, doc_hash, datetime.now().isoformat()),
|
|
)
|
|
conn.commit()
|
|
|
|
|
|
def file_hash(path: Path) -> str:
|
|
return hashlib.sha256(path.read_bytes()).hexdigest()
|
|
|
|
|
|
def find_compose_file(service_name: str) -> Path | None:
|
|
"""Try to locate a compose file for the service under hosts/."""
|
|
candidates = list(HOSTS_DIR.rglob(f"*{service_name}*"))
|
|
for c in candidates:
|
|
if c.name in ("docker-compose.yml", "docker-compose.yaml", "compose.yml", "compose.yaml"):
|
|
return c
|
|
if c.is_dir():
|
|
for compose_name in ("docker-compose.yml", "docker-compose.yaml", "compose.yml", "compose.yaml"):
|
|
compose_path = c / compose_name
|
|
if compose_path.exists():
|
|
return compose_path
|
|
return None
|
|
|
|
|
|
def generate_runbook(doc_content: str, compose_content: str | None) -> str:
|
|
compose_section = f"\n\nCompose Config:\n{compose_content}" if compose_content else ""
|
|
prompt = (
|
|
"Generate a troubleshooting runbook for this service. "
|
|
"Format as a decision tree:\n\n"
|
|
f"Service Doc:\n{doc_content}"
|
|
f"{compose_section}\n\n"
|
|
"Format:\n"
|
|
"## Common Issues\n"
|
|
"### Issue: [symptom]\n"
|
|
"1. Check: [what to check]\n"
|
|
"2. Fix: [what to do]\n"
|
|
"3. Verify: [how to confirm fix]\n\n"
|
|
"Include: restart procedures, log locations, dependency checks."
|
|
)
|
|
return ollama_generate(prompt, num_predict=3000, timeout=180)
|
|
|
|
|
|
def list_service_docs() -> list[Path]:
|
|
if not DOCS_DIR.exists():
|
|
return []
|
|
return sorted(DOCS_DIR.glob("*.md"))
|
|
|
|
|
|
def process_service(conn: sqlite3.Connection, doc_path: Path, force: bool, dry_run: bool) -> bool:
|
|
"""Process a single service doc. Returns True if a runbook was generated."""
|
|
service_name = doc_path.stem
|
|
current_hash = file_hash(doc_path)
|
|
|
|
if not force:
|
|
stored = get_stored_hash(conn, service_name)
|
|
if stored == current_hash:
|
|
log.debug("Skipping %s — hash unchanged", service_name)
|
|
return False
|
|
|
|
log.info("Generating runbook for: %s", service_name)
|
|
|
|
doc_content = doc_path.read_text()
|
|
|
|
# Try to find a matching compose file
|
|
compose_path = find_compose_file(service_name)
|
|
compose_content = None
|
|
if compose_path:
|
|
log.info(" Found compose: %s", compose_path)
|
|
compose_content = compose_path.read_text()
|
|
else:
|
|
log.debug(" No compose file found for %s", service_name)
|
|
|
|
runbook = generate_runbook(doc_content, compose_content)
|
|
|
|
if dry_run:
|
|
print(f"\n{'='*60}")
|
|
print(f"RUNBOOK: {service_name}")
|
|
print(f"{'='*60}")
|
|
print(runbook)
|
|
return True
|
|
|
|
RUNBOOKS_DIR.mkdir(parents=True, exist_ok=True)
|
|
out_path = RUNBOOKS_DIR / f"{service_name}.md"
|
|
out_path.write_text(f"# Troubleshooting Runbook: {service_name}\n\n{runbook}\n")
|
|
log.info(" Wrote %s", out_path)
|
|
|
|
record_generation(conn, service_name, current_hash)
|
|
return True
|
|
|
|
|
|
def main() -> None:
|
|
parser = argparse.ArgumentParser(description="Generate troubleshooting runbooks from service docs via LLM")
|
|
parser.add_argument("--service", type=str, help="Generate runbook for a single service (stem name)")
|
|
parser.add_argument("--all", action="store_true", help="Generate runbooks for all services")
|
|
parser.add_argument("--force", action="store_true", help="Regenerate even if hash has not changed")
|
|
parser.add_argument("--dry-run", action="store_true", help="Print runbook without writing files")
|
|
parser.add_argument("--verbose", "-v", action="store_true", help="Enable debug logging")
|
|
args = parser.parse_args()
|
|
|
|
logging.basicConfig(
|
|
level=logging.DEBUG if args.verbose else logging.INFO,
|
|
format="%(asctime)s [%(levelname)s] %(name)s: %(message)s",
|
|
)
|
|
|
|
if not args.service and not args.all:
|
|
parser.error("Specify --service NAME or --all")
|
|
|
|
if not ollama_available():
|
|
log.error("Ollama is not reachable — aborting")
|
|
sys.exit(1)
|
|
|
|
conn = init_db(DB_PATH)
|
|
generated_count = 0
|
|
|
|
if args.service:
|
|
doc_path = DOCS_DIR / f"{args.service}.md"
|
|
if not doc_path.exists():
|
|
log.error("Service doc not found: %s", doc_path)
|
|
sys.exit(1)
|
|
if process_service(conn, doc_path, args.force, args.dry_run):
|
|
generated_count += 1
|
|
else:
|
|
docs = list_service_docs()
|
|
if not docs:
|
|
log.warning("No service docs found in %s", DOCS_DIR)
|
|
sys.exit(0)
|
|
log.info("Found %d service docs", len(docs))
|
|
for doc_path in docs:
|
|
try:
|
|
if process_service(conn, doc_path, args.force, args.dry_run):
|
|
generated_count += 1
|
|
except OllamaUnavailableError as e:
|
|
log.error("Ollama failed for %s: %s — stopping", doc_path.stem, e)
|
|
break
|
|
except Exception:
|
|
log.exception("Error processing %s — skipping", doc_path.stem)
|
|
|
|
log.info("Generated %d runbook(s)", generated_count)
|
|
|
|
|
|
if __name__ == "__main__":
|
|
main()
|