oikos/oikos/scheduler.py

#!/usr/bin/env python3
"""oikos/scheduler.py — the Observe stage: periodic probes -> Signals + state cache.

Intended to run under a systemd timer (see scripts/sync/linux/ for the
existing timer pattern this should mirror — deploy target: LXC 105
alongside the MCP server) every 5-15 minutes. Each run:

  1. Probes every service's HTTP health and writes oikos/state.json — the
     snapshot `homelab service <name> health` serves by default (cache-
     first reads); pass --live on that command to force a fresh probe
     instead of trusting the cache.
  2. Probes disk usage on hubris + strong (best-effort SSH; degrades to a
     "probe-unreachable" info Signal rather than a false disk-threshold
     alarm when unreachable).
  3. Runs the Week-3 drift detectors (oikos/drift.py).
  4. Raises Signals for anything over threshold or drifted — skipping
     duplicates via oikos_signal.open_signal_for() — and auto-resolves
     any open Signal whose condition has since cleared.

Temperature probing is NOT implemented here: LXCs don't expose host
sensors, and hubris/strong's actual sensor path (lm-sensors vs vendor
tool) hasn't been confirmed on either box yet — a real check needs that
groundwork first rather than a guessed command. Backup-freshness is
likewise deferred: `backs-up-to` isn't populated in inventory yet (see
oikos/ontology.yaml — it's a documented thin field), so there's nothing
structured to check freshness against.

CLI:
    python3 oikos/scheduler.py run             # probe, raise signals, write state.json
    python3 oikos/scheduler.py run --dry-run   # probe and print only, no side effects
"""

from __future__ import annotations

import json
import subprocess
import sys
from datetime import datetime, timezone
from pathlib import Path

import yaml

REPO = Path(__file__).resolve().parent.parent
sys.path.insert(0, str(REPO))
from oikos import drift as oikos_drift  # noqa: E402
from oikos import signal as oikos_signal  # noqa: E402

INVENTORY = REPO / "inventory.yaml"
STATE_FILE = REPO / "oikos" / "state.json"

DISK_WARN_PCT = 85
DISK_CRIT_PCT = 95
HEALTH_TIMEOUT_S = 5


def _load_inventory() -> dict:
    return yaml.safe_load(INVENTORY.read_text())


# Services whose public `url`/`endpoint` isn't a plain-GET health check —
# mirrors the special-casing bin/homelab's `doctor` command already does.
# A generic per-service `health:` inventory field (checked URL/command
# distinct from the public url) would remove the need for this; until
# that's backfilled, these are the two known exceptions.
_HEALTH_OVERRIDES = {
    # FastMCP's streamable-http endpoint expects a JSON-RPC POST handshake,
    # not a bare GET — a plain curl gets 4xx even when the server is fully
    # healthy (confirmed against the live server: consistently 400, never
    # 2xx/3xx, for a GET). Any HTTP response at all (vs. no response/
    # connection refused) proves the process is up and answering, which is
    # what "service-down" alerting actually cares about here.
    "homelab_mcp": {"extra_curl_args": ["-H", "Accept: text/event-stream"],
                    "any_response_ok": True},
    "secrets_issuance": {
        "url_transform": lambda url: url.rstrip("/").rsplit("/", 1)[0] + "/health",
    },
    # Token-gated at the app level (401 without a token is correct, not
    # down) — see knowledge/wiki/containers/128-trmnl.md, which documents a dedicated
    # /health endpoint returning 200 unauthenticated.
    "trmnl": {"url_transform": lambda url: url.rstrip("/") + "/health"},
}


def probe_service_health(name: str, entry: dict) -> dict:
    url = entry.get("url") or entry.get("endpoint")
    if not url:
        return {"service": name, "checked": False}
    override = _HEALTH_OVERRIDES.get(name, {})
    check_url = override.get("url_transform", lambda u: u)(url)
    curl_cmd = ["curl", "-sS", "-o", "/dev/null", "-w", "%{http_code}",
               "--max-time", str(HEALTH_TIMEOUT_S), *override.get("extra_curl_args", []), check_url]
    try:
        proc = subprocess.run(curl_cmd, capture_output=True, text=True)
        code = proc.stdout.strip()
    except OSError:
        code = ""
    ok = bool(code) if override.get("any_response_ok") else code.startswith(("2", "3"))
    return {"service": name, "checked": True, "url": url, "checked_url": check_url,
           "http_code": code or None, "ok": ok}


