Set-OPS-Public/scripts/devis_placement.py

#!/usr/bin/env python3
"""Devis du PLACEMENT — les quatre objets d'hebergeur que le tenant monte designe.

D-80 : un tenant est agnostique de son underlay ; tout son adressage derive du seed.
Ce qui le lie a une fabric tient en QUATRE valeurs, et elles nomment des objets qui
appartiennent a l'hebergeur :

    proxmox_clone_noeud        sur quel noeud cloner
    proxmox_clone_stockage     ou poser le disque
    proxmox_clone_pont         ou brancher la carte
    proxmox_clone_vmid_modele  quel gabarit cloner

**P37 en verifie trois**, statiquement, contre les listes que l'hebergeur DECLARE
(`proxmox-hebergeur.yml`). C'est utile et insuffisant : une liste declaree peut elle-meme
etre perimee, et le GABARIT n'y figure pas — c'est un objet du cluster, pas une
declaration.

Ce devis interroge donc le CLUSTER. Il ne verifie pas ce qui est ecrit : il verifie ce qui
existe. Sur un site inconnu, c'est la premiere chose a savoir — sans quoi l'ecart se
decouvre au premier clone, apres quarante minutes de deploiement.

N'ECRIT RIEN. Lecture seule, sans effet de bord.

  python3 scripts/devis_placement.py            # tableau lisible
  python3 scripts/devis_placement.py --verifier # code de sortie (0 = tout existe)
"""
from __future__ import annotations

import argparse
import os
import sys
from pathlib import Path

import yaml

RACINE = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(RACINE / "scripts"))

import underlay as underlay_mod  # noqa: E402
from inventory_rules import instance_courante  # noqa: E402
from proxmox_api import Cluster  # noqa: E402

INVENTAIRES = ("principal", "production", "lab")


def placement_du_tenant() -> tuple[dict, Path | None]:
    """Les valeurs de placement declarees par le tenant VISE.

    SETOPS_INSTANCE d'abord ; le symlink `instance/` n'en est que le cas courant. Code en
    dur, il rendait ce devis incapable de regarder un AUTRE tenant : viser patient 0
    mesurait en silence le placement de l'instance montee, et rendait un verdict juste
    pour le mauvais tenant (mesure du 2026-08-20 — les deux avaient les memes quatre
    valeurs, ce qui est exactement la circonstance ou l'erreur ne se voit pas).
    """
    base = instance_courante()
    for nom in INVENTAIRES:
        f = base / "inventories" / nom / "group_vars" / "proxmox.yml"
        if f.is_file():
            return (yaml.safe_load(f.read_text(encoding="utf-8")) or {}), f
    return {}, None


def _liste(api: Cluster, chemin: str) -> list:
    """Une reponse de l'API garantie ITERABLE — ou un refus lisible.

    POURQUOI (mesure du 2026-08-20, un soir de reconstruction). Sur erreur,
    `Cluster.__call__` rend `{"_erreur": "..."}` — un DICT. Itere comme une liste, un
    dict rend ses CLEFS : le devis mourait sur
    `AttributeError: 'str' object has no attribute 'get'`, dix lignes de trace Python
    pour dire « le nom `asgard` ne se resout pas ».

    C'est le pire endroit pour un message illisible : ce devis est precisement ce qu'on
    lance AVANT quarante minutes de deploiement, quand on cherche a savoir si le terrain
    est bon. `Cluster.rate()` existait pour ca ; il n'etait pas appele ici.
    """
    rep = api(chemin)
    faute = api.rate(rep)
    if faute:
        raise SystemExit(
            f"Le cluster ne repond pas sur `{chemin}` : {faute}\n"
            f"  hote interroge : {api.hote}\n"
            f"  a verifier : le nom se resout-il d'ici, le VPN est-il monte, "
            f"le jeton d'API est-il encore valide ?")
    if not isinstance(rep, list):
        raise SystemExit(
            f"Reponse inattendue du cluster sur `{chemin}` : "
            f"{type(rep).__name__} au lieu d'une liste.\n"
            f"  hote interroge : {api.hote}")
    return rep


