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"))

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 = Path(os.environ.get("SETOPS_INSTANCE") or (RACINE / "instance"))
    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 constater(api: Cluster, place: dict) -> list[dict]:
    """Confronte chaque valeur au cluster. Renvoie une ligne par objet."""
    lignes: list[dict] = []

    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 pont doit exister SUR LE NOEUD retenu : un pont partiel est un piege, la VM ne
    # demarre que sur certains noeuds et l'erreur ne le dit pas.
    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")]
    voulu = place.get("proxmox_clone_pont")
    lignes.append({"objet": "pont", "voulu": voulu,
                   "existe": voulu in ponts, "offre": ponts,
                   "note": "" if n in noeuds else "noeud introuvable : pont non verifiable"})

    # 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':<10} {'DEMANDE':<22} {'VERDICT'}")
    ok = True
    for l in lignes:
        verdict = "existe" if l["existe"] else "INTROUVABLE"
        ok = ok and l["existe"]
        print(f"  {l['objet']:<10} {str(l['voulu']):<22} {verdict}")
        if l.get("note"):
            print(f"  {'':<10} {'':<22} -> {l['note']}")
        if not l["existe"] and l["offre"]:
            offert = ", ".join(str(x) for x in l["offre"][:6])
            print(f"  {'':<10} {'':<22} offert : {offert}")
    print("\n" + ("CONFORME : les quatre objets de placement existent sur ce cluster."
                  if ok else
                  "ECART : le tenant designe des 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

    lignes = constater(api, place)
    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, pont et gabarit existent sur "
              f"« {depot.name} ».")
        return 0
    return 0 if afficher(lignes, source, depot) else 1


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