Paillasse/openstack/tests/capacite.py

#!/usr/bin/env python3
"""Test de capacité (étape 13) : N VM eleve.petit lancées en même temps.

À lancer en root sur le serveur (client openstack de Kolla, virsh dans le
conteneur nova_libvirt) :

  sudo python3 capacite.py --vm 20 --sortie /root/capacite-AAAA-MM-JJ

Phases : démarrage simultané, repos, charge CPU, charge mémoire (arrêtée
d'elle-même si la mémoire disponible de l'hôte passe sous --plancher-go).
Crée un réseau « capacite-test » et des VM « capacite-N » dans le projet
paillasse ; ne supprime rien (nettoyage séparé, après accord).
"""
import argparse
import base64
import csv
import json
import os
import subprocess
import threading
import time

ENV = dict(os.environ, PATH="/opt/kolla-venv/bin:" + os.environ["PATH"],
           OS_CLIENT_CONFIG_FILE="/etc/kolla/clouds-paillasse.yaml", OS_CLOUD="paillasse-service")


def openstack(*args):
    return subprocess.run(["openstack", *args], env=ENV, check=True, capture_output=True, text=True).stdout


def agent(uuid, commande, delai=5):
    r = subprocess.run(["docker", "exec", "nova_libvirt", "virsh", "qemu-agent-command", uuid,
                        json.dumps(commande), "--timeout", str(delai)], capture_output=True, text=True)
    return json.loads(r.stdout) if r.returncode == 0 else None


def executer(uuid, script, attendre=True, delai=30):
    r = agent(uuid, {"execute": "guest-exec", "arguments": {"path": "/bin/sh", "arg": ["-c", script], "capture-output": True}})
    if not r or not attendre:
        return r is not None, ""
    fin = time.time() + delai
    while time.time() < fin:
        s = agent(uuid, {"execute": "guest-exec-status", "arguments": {"pid": r["return"]["pid"]}})
        if s and s["return"].get("exited"):
            return s["return"].get("exitcode") == 0, base64.b64decode(s["return"].get("out-data", "")).decode()
        time.sleep(0.3)
    return False, "délai"


def meminfo():
    m = {}
    for ligne in open("/proc/meminfo"):
        cle, val = ligne.split(":")
        m[cle] = int(val.split()[0]) // 1024
    return m


def cpu_brut():
    v = list(map(int, open("/proc/stat").readline().split()[1:]))
    return sum(v), v[3] + v[4]


def rss_qemu():
    r = subprocess.run(["ps", "-C", "qemu-system-x86_64,qemu-kvm", "-o", "rss=,args="], capture_output=True, text=True)
    total, par_vm = 0, {}
    for ligne in r.stdout.splitlines():
        rss, args = ligne.strip().split(None, 1)
        mo = int(rss) // 1024
        total += mo
        for mot in args.split():
            if mot.startswith("uuid="):
                par_vm[mot[5:]] = mo
        if "-uuid" in args:
            par_vm[args.split("-uuid")[1].split()[0]] = mo
    return total, par_vm


class Echantillonneur(threading.Thread):
    """Relève l'hôte toutes les 2 s : mémoire, CPU, charge, RSS des VM."""

    def __init__(self, chemin):
        super().__init__(daemon=True)
        self.chemin, self.phase, self.arret = chemin, "base", threading.Event()

    def run(self):
        with open(self.chemin, "w", newline="") as f:
            w = csv.writer(f)
            w.writerow(["t", "phase", "mem_disponible_mo", "mem_utilisee_mo", "cpu_pct", "charge_1min", "rss_qemu_total_mo"])
            t0, (tot0, idle0) = time.time(), cpu_brut()
            while not self.arret.is_set():
                time.sleep(2)
                tot, idle = cpu_brut()
                cpu = 100 * (1 - (idle - idle0) / max(1, tot - tot0))
                tot0, idle0 = tot, idle
                m = meminfo()
                w.writerow([round(time.time() - t0, 1), self.phase, m["MemAvailable"], m["MemTotal"] - m["MemAvailable"],
                            round(cpu, 1), open("/proc/loadavg").read().split()[0], rss_qemu()[0]])
                f.flush()