def _ponts_derives(base: Path | None) -> set[str]:
    """Les reseaux ou les VM de CE tenant atterriront reellement.

    Ce ne sont pas les valeurs du plan : `instancier` pose dans chaque hote le pont
    DERIVE de sa zone (`proxmox_pont` = le VNet du tenant), et `make creer-vm` le passe au
    clone en ecrasant le defaut. C'est donc la seule chose a confronter au cluster.
    """
    ponts: set[str] = set()
    if base is None:
        return ponts
    for f in sorted(base.glob("inventories/*/hosts.yml")):
        data = yaml.safe_load(f.read_text(encoding="utf-8")) or {}

        def marcher(noeud):
            if not isinstance(noeud, dict):
                return
            for hote in (noeud.get("hosts") or {}).values():
                if isinstance(hote, dict) and hote.get("proxmox_pont"):
                    ponts.add(str(hote["proxmox_pont"]))
            for enfant in (noeud.get("children") or {}).values():
                marcher(enfant or {})
        marcher((data.get("all") or {}))
    return ponts


def constater(api: Cluster, place: dict, base: Path | None = None,
              routage: str = "sdn") -> list[dict]:
    """Confronte chaque valeur au cluster. Renvoie une ligne par objet."""
    lignes: list[dict] = []
    place = {**place, "_instance": base}

    noeuds = [n.get("node") for n in _liste(api, "/nodes")]
    voulu = place.get("proxmox_clone_noeud")
    lignes.append({"objet": "noeud", "voulu": voulu,
                   "existe": voulu in noeuds, "offre": noeuds})

    # Un stockage ne suffit pas : il doit accepter le contenu `images`, sinon on ne peut
    # pas y poser un disque de VM — et l'erreur ne se lit qu'au clone.
    stockages = {}
    for s in _liste(api, "/storage"):
        stockages[s.get("storage")] = str(s.get("content") or "")
    voulu = place.get("proxmox_clone_stockage")
    lignes.append({"objet": "stockage", "voulu": voulu,
                   "existe": voulu in stockages and "images" in stockages[voulu],
                   "offre": [k for k, v in stockages.items() if "images" in v],
                   "note": ("" if voulu not in stockages or "images" in stockages[voulu]
                            else f"present mais n'accepte pas `images` ({stockages[voulu]})")})

    # LE RESEAU OU LES VM ATTERRISSENT — pas `proxmox_clone_pont`.
    #
    # Mesure du 2026-08-20, remarque de l'exploitant : « le pont ne me semble pas approprie
    # depuis qu'on cree des VNets par tenant ». Il avait raison, et le defaut etait pire
    # que cosmetique. Ce devis validait `vmbr1` — le REPLI des clones manuels — pendant que
    # le point d'attache reel des VM de la flotte n'etait controle par personne :
    # `creer-vm` lit le pont DERIVE de chaque hote (`SETOPS_PONT`, le VNet de sa zone) et
    # ecrase le defaut. Resultat : « CONFORME » sur un tenant dont aucun VNet n'existait.
    #
    # D-80 corrigee le 2026-08-13 dit deja que la liaison de placement est nœud, stockage
    # et gabarit — le pont se derive. Le devis, lui, continuait de compter quatre objets.
    ponts_derives = sorted(_ponts_derives(place.get("_instance")))
    if routage == "sdn":
        vnets = {v.get("vnet") for v in _liste(api, "/cluster/sdn/vnets")}
        manquants = [p for p in ponts_derives if p not in vnets]
        lignes.append({
            "objet": "reseaux VM", "voulu": ", ".join(ponts_derives) or "(aucun)",
            "existe": bool(ponts_derives) and not manquants,
            "offre": sorted(vnets),
            "note": ("" if not manquants else
                     f"VNet(s) absent(s) du cluster : {', '.join(manquants)} — "
                     f"passer `make sdn-appliquer` AVANT de creer les VM")})
    else:
        n = place.get("proxmox_clone_noeud")
        ponts: list[str] = []
        if n in noeuds:
            ponts = [i.get("iface") for i in _liste(api, f"/nodes/{n}/network")
                     if str(i.get("type")) in ("bridge", "OVSBridge")]
        manquants = [p for p in ponts_derives if p not in ponts]
        lignes.append({
            "objet": "reseaux VM", "voulu": ", ".join(ponts_derives) or "(aucun)",
            "existe": bool(ponts_derives) and not manquants,
            "offre": ponts,
            "note": ("noeud introuvable : ponts non verifiables" if n not in noeuds else
                     "" if not manquants else
                     f"pont(s) absent(s) sur {n} : {', '.join(manquants)}")})

    # Le gabarit : celui que P37 ne peut PAS voir. Presence, et surtout `template=1` —
    # cloner une VM ordinaire marcherait, et produirait quatorze copies d'une machine
    # vivante.
    vmid = place.get("proxmox_clone_vmid_modele")
    vms = {v.get("vmid"): v for v in _liste(api, "/cluster/resources?type=vm")}
    trouve = vms.get(int(vmid)) if vmid is not None else None
    est_gabarit = bool(trouve and str(trouve.get("template", 0)) == "1")
    lignes.append({"objet": "gabarit", "voulu": vmid,
                   "existe": est_gabarit,
                   "offre": [f"{v.get('vmid')} ({v.get('name')})" for v in vms.values()
                             if str(v.get("template", 0)) == "1"],
                   "note": ("" if est_gabarit else
                            (f"VMID {vmid} existe (« {trouve.get('name')} ») mais n'est PAS "
                             f"un gabarit" if trouve else f"aucune VM au VMID {vmid}"))})
    return lignes