def _ssh(host_ip: str, remote_cmd: str, timeout: int = 10) -> str | None:
    try:
        proc = subprocess.run(
            ["ssh", "-o", "BatchMode=yes", "-o", f"ConnectTimeout={min(timeout, 5)}",
             f"root@{host_ip}", remote_cmd],
            capture_output=True, text=True, timeout=timeout,
        )
    except (subprocess.TimeoutExpired, OSError):
        return None
    return proc.stdout.strip() if proc.returncode == 0 else None


def probe_disk(name: str, entry: dict) -> dict | None:
    """`df -P /` on a proxmox-host's own root fs. LXC-level disk usage is
    already covered by MCP's get_lxc_state; this probe is host-level."""
    if entry.get("kind") != "proxmox-host" or not entry.get("lan_ip"):
        return None
    out = _ssh(entry["lan_ip"], "df -P / | tail -1")
    if out is None:
        return {"host": name, "checked": False}
    parts = out.split()
    if len(parts) < 5 or not parts[4].rstrip("%").isdigit():
        return {"host": name, "checked": False}
    return {"host": name, "checked": True, "used_pct": int(parts[4].rstrip("%"))}


def run(dry_run: bool = False) -> dict:
    inv = _load_inventory()
    now = datetime.now(timezone.utc).isoformat(timespec="seconds")

    services_state = {}
    for name, entry in inv.get("services", {}).items():
        if not isinstance(entry, dict):
            continue
        result = probe_service_health(name, entry)
        services_state[name] = result
        if not result.get("checked"):
            continue
        sig_kind = "service-down"
        if result["ok"]:
            open_sig = oikos_signal.open_signal_for(f"service:{name}", sig_kind)
            if open_sig and not dry_run:
                oikos_signal.resolve(open_sig["id"], note="health probe recovered")
        else:
            if not dry_run and oikos_signal.open_signal_for(f"service:{name}", sig_kind) is None:
                oikos_signal.raise_signal(
                    sig_kind, "critical", f"service:{name}",
                    f"{result['url']} -> {result.get('http_code') or 'no response'}",
                    likely_cause="backend down, crashed, or ingress misconfigured",
                    recommended_action={"runbook": "service-health-check", "risk": "reversible_low"},
                    verification=f"homelab service {name} health --live",
                )

    hosts_state = {}
    for name, entry in inv.get("hosts", {}).items():
        disk = probe_disk(name, entry)
        if disk is None:
            continue
        hosts_state[name] = disk
        if not disk.get("checked"):
            continue
        pct = disk["used_pct"]
        sig_kind = "disk-threshold"
        if pct >= DISK_CRIT_PCT:
            severity = "critical"
        elif pct >= DISK_WARN_PCT:
            severity = "warning"
        else:
            severity = None
        open_sig = oikos_signal.open_signal_for(f"host:{name}", sig_kind)
        if severity is None:
            if open_sig and not dry_run:
                oikos_signal.resolve(open_sig["id"], note=f"disk usage back to {pct}%")
        elif open_sig is None and not dry_run:
            oikos_signal.raise_signal(
                sig_kind, severity, f"host:{name}", f"root filesystem {pct}% used",
                likely_cause="data growth or a runaway log",
                recommended_action={"runbook": "config-change-deploy", "risk": "config_mutation"},
                verification=f"homelab ssh {name} -- df -h /",
            )

    drift_findings = oikos_drift.run_all(inv, raise_signals=not dry_run)

    snapshot = {
        "generated_at": now,
        "services": services_state,
        "hosts": hosts_state,
        "drift_findings": len(drift_findings),
    }
    if not dry_run:
        STATE_FILE.write_text(json.dumps(snapshot, indent=2) + "\n")
    return snapshot


def read_state() -> dict | None:
    """Load the last scheduler snapshot, or None if it's never run."""
    if not STATE_FILE.exists():
        return None
    try:
        return json.loads(STATE_FILE.read_text())
    except json.JSONDecodeError:
        return None


def cached_service_health(name: str) -> dict | None:
    state = read_state()
    if state is None:
        return None
    entry = state.get("services", {}).get(name)
    if entry is None:
        return None
    return {**entry, "as_of": state.get("generated_at")}


def main() -> int:
    import argparse
    p = argparse.ArgumentParser(description="oikos scheduler (Observe stage)")
    sub = p.add_subparsers(dest="cmd", required=True)
    r = sub.add_parser("run")
    r.add_argument("--dry-run", action="store_true")
    args = p.parse_args()
    if args.cmd == "run":
        snapshot = run(dry_run=args.dry_run)
        print(json.dumps(snapshot, indent=2))
    return 0


if __name__ == "__main__":
    sys.exit(main())