def main():
    p = argparse.ArgumentParser()
    p.add_argument("--vm", type=int, default=20)
    p.add_argument("--sortie", required=True)
    p.add_argument("--plancher-go", type=float, default=3.0)
    a = p.parse_args()
    os.makedirs(a.sortie, exist_ok=True)
    resume = {"vm": a.vm, "base": meminfo()}

    ech = Echantillonneur(os.path.join(a.sortie, "hote.csv"))
    ech.start()
    time.sleep(10)

    if "capacite-test" not in openstack("network", "list", "-f", "value", "-c", "Name").split():
        openstack("network", "create", "--disable-port-security", "capacite-test")
        openstack("subnet", "create", "--network", "capacite-test", "--subnet-range", "10.250.0.0/24", "--no-dhcp", "capacite-test")

    # 1. Démarrage simultané
    ech.phase = "demarrage"
    t0 = time.time()
    openstack("server", "create", "--image", "paillasse-debian", "--flavor", "eleve.petit", "--network", "capacite-test",
              "--min", str(a.vm), "--max", str(a.vm), "capacite")
    vms, actives, pretes = {}, {}, {}
    while time.time() - t0 < 600 and len(pretes) < a.vm:
        for s in json.loads(openstack("server", "list", "--name", "^capacite-", "-f", "json", "-c", "ID", "-c", "Name", "-c", "Status")):
            vms[s["ID"]] = s["Name"]
            if s["Status"] == "ACTIVE" and s["ID"] not in actives:
                actives[s["ID"]] = round(time.time() - t0, 1)
            if s["Status"] == "ERROR":
                actives.setdefault(s["ID"], "ERROR")
        for uuid in list(actives):
            if uuid not in pretes and actives[uuid] != "ERROR" and agent(uuid, {"execute": "guest-ping"}, 2):
                pretes[uuid] = round(time.time() - t0, 1)
        time.sleep(1)
    with open(os.path.join(a.sortie, "demarrage.csv"), "w", newline="") as f:
        w = csv.writer(f)
        w.writerow(["vm", "uuid", "active_s", "agent_pret_s"])
        for uuid, nom in sorted(vms.items(), key=lambda x: x[1]):
            w.writerow([nom, uuid, actives.get(uuid, ""), pretes.get(uuid, "")])
    resume["demarrage"] = {"pretes": len(pretes), "erreurs": sum(1 for v in actives.values() if v == "ERROR"),
                           "premiere_s": min(pretes.values(), default=None), "derniere_s": max(pretes.values(), default=None)}

    # 2. Repos
    ech.phase = "repos"
    time.sleep(120)
    total, par_vm = rss_qemu()
    nos = {u: par_vm.get(u) for u in pretes}
    resume["repos"] = {"mem": meminfo(), "rss_par_vm_mo": nos, "rss_moyen_mo": round(sum(v for v in nos.values() if v) / max(1, len(nos)))}

    # 3. Charge CPU : un cœur à 100 % dans chaque VM pendant 60 s
    ech.phase = "cpu"
    for uuid in pretes:
        executer(uuid, "timeout 60 sh -c 'while :; do :; done'", attendre=False)
    time.sleep(20)
    latences = []
    for uuid in list(pretes)[:5]:
        t = time.time()
        executer(uuid, "true")
        latences.append(round(time.time() - t, 2))
    resume["cpu"] = {"latence_commande_s": latences, "charge_1min": open("/proc/loadavg").read().split()[0]}
    time.sleep(45)

    # 4. Charge mémoire : 350 Mo touchés dans chaque VM, une à une
    ech.phase = "memoire"
    paliers = []
    for i, uuid in enumerate(pretes, 1):
        dispo = meminfo()["MemAvailable"]
        if dispo < a.plancher_go * 1024:
            resume["memoire_arret"] = f"plancher atteint avant la VM {i} ({dispo} Mo disponibles)"
            break
        ok, _ = executer(uuid, "head -c 350M /dev/zero > /dev/shm/charge && sync", delai=60)
        time.sleep(3)
        paliers.append({"vm": i, "ok": ok, "mem_disponible_mo": meminfo()["MemAvailable"], "rss_vm_mo": rss_qemu()[1].get(uuid)})
    resume["memoire"] = paliers
    if len(paliers) >= 2:
        resume["memoire_cout_par_vm_mo"] = round((paliers[0]["mem_disponible_mo"] - paliers[-1]["mem_disponible_mo"]) / (len(paliers) - 1))

    ech.phase = "fin"
    time.sleep(6)
    ech.arret.set()
    json.dump(resume, open(os.path.join(a.sortie, "resume.json"), "w"), indent=2, ensure_ascii=False)
    print(json.dumps({k: v for k, v in resume.items() if k not in ("memoire",)}, indent=1, ensure_ascii=False)[:3000])


if __name__ == "__main__":
    main()