def afficher(lignes: list[dict], source: Path | None, depot: Path) -> bool:
    # `.resolve()` : sans lui, un tenant monte par symlink s'annonce « instance » — le nom
    # du lien, pas celui du tenant. Un devis doit nommer ce qu'il a mesure.
    tenant = source.parents[3].resolve().name if source else "?"
    print(f"Devis du placement — tenant « {tenant} » sur l'hebergeur « {depot.name} »\n")
    print(f"  {'OBJET':<12} {'DEMANDE':<34} {'VERDICT'}")
    ok = True
    for l in lignes:
        verdict = "existe" if l["existe"] else "INTROUVABLE"
        ok = ok and l["existe"]
        print(f"  {l['objet']:<12} {str(l['voulu']):<34} {verdict}")
        if l.get("note"):
            print(f"  {'':<12} {'':<34} -> {l['note']}")
        if not l["existe"] and l["offre"]:
            offert = ", ".join(str(x) for x in l["offre"][:6])
            print(f"  {'':<12} {'':<34} offert : {offert}")
    print("\n" + ("CONFORME : le noeud, le stockage, le gabarit et les reseaux des VM "
                  "existent sur ce cluster." if ok else
                  "ECART : le tenant a besoin d'objets que ce cluster n'offre pas."))
    return ok


def main() -> int:
    ap = argparse.ArgumentParser(description=__doc__)
    ap.add_argument("--verifier", action="store_true", help="code de sortie seulement")
    a = ap.parse_args()

    place, source = placement_du_tenant()
    if not place:
        print("Aucun placement declare par le tenant monte — rien a confronter.")
        return 0
    try:
        api, depot = Cluster.depuis_hebergeur()
    except Exception as e:  # noqa: BLE001 — l'API est un tiers : on rend la cause lisible
        print(f"Cluster injoignable : {e}", file=sys.stderr)
        return 2

    base = source.parents[3] if source else None
    routage = underlay_mod.routage_tenants(underlay_mod.charger())
    lignes = constater(api, place, base, routage)
    if a.verifier:
        mauvais = [l for l in lignes if not l["existe"]]
        if mauvais:
            for l in mauvais:
                print(f"placement : {l['objet']} « {l['voulu']} » introuvable"
                      + (f" — {l['note']}" if l.get("note") else ""), file=sys.stderr)
            return 1
        print(f"Placement conforme : noeud, stockage, gabarit et reseaux des VM "
              f"existent sur « {depot.name} ».")
        return 0
    return 0 if afficher(lignes, source, depot) else 1


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