Set-OPS-Public/scripts/raser.py

#!/usr/bin/env python3
"""Rase les VM d'un ecosysteme — la seule commande destructive du moteur.

Elle n'existe que pour rendre la reconstruction from-zero REPETABLE : un test qu'on ne
peut jouer qu'une fois, a la main, n'est pas une recette. Tout le reste du depot cree ou
reconcilie ; celle-ci detruit, et elle est ecrite en consequence.

QUATRE VERROUS, et aucun n'est cosmetique :

  1. Ne touche QUE les VMID **derives du plan** de l'instance active. Une VM qui n'est
     pas au plan n'est jamais candidate — le gabarit dore en particulier, dont le VMID
     ne se derive pas.
  2. **Le nom doit correspondre.** Un VMID present sous un AUTRE nom que celui du plan
     est la machine de quelqu'un d'autre : l'operation entiere est refusee, pas
     seulement cette VM. Ce n'est pas theorique — le 2026-08-07, une VM heritee portait
     un VMID du plan sous le nom `web-frontal-01`, et `proxmox_kvm` avait rapporte `ok`
     sans rien faire.
  3. **Il faut NOMMER l'ecosysteme** (`INSTANCE=<nom>`) et que ce nom corresponde a
     l'instance montee. Le symlink `instance/` peut pointer n'importe ou ; taper le nom
     est ce qui distingue « je rase le POC » de « je rase la production ».
  4. `CONFIRMER=true`. Sans lui : inventaire de ce qui serait detruit, et rien d'autre.

Usage :
    python3 scripts/raser.py                              # dit ce qu'il ferait
    python3 scripts/raser.py --instance chezlepro --confirmer
"""
from __future__ import annotations

import argparse
import os
import subprocess
import sys
import time
from pathlib import Path

sys.path.insert(0, str(Path(__file__).resolve().parent))
from proxmox_api import Cluster  # noqa: E402

RACINE = Path(__file__).resolve().parent.parent


def instance_active() -> tuple[Path, str]:
    """Dossier de l'instance montee, et son nom court (celui qu'il faudra taper)."""
    base = Path(os.environ.get("SETOPS_INSTANCE") or (RACINE / "instance"))
    if not base.exists():
        raise SystemExit("Aucune instance montee : rien a raser.")
    reel = base.resolve()
    nom = reel.name.removeprefix("OPS-").lower()
    return reel, nom


def plan_derive(base: Path) -> list[tuple[str, int]]:
    """(nom d'hote, vmid) tels que le PLAN les derive — la seule source légitime."""
    env = dict(os.environ, SETOPS_INSTANCE=str(base))
    r = subprocess.run([sys.executable, str(RACINE / "scripts" / "serveurs.py"), "lister"],
                       capture_output=True, text=True, env=env, cwd=RACINE)
    if r.returncode != 0:
        raise SystemExit(f"Plan illisible : {r.stderr.strip()[:200]}")
    couples = []
    for ligne in r.stdout.splitlines():
        if "vmid" not in ligne:
            continue
        nom = ligne.split()[0]
        for morceau in ligne.split():
            if morceau.rstrip(",").isdigit() and len(morceau.rstrip(",")) >= 6:
                couples.append((nom, int(morceau.rstrip(","))))
                break
    return couples


def main(argv: list[str] | None = None) -> int:
    ap = argparse.ArgumentParser(description="Rase les VM derivees du plan de l'instance active.")
    ap.add_argument("--instance", help="nom de l'ecosysteme a raser (doit correspondre au monte)")
    ap.add_argument("--confirmer", action="store_true")
    a = ap.parse_args(argv)

    base, nom_actif = instance_active()
    couples = plan_derive(base)
    if not couples:
        print("Le plan ne derive aucune VM : rien a raser.")
        return 0

    cl, _ = Cluster.depuis_hebergeur()
    rep = cl("/cluster/resources?type=vm")
    if err := cl.rate(rep):
        raise SystemExit(f"Cluster injoignable : {err}")
    reelles = {int(v["vmid"]): v for v in rep if v.get("type") == "qemu"}

    presentes, absentes, usurpees = [], [], []
    for nom, vmid in sorted(couples, key=lambda c: c[1]):
        v = reelles.get(vmid)
        if v is None:
            absentes.append((nom, vmid))
        elif v.get("name") != nom:
            usurpees.append((nom, vmid, v.get("name")))
        else:
            presentes.append((nom, vmid, v.get("node")))

    print(f"Écosystème monté : « {nom_actif} » ({base})\n")
    for nom, vmid, noeud in presentes:
        print(f"  À DÉTRUIRE   {vmid}  {nom:16} sur {noeud}")
    for nom, vmid in absentes:
        print(f"  absente      {vmid}  {nom:16} (rien à faire)")
    for nom, vmid, autre in usurpees:
        print(f"  CONFLIT      {vmid}  le plan dit « {nom} », le cluster dit « {autre} »")

    if usurpees:
        print(f"\nREFUS : {len(usurpees)} VMID du plan portent un AUTRE nom sur le cluster.\n"
              f"Ce sont des machines qui ne nous appartiennent pas. Les renommer ou corriger\n"
              f"le plan avant de raser quoi que ce soit — aucune VM n'a ete touchee.")
        return 2

    if not presentes:
        print("\nAucune VM du plan n'est presente : rien a faire.")
        return 0

    if not a.confirmer:
        print(f"\n{len(presentes)} VM seraient DETRUITES, disques compris.\n"
              f"Relancer avec :  make raser CONFIRMER=true INSTANCE={nom_actif}")
        return 0

    if (a.instance or "").lower() != nom_actif:
        print(f"\nREFUS : il faut NOMMER l'ecosysteme a raser.\n"
              f"Monte : « {nom_actif} » ; recu : « {a.instance or '(rien)'} ».\n"
              f"Taper le nom est ce qui distingue le POC de la production.")
        return 2

    echecs = 0
    for nom, vmid, noeud in presentes:
        print(f"\n  {vmid} {nom} — arret…", flush=True)
        cl(f"/nodes/{noeud}/qemu/{vmid}/status/stop", "POST")
        for _ in range(30):
            time.sleep(2)
            st = cl(f"/nodes/{noeud}/qemu/{vmid}/status/current")
            if isinstance(st, dict) and st.get("status") == "stopped":
                break
        r = cl(f"/nodes/{noeud}/qemu/{vmid}?purge=1&destroy-unreferenced-disks=1", "DELETE")
        if err := cl.rate(r):
            print(f"  {vmid} {nom} — ECHEC : {err}")
            echecs += 1
        else:
            print(f"  {vmid} {nom} — detruite")

    print(f"\n{len(presentes) - echecs}/{len(presentes)} VM detruites."
          + (f" {echecs} echec(s)." if echecs else ""))
    return 1 if echecs else 0


if __name__ == "__main__":
    raise SystemExit(main())