#!/usr/bin/env python3 """Regles partagees pour les inventaires statiques Set-OPS.""" from __future__ import annotations import ipaddress import os import re import tempfile from contextlib import contextmanager from pathlib import Path import yaml GROUPES_OPERATIONNELS_PREFIXES = ("serveur_", "client_") GROUPE_HOTES_ACTIFS = "hotes_actifs" GROUPE_HOTES_PLANIFIES = "hotes_planifies" GROUPES_ETAT_HOTE = {GROUPE_HOTES_ACTIFS, GROUPE_HOTES_PLANIFIES} RACINE_DEPOT = Path(__file__).resolve().parents[1] ORDRE_INVENTAIRE = ("principal", "production") ORDRE_INVENTAIRE_MODELE = ("lab", "principal", "production") @contextmanager def ecriture_atomique(path: Path, encoding: str = "utf-8"): """Ouvre `path` en ecriture de facon ATOMIQUE : tout, ou rien. POURQUOI, ET POURQUOI ICI (2026-09-08). Douze sites du moteur ecrivaient le plan par `path.open("w")`. Cette forme TRONQUE le fichier avant d'ecrire : entre les deux, il est VIDE. Une exception dans `yaml.safe_dump`, un disque plein, un Ctrl-C — et `instance/plan/serveurs.yml` reste tronque. L'asymetrie est ce qui rend le defaut grave : `hosts.yml` se regenere d'un `make instancier-appliquer`, le PLAN ne se regenere de rien. C'est la source unique de verite ; git est le filet, mais encore faut-il savoir qu'on est tombe. Le remede tient en deux gestes : ecrire dans un temporaire du MEME dossier, puis `os.replace()` — atomique sur POSIX (meme systeme de fichiers, donc simple rename). Le lecteur voit l'ancien fichier, puis le nouveau. Jamais rien entre les deux. POSE DANS CE MODULE, ET PAS DANS CHAQUE APPELANT. Les sept ecrivains l'importent deja. Douze copies d'un meme geste ne vieillissent pas ensemble — c'est exactement la lecon des neuf resolutions d'instance que **P41** garde depuis. Une source, pas douze. Usage : with ecriture_atomique(chemin) as f: f.write(entete) yaml.safe_dump(donnees, f) """ path = Path(path) path.parent.mkdir(parents=True, exist_ok=True) # Le temporaire vit DANS le dossier cible : `os.replace` n'est atomique qu'au sein # d'un meme systeme de fichiers. Un /tmp sur une autre partition casserait la # garantie sans rien dire. fd, tmp = tempfile.mkstemp(dir=str(path.parent), prefix=f".{path.name}.", suffix=".tmp") try: with os.fdopen(fd, "w", encoding=encoding) as fichier: yield fichier fichier.flush() # Sans `fsync`, le rename peut atteindre le disque AVANT le contenu : au # retour d'une coupure brutale, on aurait un fichier neuf et vide — le defaut # qu'on vient de fermer, deplace d'un cran. os.fsync(fichier.fileno()) # Conserver les droits d'un fichier existant : `mkstemp` cree en 0600, et le plan # est lisible par le groupe sur les runners. if path.exists(): os.chmod(tmp, path.stat().st_mode & 0o7777) os.replace(tmp, path) tmp = None finally: if tmp is not None and os.path.exists(tmp): os.unlink(tmp) def dependances_groupes(racine: Path | None = None) -> dict: """Le registre des dependances causales entre groupes (docs/dependances-groupes.yml).""" f = (racine or RACINE_DEPOT) / "docs" / "dependances-groupes.yml" if not f.is_file(): return {} return (yaml.safe_load(f.read_text(encoding="utf-8")) or {}).get("groupes") or {} def variable_effective(cle: str, dossier_group_vars: Path | None = None, role_defaut: str | None = None, racine: Path | None = None) -> str | None: """La valeur d'une variable pour une instance : ses group_vars, sinon le defaut du role. SOURCE UNIQUE (2026-08-22). Trois lecteurs en avaient besoin le meme jour — P35 pour l'interrupteur `_bd`, le controle de dependances pour les clauses `sauf_si`, et le generateur d'inventaire. Trois copies auraient recommence exactement ce qu'on vient de refermer avec la resolution d'instance. Lecture TEXTUELLE : aucune voute n'est dechiffree, aucun Ansible n'est lance. Ces controles doivent rester instantanes — c'est ce qui fait qu'on les lance. """ base = racine or RACINE_DEPOT dossier = dossier_group_vars or (dossier_inventaire() / "group_vars") if dossier.is_dir(): for f in sorted(dossier.glob("**/*.yml")): try: d = yaml.safe_load(f.read_text(encoding="utf-8")) or {} except (OSError, yaml.YAMLError): continue if isinstance(d, dict) and cle in d: return str(d[cle]) if role_defaut: f = base / "roles" / role_defaut / "defaults" / "main.yml" if f.is_file(): d = yaml.safe_load(f.read_text(encoding="utf-8")) or {} if isinstance(d, dict) and cle in d: return str(d[cle]) return None def instance_courante() -> Path: """L'instance VISEE : `SETOPS_INSTANCE`, sinon le symlink `instance/`. SOURCE UNIQUE. Neuf modules en portaient chacun leur copie, et cinq defauts en sont sortis en cinq jours (2026-08-18 au 22) : P03 comparait chaque instance a l'inventaire d'une autre ; `verifier_ports` codait `principal/` en dur ; `verifier_intrants` et `_frontiere_absente` lisaient le symlink au lieu de la variable ; `devis_placement` rendait un verdict juste sur le mauvais tenant ; P35 aussi. Aucun n'etait une faute d'inattention : chacun etait une copie ecrite de bonne foi, a un moment ou le besoin semblait local. C'est le mode de panne de la duplication — pas l'erreur, mais la DERIVE, qui ne se voit jamais depuis l'interieur d'un fichier. """ return Path(os.environ.get("SETOPS_INSTANCE") or (RACINE_DEPOT / "instance")) def dossier_inventaire(instance: Path | None = None, ordre: tuple[str, ...] = ORDRE_INVENTAIRE) -> Path: """Le REPERTOIRE d'inventaire de l'instance, dans l'ordre de preference donne. Trois niveaux, et le troisieme manquait a la moitie des copies : un `hosts.yml` existant, puis un REPERTOIRE existant (le cas d'une instance neuve, dont l'inventaire n'est pas encore genere — c'est ainsi que `make instancier` echouait sur le modele public), puis le premier nom par defaut. """ base = Path(instance) if instance is not None else instance_courante() for nom in ordre: if (base / "inventories" / nom / "hosts.yml").is_file(): return base / "inventories" / nom for nom in ordre: if (base / "inventories" / nom).is_dir(): return base / "inventories" / nom return base / "inventories" / ordre[0] def inventaire_de(instance: Path | None = None, ordre: tuple[str, ...] = ORDRE_INVENTAIRE) -> Path: """Le fichier `hosts.yml` de l'instance — `SETOPS_INVENTAIRE` le force (et est garde).""" base = Path(instance) if instance is not None else instance_courante() return inventaire_force(base) or (dossier_inventaire(base, ordre) / "hosts.yml") def plan_de(instance: Path | None = None) -> Path: """Le repertoire `plan/` de l'instance visee.""" base = Path(instance) if instance is not None else instance_courante() return base / "plan" def inventaire_force(instance: Path) -> Path | None: """Inventaire impose par SETOPS_INVENTAIRE, ou None. REFUSE s'il vise AILLEURS. POURQUOI CETTE GARDE (mesure du 2026-08-18). Deux variables designent la cible, et la seconde gagne : `SETOPS_INSTANCE` dit QUELLE instance, `SETOPS_INVENTAIRE` FORCE le fichier. Le Makefile exporte la seconde (ligne 13), derivee de l'instance ACTIVE. Tout code qui redirige la premiere vers une AUTRE instance sans retirer la seconde lit donc le plan de l'une et l'inventaire de l'autre. C'est arrive a P03, la preuve du diff-vide : sous `make prouver` elle generait le plan de chaque instance federee et le comparait a l'inventaire applique de la seule instance active. Elle annoncait « 17 hotes avec ecart » sur un lab qui n'en avait aucun — et, plus grave, l'inventaire applique des autres tenants n'etait JAMAIS lu, l'angle meme que cette preuve existe pour couvrir (D-77, tenant perime). Une contradiction entre les deux ne peut pas se deviner : on refuse, en nommant les deux valeurs. Le geste correct au site d'appel est `env.pop("SETOPS_INVENTAIRE")`. """ force = os.environ.get("SETOPS_INVENTAIRE") if not force: return None cible = Path(force) base = Path(instance).resolve() if base != cible.resolve() and base not in cible.resolve().parents: raise SystemExit( "REFUS : SETOPS_INVENTAIRE designe un inventaire HORS de l'instance demandee.\n" f" SETOPS_INSTANCE {base}\n" f" SETOPS_INVENTAIRE {cible.resolve()}\n" "Comparer le plan d'une instance a l'inventaire d'une AUTRE rend un verdict\n" "qui ne veut rien dire. Retirer SETOPS_INVENTAIRE de l'environnement fils\n" "(env.pop) partout ou l'on redirige SETOPS_INSTANCE." ) return cible def liens_acceptes(groupe: str, racine: Path | None = None) -> dict: """Liens qu'un role accepte : roles//meta/liens.yml -> setops_liens.accepte. Source unique partagee par le validateur, le GUI et scripts/instancier.py. Retourne {} si le role n'en declare aucun. """ fichier = (racine or RACINE_DEPOT) / "roles" / groupe / "meta" / "liens.yml" if not fichier.is_file(): return {} data = yaml.safe_load(fichier.read_text(encoding="utf-8")) or {} return ((data.get("setops_liens") or {}).get("accepte")) or {} def catalogue_liens(racine: Path | None = None) -> dict: """{groupe: {role_de_lien: {cible, injecte_sur, variables}}} pour tous les roles. Alimente la liste deroulante « rôle du lien » du GUI : on ne propose que ce que le role porteur declare accepter. """ base = (racine or RACINE_DEPOT) / "roles" if not base.is_dir(): return {} catalogue = {} for meta in sorted(base.glob("*/meta/liens.yml")): accepte = liens_acceptes(meta.parents[1].name, racine) if accepte: catalogue[meta.parents[1].name] = accepte return catalogue def integrations_universelles(racine: Path | None = None) -> dict[str, dict]: """{role: {raison, sauf_role}} — les integrations declarees universelles. Une integration universelle n'est PAS recopiee sur chaque serveur du plan : le role la declare une fois dans roles//meta/integration.yml, et tout hote la recoit. Le plan ne porte plus que les integrations reellement facultatives (client_backup, client_smtp, client_resolveur) et les exemptions. Pourquoi : recopiee 14 fois, la meme ligne finit par manquer une quinzieme — et un hote non supervise, non journalise ou sans certificat ne proteste pas. Le defaut est desormais « oui », et c'est l'exception qui doit se justifier. """ base = (racine or RACINE_DEPOT) / "roles" if not base.is_dir(): return {} politiques = {} for meta in sorted(base.glob("client_*/meta/integration.yml")): data = yaml.safe_load(meta.read_text(encoding="utf-8")) or {} politique = data.get("integration") or {} if politique.get("universelle"): politiques[meta.parents[1].name] = { "raison": str(politique.get("raison", "")).strip(), "sauf_role": politique.get("sauf_role") or None, } return politiques def integrations_de(srv: dict, services_hote: set[str] | None = None, universelles: dict[str, dict] | None = None, racine: Path | None = None, services_du_plan: set[str] | None = None) -> set[str]: """Integrations effectives d'un serveur : les universelles + celles du plan. `services_hote` = groupes de service que cet hote rend (derives des applications). Il sert aux exemptions : `sauf_role` retire l'integration a l'hote qui rend le service lui-meme — l'AC ne s'enrole pas aupres d'elle-meme. L'exemption suit donc le service, jamais un nom d'hote. `services_du_plan` = TOUS les groupes de service de cet ecosysteme. UNE INTEGRATION UNIVERSELLE N'EST POSEE QUE SI SON SERVICE CENTRAL EXISTE ICI (2026-08-22). Pourquoi cette regle a manque. « Tout hote est mesure » est vrai dans un ecosysteme qui porte un Prometheus. Dans un ecosysteme qui n'en a pas — un patient 0 minimal, une petite offre — la meme phrase pose sur chaque machine un client qui n'a personne a qui parler, et le controle de dependances refuse le deploiement. Le moteur supposait l'ecosysteme COMPLET ; c'est la troisieme fois que cette hypothese se paie, apres les intrants (P32) et les bases (P35). La regle est DERIVEE, pas declaree : le service central d'une integration est celui que le registre des dependances lui donne deja (`requiert_groupes_actifs`). Rien de neuf a tenir a jour, donc rien de neuf a oublier. """ if universelles is None: universelles = integrations_universelles(racine) rendus = services_hote or set() effectives = {role for role, p in universelles.items() if not (p["sauf_role"] and p["sauf_role"] in rendus)} if services_du_plan is not None: deps = dependances_groupes(racine) effectives = { role for role in effectives if all(exige in services_du_plan for exige in (deps.get(role, {}).get("requiert_groupes_actifs") or [])) } return effectives | set(srv.get("integrations") or []) def pool_de(nom_instance: str, index: int) -> str: """Nom du pool Proxmox d'un tenant. DERIVE, jamais saisi. Pourquoi le pool. Onze des quatorze serveurs portent le meme nom court chez deux tenants : c'est la preuve que la nomenclature est un vrai gabarit, et il ne faut pas la casser. Mais la console Proxmox affiche le NOM — deux `infra-pki-01` y sont indiscernables a l'oeil, alors que tout le reste (IP, VMID, VLAN, FQDN) differe. Le pool restitue l'appartenance sans rien renommer. Pourquoi DERIVE malgre les pools existants. Le cluster porte deja une convention `Env.Tenant` (`Prod.Chezlepro`, `Lab.KBR`...) qui regroupe des VM anterieures a Set-OPS — ZIMBRA, erplibre, eregion... Ces pools-la sont l'ANCIEN MONDE : on n'y touche pas, et on n'y verse pas la flotte generee. Melanger les deux effacerait justement la frontiere que Set-OPS etablit. La flotte derivee a donc ses propres pools, nommes comme tout le reste : depuis le seed. LE NOM DU POOL EST CELUI DU DEPOT (2026-09-12, decision de l'exploitant). C'etait `Chezlepro-17` — le seed lu dans le nom. Juste, mais il obligeait a connaitre le codage pour relier un pool a son depot, et il CHANGEAIT si l'index changeait : la renumerotation du site l'a montre le jour meme. Un pool porte desormais le nom du depot, tel quel : `OPS-Chezlepro`, `OPS-Technolibre`. Ce qu'on lit dans la console est ce qu'on tape dans un terminal, et un renumerotage ne renomme plus rien. L'unicite ne vient plus de l'index mais du nom de dossier — qui est deja unique par construction, deux depots freres ne pouvant porter le meme. `index` reste au parametre : il ne sert plus au nom, il sert a refuser un appel incoherent. Les pools `Env.Tenant` anterieurs a Set-OPS (`Prod.Chezlepro`, `Lab.KBR`...) restent l'ANCIEN MONDE : on n'y touche pas, et on n'y verse pas la flotte generee. """ valider_index(index) base = re.sub(r"[^A-Za-z0-9._-]", "", nom_instance) or "tenant" return base # LE POOL DU SITE NE DERIVE DE RIEN, ET C'EST LE POINT (2026-09-12). Les machines du # genome — celles qui portent le cache, la forge, l'AC, les noms et le depot — ne # dependent d'aucun index : elles sont l'infrastructure SUR laquelle les index vivent. # Un nom invariable dit exactement cela, et il reste le meme d'un hebergeur a l'autre. # # A ne pas confondre avec le pool `Set-OPS` deja present sur le cluster de reference, # qui appartient a l'ancien monde. POOL_SITE = "Site-OPS" def est_groupe_operationnel(groupe: str) -> bool: return groupe.startswith(GROUPES_OPERATIONNELS_PREFIXES) def groupes_operationnels_connus(enfants: dict, dossier_playbooks: Path) -> list[str]: groupes = set(enfants) groupes.update(playbook.stem for playbook in dossier_playbooks.glob("*.yml")) return sorted(groupe for groupe in groupes if est_groupe_operationnel(groupe)) def charger_bases_donnees(path: Path | None) -> dict: """Charge le registre des bases de donnees (serveurs_bd + bases_donnees).""" if not path or not path.exists(): return {"serveurs_bd": {}, "bases_donnees": {}} with path.open("r", encoding="utf-8") as fichier: data = yaml.safe_load(fichier) or {} if not isinstance(data, dict): raise ValueError(f"{path} ne contient pas une table YAML.") data.setdefault("serveurs_bd", {}) data.setdefault("bases_donnees", {}) return data PORTEES_BD = {"groupe", "hote", "application"} def valider_bases(registre: dict, applications: dict | None = None) -> None: """Valide la coherence du registre des bases de donnees. Si `applications` (le registre des applications) est fourni, les bases de portee 'application' sont verifiees : leur consommateur doit etre une application declaree. """ serveurs = registre.get("serveurs_bd") or {} bases = registre.get("bases_donnees") or {} apps = (applications or {}).get("applications") or {} if not isinstance(serveurs, dict) or not isinstance(bases, dict): raise ValueError("serveurs_bd et bases_donnees doivent etre des tables.") for nom, srv in serveurs.items(): if not isinstance(srv, dict): raise ValueError(f"Serveur BD '{nom}': table attendue.") for champ in ("type", "hote", "port"): if not str(srv.get(champ, "")).strip(): raise ValueError(f"Serveur BD '{nom}': champ '{champ}' requis.") noms_bases: set[str] = set() for cle, base in bases.items(): if not isinstance(base, dict): raise ValueError(f"Base '{cle}': table attendue.") for champ in ("serveur", "base", "proprietaire", "secret", "consommateur"): if not str(base.get(champ, "")).strip(): raise ValueError(f"Base '{cle}': champ '{champ}' requis.") if base["serveur"] not in serveurs: raise ValueError(f"Base '{cle}': serveur '{base['serveur']}' inconnu.") portee = str(base.get("portee", "groupe")).strip() if portee not in PORTEES_BD: attendu = ", ".join(sorted(PORTEES_BD)) raise ValueError(f"Base '{cle}': portee '{portee}' inconnue (attendu: {attendu}).") if portee == "groupe" and not est_groupe_operationnel(base["consommateur"]): raise ValueError(f"Base '{cle}': portee groupe mais consommateur '{base['consommateur']}' non operationnel.") if portee == "application" and apps and base["consommateur"] not in apps: raise ValueError(f"Base '{cle}': application consommatrice '{base['consommateur']}' inconnue (instance/plan/applications.yml).") couple = f"{base['serveur']}/{base['base']}" if couple in noms_bases: raise ValueError(f"Base en double sur un meme serveur: {couple}") noms_bases.add(couple) def _base_resolue(cle: str, base: dict, serveurs: dict) -> dict: return {"cle": cle, "base": base, "serveur": serveurs.get(base.get("serveur"), {})} def bases_du_groupe(registre: dict, groupe: str) -> list[dict]: """Bases de portee 'groupe' consommees par un groupe applicatif.""" serveurs = registre.get("serveurs_bd") or {} resultat = [] for cle, base in (registre.get("bases_donnees") or {}).items(): if str(base.get("portee", "groupe")) == "groupe" and base.get("consommateur") == groupe: resultat.append(_base_resolue(cle, base, serveurs)) return resultat def bases_de_application(registre: dict, application: str | None = None, groupe: str | None = None, hote: str | None = None) -> list[dict]: """Bases vues par une application instanciee, toutes portees confondues. Une application `A` (du groupe `G`, sur l'hote `H`) recoit les bases ou : (portee=application ET consommateur=A) OU (portee=groupe ET consommateur=G) OU (portee=hote ET consommateur=H). """ serveurs = registre.get("serveurs_bd") or {} resultat = [] for cle, base in (registre.get("bases_donnees") or {}).items(): portee = str(base.get("portee", "groupe")) conso = base.get("consommateur") if ((portee == "application" and application is not None and conso == application) or (portee == "groupe" and groupe is not None and conso == groupe) or (portee == "hote" and hote is not None and conso == hote)): resultat.append(_base_resolue(cle, base, serveurs)) return resultat def charger_applications(path: Path | None) -> dict: """Charge le registre des applications (applications -> groupe, hote).""" if not path or not path.exists(): return {"applications": {}} with path.open("r", encoding="utf-8") as fichier: data = yaml.safe_load(fichier) or {} if not isinstance(data, dict): raise ValueError(f"{path} ne contient pas une table YAML.") data.setdefault("applications", {}) return data def valider_applications(registre: dict, domaines: dict | None = None, serveurs: dict | None = None) -> None: """Valide la coherence du registre des applications. Si `domaines` (le registre des domaines publics) est fourni, chaque FQDN de `expose` doit avoir un domaine parent declare. Si `serveurs` (le registre des serveurs) est fourni, l'hote porteur de chaque application doit y etre declare — c'est le garde-fou contre l'hote FANTOME (une application posee sur une VM qui n'existe pas). Sans lui, le plan passe la validation et casse au deploiement. """ apps = registre.get("applications") or {} doms = (domaines or {}).get("domaines_publics") or {} if not isinstance(apps, dict): raise ValueError("applications doit etre une table.") for nom, app in apps.items(): if not isinstance(app, dict): raise ValueError(f"Application '{nom}': table attendue.") for champ in ("groupe", "hote"): if not str(app.get(champ, "")).strip(): raise ValueError(f"Application '{nom}': champ '{champ}' requis.") hotes_connus = (serveurs or {}).get("serveurs") if hotes_connus and app["hote"] not in hotes_connus: raise ValueError( f"Application '{nom}': hote '{app['hote']}' absent du registre des " f"serveurs (hote fantome). Declare-le dans plan/serveurs.yml, ou " f"deplace l'application sur un hote existant.") if not est_groupe_operationnel(app["groupe"]): raise ValueError(f"Application '{nom}': groupe '{app['groupe']}' non operationnel.") requiert = app.get("requiert") or [] if not isinstance(requiert, list): raise ValueError(f"Application '{nom}': 'requiert' doit etre une liste.") for dep in requiert: if dep not in apps: raise ValueError(f"Application '{nom}': requiert '{dep}' inconnue.") expose = app.get("expose") or [] if not isinstance(expose, list): raise ValueError(f"Application '{nom}': 'expose' doit etre une liste.") for fqdn in expose: if doms and domaine_parent(fqdn, doms) is None: raise ValueError(f"Application '{nom}': FQDN expose '{fqdn}' sans domaine parent declare.") port = app.get("port") if port is not None and not str(port).strip().isdigit(): raise ValueError(f"Application '{nom}': 'port' doit etre un entier.") liens = app.get("liens") or [] if not isinstance(liens, list): raise ValueError(f"Application '{nom}': 'liens' doit etre une liste.") accepte = None for lien in liens: if not isinstance(lien, dict): raise ValueError(f"Application '{nom}': chaque lien doit etre une table {{vers, role}}.") vers = str(lien.get("vers", "")).strip() role = str(lien.get("role", "")).strip() if not vers or not role: raise ValueError(f"Application '{nom}': lien incomplet (champs 'vers' et 'role' requis).") if vers == nom: raise ValueError(f"Application '{nom}': lien vers elle-meme.") if vers not in apps: raise ValueError(f"Application '{nom}': cible de lien '{vers}' inconnue.") if accepte is None: accepte = liens_acceptes(str(app.get("groupe", ""))) # Un role sans meta/liens.yml n'accepte aucun lien : on ne bloque que # s'il en declare (sinon instancier.py tranchera avec le meme message). if accepte and role not in accepte: raise ValueError( f"Application '{nom}': role de lien '{role}' non accepte par " f"'{app.get('groupe')}' (roles/{app.get('groupe')}/meta/liens.yml). " f"Acceptes : {', '.join(sorted(accepte))}.") def applications_de_hote(registre: dict, hote: str) -> list[dict]: """Applications instanciees sur un hote donne (plusieurs possibles).""" return [ {"id": nom, **app} for nom, app in (registre.get("applications") or {}).items() if app.get("hote") == hote ] def domaine_parent(fqdn: str, domaines_publics: dict) -> str | None: """Domaine declare le plus long dont `fqdn` releve (apex inclus).""" fqdn = str(fqdn).strip() candidats = [d for d in domaines_publics if fqdn == d or fqdn.endswith("." + d)] return max(candidats, key=len) if candidats else None def expositions_des_applications(applications: dict, domaines: dict, edge: str | None = None) -> list[dict]: """Expositions publiques derivees des applications (champ 'expose'). Pour chaque FQDN qu'une application expose, on resout son domaine parent et l'edge de ce domaine. Si `edge` est fourni, ne retient que les expositions de cet edge. Le rendu de l'amont (IP interne) reste a la charge de l'appelant. """ apps = (applications or {}).get("applications") or {} doms = (domaines or {}).get("domaines_publics") or {} resultat = [] for app_id, app in apps.items(): for fqdn in (app.get("expose") or []): dom = domaine_parent(fqdn, doms) conf = doms.get(dom, {}) if dom else {} # SANS EDGE DECLARE, LE SERVICE SE SERT LUI-MEME (2026-08-25). # # `edge` se resout par le DOMAINE : chez un tenant, `domaines.yml` dit quel # nginx front telle zone, et tout FQDN de cette zone y aboutit. Le SITE n'a # pas d'edge — chaque service ecoute lui-meme, avec son propre certificat. # # Sans ce repli, la derivation rendait `edge: None` : le gabarit de # `hosts_statiques` cherchait alors `groups[None]`, ne trouvait rien, et # ecrivait un plancher SANS alias. Le certificat de la forge portait # `forge.genese.internal`, et ce nom ne resolvait nulle part — du TLS correct # sur un nom que personne ne pouvait appeler. edge_dom = conf.get("edge") or app.get("groupe") if edge is not None and edge_dom != edge: continue # SANS DOMAINE PUBLIC DECLARE, LE DOMAINE EST CELUI QUE PORTE LE FQDN # (2026-08-25). Meme repli que pour `edge` juste au-dessus, et il manquait. # # `domaine_parent` cherche le FQDN parmi les domaines PUBLICS. Le SITE n'en a # aucun — tous ses noms sont internes — donc `dom` valait None, et le modele # de zone PowerDNS, qui ne retient que les expositions dont le domaine EST sa # zone, les ecartait toutes. # # Consequence mesuree le 2026-08-25 : `forge.genese.internal` et # `pki.genese.internal` n'avaient aucun enregistrement A. Seul le plancher # /etc/hosts savait les resoudre — or un service ne doit pas dependre d'un # plancher pour etre joignable ; le plancher est un filet, pas le sol. if not dom and "." in str(fqdn): dom = str(fqdn).split(".", 1)[1] resultat.append({ "fqdn": fqdn, "application": app_id, "groupe": app.get("groupe"), "hote": app.get("hote"), "port": app.get("port"), "edge": edge_dom, "domaine": dom, # Exposition WebSocket-aware (ex. Collabora) : le vhost ajoute # alors l'upgrade Upgrade/Connection. Declare par l'app (websocket: true). "websocket": bool(app.get("websocket")), }) return resultat def zones_inverses(adresses, supernet: str = "") -> list[str]: """Les zones `in-addr.arpa` que cet ecosysteme revendique — et rien de plus. Un TENANT tire un supernet /16 de son index : une seule zone suffit a couvrir tous ses sous-reseaux, et c'est celle qu'on rend. Un SITE ne derive pas — il declare plusieurs /24 et n'a pas de supernet unique. On rend alors une zone PAR /24 reellement occupe. Revendiquer `0.10.in-addr.arpa` pour le site aurait ete plus simple et faux : cette zone couvre aussi la frontiere, le transit et les hyperviseurs, qui ne sont pas a lui. Une autorite qu'on s'attribue sans l'exercer est une panne differee — le resolveur repondrait NXDOMAIN pour des adresses qu'un autre sait nommer. DEUX APPELANTS, UNE SEULE DERIVATION : `serveur_powerdns` ecrit ces zones, `serveur_resolveur` les delegue a l'autoritatif. Deux calculs separes finiraient par diverger, et la divergence ne se verrait qu'au premier PTR interroge. """ texte = str(supernet or "") if texte.endswith("/16") and texte.count(".") == 3: a, b = texte.split(".")[0], texte.split(".")[1] return [f"{b}.{a}.in-addr.arpa"] vues: list[str] = [] for adresse in (adresses or []): bouts = str(adresse or "").split(".") if len(bouts) != 4 or not all(x.isdigit() for x in bouts): continue zone = f"{bouts[2]}.{bouts[1]}.{bouts[0]}.in-addr.arpa" if zone not in vues: vues.append(zone) return sorted(vues) def chaine_connexion(entree: dict, serveur: dict, secret: str = "***") -> str: """Derive une chaine de connexion (mot de passe masque par defaut).""" type_bd = (serveur or {}).get("type", "postgres") hote = (serveur or {}).get("hote", "?") port = (serveur or {}).get("port", "?") return f"{type_bd}://{entree.get('proprietaire', '?')}:{secret}@{hote}:{port}/{entree.get('base', '?')}" AUTORITES_DNS = {"primaire-cache", "auto-heberge", "delegue"} # LES ENREGISTREMENTS QU'UNE ZONE PUBLIQUE PEUT DECLARER (2026-09-16). # # Avant, une zone publique ne portait que ses expositions web : des A vers l'adresse # publique du site. Or la zone d'un locataire porte aussi son COURRIEL (MX, SPF, DMARC), # ses verifications et ses noms historiques. Basculer les serveurs de noms sur une zone # qui les ignore couperait le courriel de production a la minute ou le registraire suit. # # La liste est courte a dessein : chaque type ajoute est un type rendu, et donc un type a # eprouver. SRV, NS et SOA n'y sont pas — le SOA et les NS sont ceux du site. TYPES_ENREGISTREMENT = {"A", "AAAA", "CNAME", "MX", "TXT", "CAA"} ETIQUETTES_CAA = {"issue", "issuewild", "iodef"} _NOM_HOTE = re.compile(r"^(?=.{1,253}$)([A-Za-z0-9_]([A-Za-z0-9_-]{0,61}[A-Za-z0-9])?)(\.[A-Za-z0-9_]([A-Za-z0-9_-]{0,61}[A-Za-z0-9])?)*$") def _valider_enregistrement(nom_domaine: str, entree) -> str: """Valide un enregistrement declare ; rend son FQDN. Leve ValueError sinon.""" if not isinstance(entree, dict): raise ValueError(f"Domaine '{nom_domaine}': enregistrement invalide (table attendue).") nom = str(entree.get("nom", "")).strip() rtype = str(entree.get("type", "")).strip().upper() valeur = entree.get("valeur") lieu = f"Domaine '{nom_domaine}', enregistrement {nom or '?'} {rtype or '?'}" if not nom: raise ValueError(f"{lieu}: champ 'nom' requis (`@` pour la zone elle-meme).") if nom != "@" and (nom.endswith(".") or not _NOM_HOTE.match(nom)): raise ValueError(f"{lieu}: nom relatif attendu, sans point final (ex. `mx`, `_dmarc`).") # `mx.chezlepro.ca` ecrit sous `chezlepro.ca` deviendrait `mx.chezlepro.ca.chezlepro.ca`. if nom.lower() == nom_domaine.lower() or nom.lower().endswith("." + nom_domaine.lower()): raise ValueError(f"{lieu}: nom relatif a la zone attendu (`mx`, pas `mx.{nom_domaine}`).") if rtype not in TYPES_ENREGISTREMENT: raise ValueError(f"{lieu}: type inconnu (attendu: {', '.join(sorted(TYPES_ENREGISTREMENT))}).") if valeur is None or not str(valeur).strip(): raise ValueError(f"{lieu}: champ 'valeur' requis.") valeur = str(valeur).strip() ttl = entree.get("ttl") if ttl is not None and (not isinstance(ttl, int) or isinstance(ttl, bool) or not 60 <= ttl <= 604800): raise ValueError(f"{lieu}: 'ttl' doit etre un entier entre 60 et 604800.") if rtype in ("A", "AAAA"): # UNE ZONE PUBLIQUE NE PORTE QUE DES ADRESSES PUBLIQUES : une adresse privee ne # resout pour personne dehors, et renseigne sur le reseau dedans. try: ip = ipaddress.ip_address(valeur) except ValueError: raise ValueError(f"{lieu}: '{valeur}' n'est pas une adresse IP.") from None if (rtype == "A") != (ip.version == 4): raise ValueError(f"{lieu}: '{valeur}' n'est pas une adresse IPv{4 if rtype == 'A' else 6}.") if not ip.is_global: raise ValueError(f"{lieu}: '{valeur}' n'est pas une adresse publique.") elif rtype in ("CNAME", "MX"): # UNE CIBLE EST TOUJOURS ABSOLUE. Dans un fichier de zone, `mx.chezlepro.ca` sans # point final devient `mx.chezlepro.ca.chezlepro.ca.` ; le gabarit ajoute le point. if not _NOM_HOTE.match(valeur.rstrip(".")): raise ValueError(f"{lieu}: '{valeur}' n'est pas un nom d'hote complet.") if rtype == "CNAME" and nom == "@": raise ValueError(f"{lieu}: un CNAME a l'apex est interdit (le SOA et les NS y vivent).") if rtype == "MX": prio = entree.get("priorite") if not isinstance(prio, int) or isinstance(prio, bool) or not 0 <= prio <= 65535: raise ValueError(f"{lieu}: 'priorite' entiere (0-65535) requise pour un MX.") elif rtype in ("TXT", "CAA"): # Le gabarit met la valeur entre guillemets et la coupe en tranches de 255 : un # guillemet ou une barre oblique inverse la casserait sans que rien ne le dise. # Et en ASCII : les tranches se comptent en octets sur le fil, en caracteres ici. if not valeur.isascii(): raise ValueError(f"{lieu}: 'valeur' doit etre en ASCII.") if '"' in valeur or "\\" in valeur: raise ValueError(f"{lieu}: guillemet ou barre oblique inverse refuses dans 'valeur'.") if rtype == "CAA" and str(entree.get("etiquette", "")).strip() not in ETIQUETTES_CAA: raise ValueError(f"{lieu}: 'etiquette' requise pour un CAA ({', '.join(sorted(ETIQUETTES_CAA))}).") return fqdn_exposition(nom_domaine, nom) def charger_domaines(path: Path | None) -> dict: """Charge le registre des domaines publics (domaines_publics).""" if not path or not path.exists(): return {"domaines_publics": {}} with path.open("r", encoding="utf-8") as fichier: data = yaml.safe_load(fichier) or {} if not isinstance(data, dict): raise ValueError(f"{path} ne contient pas une table YAML.") data.setdefault("domaines_publics", {}) return data def fqdn_exposition(domaine: str, nom: str) -> str: """FQDN publie pour une etiquette ('@' ou le domaine lui-meme = apex).""" nom = str(nom).strip() if nom in ("@", "", domaine): return domaine return f"{nom}.{domaine}" def valider_domaines(registre: dict) -> None: """Valide la coherence du registre des domaines publics.""" domaines = registre.get("domaines_publics") or {} if not isinstance(domaines, dict): raise ValueError("domaines_publics doit etre une table.") fqdns: set[str] = set() for nom_domaine, conf in domaines.items(): if not isinstance(conf, dict): raise ValueError(f"Domaine '{nom_domaine}': table attendue.") autorite = str(conf.get("autorite", "")).strip() if autorite not in AUTORITES_DNS: attendu = ", ".join(sorted(AUTORITES_DNS)) raise ValueError(f"Domaine '{nom_domaine}': autorite '{autorite}' inconnue (attendu: {attendu}).") if not str(conf.get("edge", "")).strip(): raise ValueError(f"Domaine '{nom_domaine}': champ 'edge' requis.") secondaires = conf.get("secondaires") or [] if not isinstance(secondaires, list): raise ValueError(f"Domaine '{nom_domaine}': 'secondaires' doit etre une liste.") exposition = conf.get("exposition") or [] if not isinstance(exposition, list): raise ValueError(f"Domaine '{nom_domaine}': 'exposition' doit etre une liste.") for entree in exposition: if not isinstance(entree, dict): raise ValueError(f"Domaine '{nom_domaine}': entree d'exposition invalide (table attendue).") for champ in ("nom", "cible"): if not str(entree.get(champ, "")).strip(): raise ValueError(f"Domaine '{nom_domaine}': exposition, champ '{champ}' requis.") fqdn = fqdn_exposition(nom_domaine, entree["nom"]) if fqdn in fqdns: raise ValueError(f"Exposition en double: {fqdn}") fqdns.add(fqdn) # DNSSEC NE SIGNE QUE CE QUE SET-OPS ECRIT. Ailleurs, `dnssec: true` ne ferait rien — # et laisserait croire a une zone signee. if conf.get("dnssec") is True and autorite != "primaire-cache": raise ValueError( f"Domaine '{nom_domaine}': `dnssec: true` n'est applique qu'a une zone " f"`primaire-cache` (autorite actuelle: {autorite}).") enregistrements = conf.get("enregistrements") or [] if not isinstance(enregistrements, list): raise ValueError(f"Domaine '{nom_domaine}': 'enregistrements' doit etre une liste.") # UN ENREGISTREMENT QUE RIEN NE REND EST UNE PROMESSE SILENCIEUSE. Seules les zones # `primaire-cache` sont ecrites par Set-OPS ; ailleurs, la declaration n'aurait aucun # effet et laisserait croire le contraire. if enregistrements and autorite != "primaire-cache": raise ValueError( f"Domaine '{nom_domaine}': 'enregistrements' n'est rendu que pour une zone " f"`primaire-cache` (autorite actuelle: {autorite}).") par_nom: dict[str, list] = {} for entree in enregistrements: fqdn = _valider_enregistrement(nom_domaine, entree) par_nom.setdefault(fqdn, []).append(str(entree.get("type")).strip().upper()) for fqdn, liste in sorted(par_nom.items()): types = set(liste) # UN CNAME EST SEUL A SON NOM (RFC 1034) : un resolveur qui le suit ignore le reste. if "CNAME" in types and (len(liste) > 1 or fqdn in fqdns): raise ValueError(f"Domaine '{nom_domaine}': {fqdn} porte un CNAME et autre chose.") if fqdn in fqdns and types & {"A", "AAAA"}: raise ValueError(f"Domaine '{nom_domaine}': {fqdn} est deja une exposition (A rendu par Set-OPS).") def expositions_du_groupe(registre: dict, groupe: str) -> list[dict]: """Expositions publiques dont la cible est un groupe interne donne.""" resultat = [] for nom_domaine, conf in (registre.get("domaines_publics") or {}).items(): for entree in (conf.get("exposition") or []): if entree.get("cible") == groupe: resultat.append({ "domaine": nom_domaine, "fqdn": fqdn_exposition(nom_domaine, entree.get("nom", "")), "type": entree.get("type", "web"), "edge": conf.get("edge"), }) return resultat def charger_nomenclature(path: Path | None) -> dict: """Charge le registre de nomenclature (fonctions, categories, adressage).""" if not path or not path.exists(): return {} with path.open("r", encoding="utf-8") as fichier: data = yaml.safe_load(fichier) or {} if not isinstance(data, dict): raise ValueError(f"{path} ne contient pas une table YAML.") return data ETATS_SERVEUR = {"actif", "planifie"} def fonction_seq(nom: str) -> tuple[str, int | None]: """Decompose un hostname 'fonction-NN' en (fonction, NN).""" m = re.match(r"^(.+)-(\d+)$", str(nom or "")) if m: return m.group(1), int(m.group(2)) return str(nom or ""), None # --- Derivation de l'ADRESSAGE depuis le seul seed `index` -------------------- # # Principe : TOUT l'adressage (supernet, sous-reseaux, passerelles, VLAN, VMID) se # DERIVE de l'index d'instance + du numero de zone. Rien n'est ecrit a la main dans # la nomenclature (elle ne garde que le MODELE : libelles de zones + placement des # fonctions). Source unique, ci-dessous, partagee par la generation d'inventaire et # le devis reseau (scripts/devis_reseau.py). # # Conventions du modele a 6 zones (reference integral) : # - 2e octet du supernet = index TEL QUEL (index 1 -> 10.1.0.0/16) # - 3e octet de la zone = OCTET_ZONE + categorie (zone 1 -> .16, dans .16.0/20) # - VLAN sur le trunk convergE = SOCLE_VLAN + index x 10 + zone (4 chiffres, unique) # - VMID ip-miroir = VLAN(4) . octet-hote(3) . sequence(2) = 9 chiffres mnemotechniques # # LE DECALAGE DE +10 A ETE RETIRE le 2026-08-12. Il produisait `10.(10+index)`, et # personne ne savait plus pourquoi : ni le commentaire de la constante, ni le wiki, ni # le commit fondateur (36a882b) ne le justifiaient. Ses deux effets constates : # - il reservait 10.0-10.9 sous la plage tenant — utile tant que l'underlay vivait # LA ; D-77 l'a fait entrer dans la bande basse de son propre /16, ce qui a vide # cette reserve de son role ; # - il eloignait le premier tenant de 10.0.0.0/16, la plage la plus repandue en # reseau domestique. Ce risque revient donc aux INDEX BAS, et c'est assume : # choisir un index, c'est choisir sa plage — autant que ce soit lisible. # En echange, l'index se lit DIRECTEMENT dans l'adresse (index 17 -> 10.17.x.x) et la # plage utile passe de 0-245 a 0-255. OCTET_ZONE = 15 SOCLE_VLAN = 1000 # BORNES DE L'INDEX. Le 2e octet d'une IPv4 vaut 0-255 : au-dela, la derivation ne # produit plus une adresse mais une CHAINE qui y ressemble — `10.300.0.0/16` passe # silencieusement tant que personne ne la donne a `ip_network()`, et l'erreur remonte # alors tres loin de sa cause. Depuis le retrait du decalage de +10 (2026-08-12), la # plage utile est exactement celle de l'octet. INDEX_MIN, INDEX_MAX = 0, 255 INDEX_BAC_A_SABLE = 9 # index par defaut d'une nomenclature sans seed (jetable) def valider_index(index) -> int: """L'index, verifie et rendu en entier. Leve `ValueError` hors bornes. Pose a la SOURCE de la derivation plutot que dans un validateur de plan : toutes les fonctions derivees passent par ici, donc aucune ne peut fabriquer une adresse invalide, d'ou qu'on l'appelle — plan, GUI, devis ou test. """ try: i = int(index) except (TypeError, ValueError, OverflowError): # OverflowError : `int(float('inf'))`. Trouve par le test, pas par relecture — # un infini flottant n'est ni un entier valide ni une TypeError. raise ValueError(f"index invalide : {index!r} n'est pas un entier") from None if not (INDEX_MIN <= i <= INDEX_MAX): raise ValueError( f"index {i} hors bornes [{INDEX_MIN}-{INDEX_MAX}] : le 2e octet d'une IPv4 " f"ne peut pas le porter. La derivation produirait « 10.{i}.0.0/16 », qui " f"ressemble a un reseau sans en etre un.") return i def supernet_de(index: int) -> str: """/16 du tenant : 10..0.0/16 — l'index se lit dans l'adresse.""" return f"10.{valider_index(index)}.0.0/16" def base3_de(index: int, categorie: int) -> str: """Trois premiers octets d'une zone : 10..(15+categorie).""" return f"10.{valider_index(index)}.{OCTET_ZONE + int(categorie)}" def sous_reseau_de(index: int, categorie: int, cidr: int = 24) -> str: """/24 d'une zone : 10..(15+categorie).0/24.""" return f"{base3_de(index, categorie)}.0/{int(cidr)}" def passerelle_de(index: int, categorie: int) -> str: """Passerelle d'une zone (SVI) : premier hote du /24.""" return f"{base3_de(index, categorie)}.1" def vlan_de(index: int, zone: int) -> int: """VLAN 802.1Q sur le trunk convergE : 1000 + index x 10 + zone (unique globalement). La meme borne y protege un second plafond : a l'index 255 le VLAN vaut 3550+zone, sous les 4094 du 802.1Q. Sans elle, un index de trois chiffres deborderait aussi la. """ return SOCLE_VLAN + valider_index(index) * 10 + int(zone) def deriver_nomenclature(fonction: str, seq: int, nomenclature: dict) -> dict | None: """Derive VMID / VLAN / IP / passerelle / CIDR depuis la fonction et le rang. Source unique de la derivation (miroir Python de l'auto-proposition du GUI). Tout vient de l'index + de la zone : aucune lecture d'adressage stocke. """ fonctions = nomenclature.get("fonctions") or {} categories = nomenclature.get("categories") or {} f = fonctions.get(fonction) if not f or seq is None: return None cat = f.get("categorie") svc = f.get("service") c = categories.get(cat) or categories.get(str(cat)) if not c: return None index = nomenclature.get("index", INDEX_BAC_A_SABLE) hote = svc * 10 + int(seq) vlan = vlan_de(index, int(cat)) return { "vmid": f"{vlan}{hote:03d}{int(seq):02d}", "vlan": vlan, "adresse_ip": f"{base3_de(index, int(cat))}.{hote}", "passerelle": passerelle_de(index, int(cat)), "cidr": nomenclature.get("cidr_hote", 24), # La ZONE de securite, exposee : c'est d'elle que se derive le VNet du SDN # (`t17serv`). La recalculer ailleurs demanderait de relire fonctions + # categories — deux lectures pour une valeur deja connue ici. "zone": int(cat), "zone_libelle": str(c.get("libelle", f"zone{cat}")), } def charger_serveurs(path: Path | None) -> dict: """Charge le registre des serveurs (VM) du plan.""" if not path or not path.exists(): return {"serveurs": {}} with path.open("r", encoding="utf-8") as fichier: data = yaml.safe_load(fichier) or {} if not isinstance(data, dict): raise ValueError(f"{path} ne contient pas une table YAML.") data.setdefault("serveurs", {}) return data def valider_serveurs(registre: dict, nomenclature: dict | None = None) -> None: """Valide la coherence du registre des serveurs.""" serveurs = registre.get("serveurs") or {} fonctions = (nomenclature or {}).get("fonctions") or {} if not isinstance(serveurs, dict): raise ValueError("serveurs doit etre une table.") universelles = integrations_universelles() for nom, srv in serveurs.items(): if not isinstance(srv, dict): raise ValueError(f"Serveur '{nom}': table attendue.") fonction = str(srv.get("fonction", "")).strip() if not fonction: raise ValueError(f"Serveur '{nom}': champ 'fonction' requis.") if fonctions and fonction not in fonctions: raise ValueError(f"Serveur '{nom}': fonction '{fonction}' inconnue de la nomenclature.") etat = str(srv.get("etat", "")).strip() if etat and etat not in ETATS_SERVEUR: raise ValueError(f"Serveur '{nom}': etat '{etat}' inconnu (attendu: {', '.join(sorted(ETATS_SERVEUR))}).") integrations = srv.get("integrations") if integrations is not None and not isinstance(integrations, list): raise ValueError(f"Serveur '{nom}': 'integrations' doit etre une liste.") # Une integration universelle recopiee dans le plan a deux torts : elle # laisse croire que son ABSENCE chez un voisin signifie « non integre », # et elle offre une valeur a oublier. La politique du role est la seule # source ; on refuse la copie plutot que de la tolerer silencieusement. redites = sorted(set(integrations or []) & set(universelles)) if redites: raise ValueError( f"Serveur '{nom}': integration(s) {', '.join(redites)} universelle(s) — " f"declaree(s) dans roles//meta/integration.yml, elles s'appliquent " f"a tout hote. Retirer ces lignes du plan.") def reconcilier_serveur(nom: str, srv: dict, hote_inventaire: dict | None, nomenclature: dict) -> dict: """Compare les valeurs derivees (fonction+rang) aux valeurs de l'inventaire. Renvoie {derive, inventaire, divergences} ; divergences vide = reconcilie. """ _, seq = fonction_seq(nom) derive = deriver_nomenclature(str(srv.get("fonction", "")), seq, nomenclature) or {} inv = hote_inventaire or {} champs = {"vmid": "vmid", "adresse_ip": "adresse_ip", "vlan": "vlan", "passerelle": "passerelle"} divergences = [] for cle_derivee, cle_inv in champs.items(): if cle_derivee not in derive: continue attendu = str(derive[cle_derivee]) present = str(inv.get(cle_inv, "")).strip() if present and present != attendu: divergences.append(f"{cle_derivee}: inventaire={present} != derive={attendu}") return {"derive": derive, "inventaire": inv, "divergences": divergences, "absent_inventaire": hote_inventaire is None} # --- Dimensionnement derive des ressources VM ------------------------------- # ressources(VM) = socle_SE + somme(empreinte des logiciels) -> marge -> arrondi. # Les empreintes vivent dans roles//meta/empreinte.yml (propriete du # logiciel). Les groupes sans role dedie ont un repli ci-dessous. Le plan peut # tout surcharger par hote (serveurs.yml: coeurs/memoire/disque). SOCLE_SE = {"coeurs": 1, "memoire_mo": 512, "disque_go": 8} EMPREINTE_DEFAUT = {"coeurs": 1, "memoire_mo": 512, "disque_go": 5} EMPREINTES_SANS_ROLE = { "serveur_web_frontal": {"coeurs": 1, "memoire_mo": 512, "disque_go": 5}, "serveur_web_dorsal": {"coeurs": 1, "memoire_mo": 1024, "disque_go": 10}, } MARGE_MEMOIRE = 1.20 MARGE_DISQUE = 1.25 PALIER_MEMOIRE_MO = 512 MIN_MEMOIRE_MO = 1024 PALIER_DISQUE_GO = 5 MIN_DISQUE_GO = 12 MIN_COEURS = 1 MAX_COEURS = 8 CHAMPS_EMPREINTE = ("coeurs", "memoire_mo", "disque_go") # --- LE PLANCHER DE MEMOIRE (2026-09-13) --------------------------------------------- # # `balloon` valait 0 sur toute la flotte : ballooning DESACTIVE, donc l'hote ne pouvait # jamais reprendre une page meme jamais touchee. Mesure : 40 Go alloues a un ecosysteme # pour 15,5 Go lus — 39 %. # # LE BALLOONING NE COMPRIME RIEN. L'invite voit `memory` au demarrage et s'y dimensionne. # Reprendre sous ce qu'il a reserve le fait swapper A L'INTERIEUR, ce qui coute plus que # la RAM rendue. Le plancher depend donc de CE QUE LA MACHINE PORTE, pas d'un ratio unique. # # TROIS FAMILLES, et la frontiere entre elles est « est-ce que ca se dimensionne au # demarrage a partir de la RAM totale ? » : # # 1,00 ne rend rien — la machine reserve tout et s'en sert # 0,75 rend peu — elle reserve au demarrage mais laisse du mou # 0,50 rend la moitie — service au repos l'essentiel du temps # # Un plancher ECRIT AU PLAN (`memoire_min` dans `gabarit:`) prime toujours : cette table # est un defaut raisonne, pas une contrainte. # LA PREMIERE VERSION DE CETTE TABLE ETAIT ECRITE SUR UNE CRAINTE, PAS SUR UNE MESURE. # Elle donnait 0,75 a PostgreSQL et a Keycloak « parce qu'ils se dimensionnent au # demarrage ». Releve sur les machines reelles : # # shared_buffers 128 Mo — le defaut Debian, PAS derive de la RAM totale # processus Java (tas) 605 Mo — sur une machine de 3 Go # # Ce que ces machines tiennent vraiment, c'est du CACHE DE PAGES. Le reprendre ralentit # des lectures ; il ne fait pas swapper. Et leurs disques vivent sur du NVMe : une lecture # reprise coute peu. Leur plancher redescend donc au cas general. # # CE QUI RESTE HAUT, et pour une raison mesurable : un collecteur garde ses series EN # MEMOIRE et grossit avec le temps — reprendre sous son empreinte actuelle le ferait # swapper des la prochaine fenetre de retention. PLANCHER_MEMOIRE = { "serveur_prometheus": 0.75, # series en memoire, croissantes "serveur_loki": 0.75, "serveur_nextcloud": 0.75, # cache applicatif + traitements de fichiers "serveur_collabora": 0.75, # un document ouvert est un processus } # REDIS N'EST PAS DANS CETTE TABLE, ET C'EST UNE MESURE (2026-09-13, corrigee le 09-14). # « Tout en memoire par construction » justifiait un plancher de 1,00 — donc aucune # reprise possible sur la machine qui le porte. # # PREMIER RELEVE, ET IL ETAIT INCOMPLET : 16 Mo residents, « aucun `maxmemory` configure ». # La seconde affirmation etait FAUSSE — elle venait d'un `grep` sur `/etc/redis/redis.conf`, # alors que le role ecrit dans `/etc/redis/setops.conf`. Interroge la ou la verite vit, # Redis repond `maxmemory 268435456` et `maxmemory-policy allkeys-lru`. # # LA CONCLUSION NE CHANGE PAS, SA RAISON SI. Redis est BORNE a 256 Mo : il ne peut pas # prendre la machine, et epingler plusieurs gigaoctets pour un plafond de 256 Mo serait # absurde. Ce qu'il tient, ce sont des sessions et du cache applicatif, pas un jeu de # donnees. # # LE JOUR OU CE PLAFOND MONTERA, le plancher devra le suivre : un Redis borne a 2 Go doit # garder 2 Go. Le seuil est le `maxmemory` EFFECTIF (`config get maxmemory`), pas ce qu'un # fichier de configuration donne a lire. PLANCHER_DEFAUT = 0.50 def plancher_memoire(memoire_mo: int, groupes) -> int: """Ce que l'hote a le droit de REPRENDRE a cette machine, en Mo. Le plus HAUT plancher de ses groupes gagne : une machine qui porte une base ET un service au repos se dimensionne sur la base. Prendre le plus bas rendrait la memoire que la base a reservee. """ ratios = [PLANCHER_MEMOIRE[g] for g in (groupes or []) if g in PLANCHER_MEMOIRE] r = max(ratios) if ratios else PLANCHER_DEFAUT return max(512, int(round(memoire_mo * r))) def charger_empreinte_role(racine_roles: Path, groupe: str) -> dict | None: """Lit roles//meta/empreinte.yml -> empreinte normalisee, ou None.""" fichier = racine_roles / groupe / "meta" / "empreinte.yml" if not fichier.exists(): return None with fichier.open("r", encoding="utf-8") as fh: data = yaml.safe_load(fh) or {} emp = (data or {}).get("setops_empreinte") or {} if not isinstance(emp, dict): raise ValueError(f"{fichier}: 'setops_empreinte' doit etre une table.") # Un fichier qui declare les valeurs A LA RACINE au lieu de sous # `setops_empreinte:` etait lu comme VIDE : la VM recevait le minimum, en # silence. Quatre roles etaient dans ce cas, et `collab-01` s'est retrouvee # avec 1 coeur / 1 Go pour porter Nextcloud ET Collabora — elle a cesse de # repondre en SSH (2026-08-07). Un fichier qui existe mais ne dit rien est # pire qu'un fichier absent : le repli, lui, aurait donne des valeurs sensees. if not emp and any(c in (data or {}) for c in CHAMPS_EMPREINTE): raise ValueError( f"{fichier}: les valeurs sont declarees a la racine ; elles doivent " f"etre sous la cle `setops_empreinte:`. Sans elle, l'empreinte est lue " f"comme nulle et la VM recoit le minimum." ) return {cle: int(emp.get(cle, 0)) for cle in CHAMPS_EMPREINTE} def empreinte_groupe(racine_roles: Path, groupe: str) -> dict: """Empreinte d'un groupe de service : role > repli connu > defaut.""" return (charger_empreinte_role(racine_roles, groupe) or EMPREINTES_SANS_ROLE.get(groupe) or EMPREINTE_DEFAUT) def _arrondi_superieur(valeur: float, palier: int) -> int: import math return int(math.ceil(valeur / palier) * palier) def deriver_ressources(groupes_service, racine_roles: Path, socle: dict | None = None) -> dict: """Estime coeurs/memoire(Mo)/disque(Go) d'une VM depuis ses logiciels + le SE. `groupes_service` : groupes applicatifs portes par l'hote (hors socle/etat). RAM et disque additifs (contraintes dures) ; coeurs en max (burstable). """ socle = socle or SOCLE_SE coeurs_somme = 0 memoire = socle["memoire_mo"] disque = socle["disque_go"] for groupe in groupes_service: emp = empreinte_groupe(racine_roles, groupe) coeurs_somme += emp["coeurs"] memoire += emp["memoire_mo"] disque += emp["disque_go"] coeurs = max(MIN_COEURS, min(MAX_COEURS, max(socle["coeurs"], coeurs_somme))) memoire = max(MIN_MEMOIRE_MO, _arrondi_superieur(memoire * MARGE_MEMOIRE, PALIER_MEMOIRE_MO)) disque = max(MIN_DISQUE_GO, _arrondi_superieur(disque * MARGE_DISQUE, PALIER_DISQUE_GO)) return {"coeurs": coeurs, "memoire": memoire, "disque": f"{disque}G"} def charger_dependances(path: Path | None) -> dict: if not path: return {} if not path.exists(): raise ValueError(f"Fichier de dependances introuvable: {path}") with path.open("r", encoding="utf-8") as dependencies_file: data = yaml.safe_load(dependencies_file) or {} if not isinstance(data, dict): raise ValueError(f"{path} ne contient pas une table YAML.") groups = data.get("groupes", {}) if not isinstance(groups, dict): raise ValueError(f"{path} doit contenir une cle 'groupes' de type table.") for group, config in groups.items(): if not est_groupe_operationnel(group): raise ValueError(f"Groupe de dependance non operationnel: {group}") if not isinstance(config, dict): raise ValueError(f"Dependance invalide pour {group}: table attendue.") required = config.get("requiert_groupes_actifs", []) if required is None: required = [] if not isinstance(required, list) or not all(isinstance(item, str) for item in required): raise ValueError(f"requiert_groupes_actifs doit etre une liste de groupes pour {group}.") for required_group in required: if not est_groupe_operationnel(required_group): raise ValueError(f"Prerequis non operationnel pour {group}: {required_group}") return groups