Daniel Allaire
4de04ec0e1
VLAN = 1000 + index*10 + zone (toujours 4 chiffres) -> VMID 9 chiffres. Longueurs uniformes et mnemotechniques. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>
510 lines
22 KiB
Python
510 lines
22 KiB
Python
#!/usr/bin/env python3
|
|
"""Regles partagees pour les inventaires statiques Set-OPS."""
|
|
|
|
from __future__ import annotations
|
|
|
|
import re
|
|
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}
|
|
|
|
|
|
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) -> 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.
|
|
"""
|
|
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.")
|
|
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.")
|
|
|
|
|
|
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 {}
|
|
edge_dom = conf.get("edge")
|
|
if edge is not None and edge_dom != edge:
|
|
continue
|
|
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 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"}
|
|
|
|
|
|
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)
|
|
|
|
|
|
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
|
|
|
|
|
|
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).
|
|
"""
|
|
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
|
|
base3 = ".".join(str(c.get("sous_reseau", "")).split(".")[:3])
|
|
# Prefixe VMID = index d'instance (federation multi-ecosystemes). Defaut 9 :
|
|
# retro-compatible (instances sans index, bacs a sable). 1 chiffre -> 9 instances
|
|
# prod federables avec un VMID a 5 chiffres ; elargir l'index pour davantage.
|
|
index = nomenclature.get("index", 9)
|
|
hote = svc * 10 + int(seq)
|
|
# Schema du VMID. 'ip-miroir' (opt-in) : 9 chiffres mnemotechniques encodant
|
|
# index + VLAN + octet-hote + sequence -> le VMID contient litteralement l'IP
|
|
# (10.(10+index).VLAN.hote) et le tenant. 'compact' (defaut, retro-compatible) :
|
|
# index.categorie.service.sequence (largeur variable, VMID des instances existantes).
|
|
schema = nomenclature.get("vmid_schema", "compact")
|
|
if schema == "ip-miroir":
|
|
# VLAN DERIVE : 1000 + index x 10 + zone. Le socle 1000 force le VLAN a
|
|
# TOUJOURS 4 chiffres (1011..4094) -> le VMID (VLAN.octet.seq) fait TOUJOURS
|
|
# 9 chiffres : LONGUEURS UNIFORMES et mnemotechniques. Unique GLOBALEMENT sur
|
|
# un trunk convergE (retirer 1000 redonne index x 10 + zone = tenant + zone).
|
|
vlan = 1000 + index * 10 + int(cat)
|
|
vmid = f"{vlan}{hote:03d}{int(seq):02d}"
|
|
else:
|
|
vlan = c.get("vlan")
|
|
vmid = f"{index}{cat}{svc}{int(seq):02d}"
|
|
return {
|
|
"vmid": vmid,
|
|
"vlan": vlan,
|
|
"adresse_ip": f"{base3}.{hote}",
|
|
"passerelle": c.get("passerelle"),
|
|
"cidr": nomenclature.get("cidr_hote", 24),
|
|
}
|
|
|
|
|
|
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.")
|
|
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.")
|
|
|
|
|
|
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/<groupe>/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")
|
|
|
|
|
|
def charger_empreinte_role(racine_roles: Path, groupe: str) -> dict | None:
|
|
"""Lit roles/<groupe>/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.")
|
|
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
|