erplibre/script/todo/qemu_install_monitor.py
Mathieu Benoit bc0e06137f [IMP] script todo: dashboard async (I/O en thread) + install navigateur CLI
- Ticks ASYNCHRONES : les I/O bloquantes (lecture des logs, stat disque,
  subprocess « virsh list », /proc) sont déportées en THREAD via
  asyncio.to_thread ; seules les mises à jour d'UI restent sur la boucle
  d'événements Textual -> plus de gel, même sous forte charge ou disque lent.
  _collect_table (thread) rassemble statut+disque+télémétrie, l'application
  au tableau se fait ensuite sur la boucle. _tick_log/_tick_domstate idem.
  Helper _read_new (lecture incrémentale) réutilisé.
- Touche « w » : si aucun navigateur CLI n'est présent, on PROPOSE de
  l'installer selon l'OS (apt Ubuntu/Debian, dnf Fedora, pacman Arch — nos 4
  systèmes) : la commande est AFFICHÉE puis exécutée après validation (o/N).

Validé headless : ticks async mettent l'UI à jour ; browser_install_command
renvoie la bonne commande selon l'OS.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
2026-07-29 10:09:06 +00:00

681 lines
26 KiB
Python
Raw Blame History

This file contains ambiguous Unicode characters

This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.

#!/usr/bin/env python3
# © 2021-2026 TechnoLibre (http://www.technolibre.ca)
# License AGPL-3.0 or later (http://www.gnu.org/licenses/agpl)
"""Dashboard Textual : suivi des installations ERPLibre parallèles sur VM.
Principe « détachable » : chaque installation tourne dans un processus
DÉTACHÉ (setsid) qui écrit un fichier log et, à la fin, un marqueur
« __ERPLIBRE_EXIT__ <code> ». Ce module ne fait que VISUALISER ces fichiers :
quitter le dashboard n'arrête rien, on peut le rouvrir pour ré-attacher.
- launch_installs(...) : lance les process détachés + écrit un manifeste JSON.
- run_monitor(manifest_path) : ouvre le dashboard Textual sur un manifeste.
"""
from __future__ import annotations
import asyncio
import json
import os
import shlex
import shutil
import subprocess
import time
from pathlib import Path
try:
from script.todo.todo_i18n import t
except Exception: # pragma: no cover - repli si i18n indisponible
def t(key: str) -> str:
return key
EXIT_MARKER = "__ERPLIBRE_EXIT__"
SSH_OPTS = (
"-o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null "
"-o ConnectTimeout=8"
)
def session_dir() -> Path:
"""Répertoire des logs/manifestes d'installation (créé au besoin)."""
base = Path(os.path.expanduser("~/.erplibre/qemu-install"))
base.mkdir(parents=True, exist_ok=True)
return base
def _launch_one(ip: str, remote_cmd: str, log_path: str) -> None:
"""Lance une install SSH DÉTACHÉE : attend le sshd, exécute, journalise
la sortie puis écrit le marqueur de fin avec le code de sortie."""
# Sonde de disponibilité : on attend que sshd réponde ET que cloud-init
# soit TERMINÉ, via des connexions COURTES successives (jusqu'à ~20 min :
# une architecture ÉMULÉE, s390x/arm64 sur hôte x86, boote lentement).
# Au 1er boot, cloud-init régénère les clés d'hôte et REDÉMARRE sshd : une
# session SSH longue (ex. « cloud-init status --wait ») serait alors tuée
# (« Connection closed by remote host », exit 255 — cas Fedora). Chaque
# itération étant une connexion neuve, un redémarrage de sshd ne casse que
# la tentative en cours. On imprime toujours l'état (|| true) pour matcher
# sur le TEXTE, « status: running » n'ayant pas de code de sortie fiable.
ci_probe = (
"if command -v cloud-init >/dev/null 2>&1; then "
"cloud-init status 2>/dev/null || true; else echo nocloudinit; fi"
)
log_q = shlex.quote(log_path)
# On écrit un message d'attente + un battement toutes les ~30 s : sinon le
# log reste VIDE pendant tout le boot émulé et paraît « bloqué ».
msg_wait = t("Waiting for the VM to start (boot + cloud-init)")
msg_slow = t("(an emulated architecture can be slow; this is normal)")
msg_ready = t("VM ready - starting the ERPLibre install")
wrapper = (
f"echo {shlex.quote('== ' + msg_wait + ' ==')} >> {log_q}; "
f"echo {shlex.quote(' ' + msg_slow)} >> {log_q}; "
f"for i in $(seq 1 240); do "
f"st=$(ssh {SSH_OPTS} -o BatchMode=yes erplibre@{ip} "
f"{shlex.quote(ci_probe)} 2>/dev/null); "
f'case "$st" in '
f"*done*|*disabled*|*error*|*degraded*|*nocloudinit*) break;; "
f"esac; "
f'if [ $((i % 6)) -eq 0 ]; then echo " ... $((i*5))s" >> {log_q}; fi; '
f"sleep 5; done; "
f"echo {shlex.quote('== ' + msg_ready + ' ==')} >> {log_q}; "
f"ssh {SSH_OPTS} erplibre@{ip} {shlex.quote(remote_cmd)} "
f">> {log_q} 2>&1; "
f'echo "{EXIT_MARKER} $?" >> {log_q}'
)
# setsid -f : le process survit à la fermeture du menu / du dashboard.
subprocess.Popen(
["setsid", "-f", "bash", "-c", wrapper],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
start_new_session=True,
)
def _log_header(vm: dict, branch: str, when: str) -> str:
"""En-tête du log : date, VM, distribution, version, architecture, branche.
Permet d'identifier l'installation d'un coup d'œil (et de ne jamais laisser
le log vide pendant l'attente du boot)."""
distro = vm.get("distro") or "?"
version = vm.get("version") or ""
arch = vm.get("arch") or "?"
bar = "=" * 64
return (
f"{bar}\n"
f" ERPLibre — {t('installation')}\n"
f" Date : {when}\n"
f" VM : {vm['name']}\n"
f" Distribution : {distro} {version}\n"
f" Architecture : {arch}\n"
f" Branche : {branch}\n"
f" IP : {vm['ip']}\n"
f"{bar}\n\n"
)
def launch_installs(vms: list[dict], branch: str, remote_cmd: str) -> str:
"""vms : [{name, ip, distro?, version?, arch?}]. Lance chaque install
détachée, écrit un manifeste et retourne son chemin. remote_cmd : script
exécuté dans chaque VM."""
sdir = session_dir()
stamp = time.strftime("%Y%m%d-%H%M%S")
when = time.strftime("%Y-%m-%d %H:%M:%S")
logdir = sdir / stamp
logdir.mkdir(parents=True, exist_ok=True)
entries = []
for vm in vms:
log_path = str(logdir / f"{vm['name']}.log")
# En-tête d'emblée (date/distro/version/arch) : le log n'est jamais
# vide, l'utilisateur voit tout de suite QUOI s'installe.
Path(log_path).write_text(_log_header(vm, branch, when))
_launch_one(vm["ip"], remote_cmd, log_path)
entries.append(
{
"name": vm["name"],
"ip": vm["ip"],
"distro": vm.get("distro"),
"version": vm.get("version"),
"arch": vm.get("arch"),
"log": log_path,
"ssh": f"ssh erplibre@{vm['ip']}",
}
)
manifest = {
"branch": branch,
"started": time.time(),
"vms": entries,
}
manifest_path = str(logdir / "session.json")
Path(manifest_path).write_text(json.dumps(manifest, indent=2))
return manifest_path
def read_status(log_path: str) -> tuple[str, int | None]:
"""(état, code) d'un log : pending / running / done / failed. Ne lit que
la FIN du fichier (le marqueur de sortie est sur la dernière ligne) : lire
tout le log de 30 VM chaque seconde saturait la boucle d'événements du TUI
(lag, interface figée quand l'I/O ralentit)."""
try:
size = os.path.getsize(log_path)
if size == 0:
return "pending", None
with open(log_path, "rb") as fh:
if size > 4096:
fh.seek(-4096, os.SEEK_END)
tail = fh.read().decode(errors="replace")
except OSError:
return "pending", None
if not tail.strip():
return "pending", None
for line in reversed(tail.splitlines()):
if EXIT_MARKER in line:
try:
code = int(line.split()[-1])
except ValueError:
code = 1
return ("done" if code == 0 else "failed"), code
return "running", None
def _read_new(path: str, offset: int) -> tuple[str, int]:
"""Lit le log à partir de `offset` (lecture incrémentale). Renvoie
(nouveau_texte, nouvel_offset). Bloquant -> à appeler dans un thread."""
try:
with open(path, "r", errors="replace") as fh:
fh.seek(offset)
data = fh.read()
return data, fh.tell()
except OSError:
return "", offset
# --------------------------------------------------------------------------- #
# Télémétrie, historique de durées (ETA), navigateur CLI
# --------------------------------------------------------------------------- #
def _stats_path() -> Path:
"""Fichier d'historique persistant (durées d'installation par arch)."""
return session_dir() / "stats.json"
def load_stats() -> dict:
try:
return json.loads(_stats_path().read_text())
except (OSError, ValueError):
return {}
def record_duration(arch: str, secs: float) -> None:
"""Mémorise la durée d'une install (par architecture) pour estimer l'ETA
des prochaines. On garde les 20 dernières valeurs par arch."""
data = load_stats()
durations = data.setdefault("durations", {})
key = arch or "unknown"
lst = durations.setdefault(key, [])
lst.append(int(secs))
durations[key] = lst[-20:]
try:
_stats_path().write_text(json.dumps(data))
except OSError:
pass
def eta_reference(stats: dict, arch: str) -> int | None:
"""Durée d'install de RÉFÉRENCE (médiane) pour cette arch, d'après
l'historique ; repli sur toutes archis confondues. None si aucun historique."""
durations = stats.get("durations", {}) or {}
lst = durations.get(arch or "unknown") or []
if not lst:
lst = [x for v in durations.values() for x in v]
if not lst:
return None
s = sorted(lst)
return s[len(s) // 2]
def _fmt_size(nbytes) -> str:
"""Octets -> « 1.2G » / « 345M » / « 12K »."""
if nbytes is None:
return "-"
for unit, div in (("T", 1 << 40), ("G", 1 << 30), ("M", 1 << 20)):
if nbytes >= div:
return f"{nbytes / div:.1f}{unit}"
return f"{nbytes // 1024}K"
def _fmt_secs(secs) -> str:
"""Secondes -> « 45s » / « 12m » / « 1h05 »."""
secs = int(secs)
if secs < 60:
return f"{secs}s"
if secs < 3600:
return f"{secs // 60}m"
return f"{secs // 3600}h{(secs % 3600) // 60:02d}"
def vm_disk_path(vm: dict) -> str:
"""Chemin du qcow2 de la VM (défaut libvirt si non fourni)."""
return vm.get("disk") or f"/var/lib/libvirt/images/{vm['name']}.qcow2"
def disk_actual_size(path: str) -> int | None:
"""Taille RÉELLEMENT occupée du qcow2 (creux) via st_blocks."""
try:
st = os.stat(path)
return int(getattr(st, "st_blocks", 0)) * 512
except OSError:
return None
# Navigateurs web en ligne de commande, par ordre de préférence (rendu JS
# d'abord — utile pour l'UI Odoo — puis navigateurs texte classiques).
CLI_BROWSERS = ("browsh", "carbonyl", "w3m", "links", "elinks", "lynx")
def cli_browser() -> str | None:
"""Premier navigateur CLI disponible dans le PATH, sinon None."""
for name in CLI_BROWSERS:
if shutil.which(name):
return name
return None
def _os_id() -> str:
"""ID de la distribution hôte (/etc/os-release), ex. « ubuntu », « fedora »."""
try:
for line in open("/etc/os-release", encoding="utf-8"):
if line.startswith("ID="):
return line.split("=", 1)[1].strip().strip('"').lower()
except OSError:
pass
return ""
def browser_install_command() -> list | None:
"""Commande d'installation d'un navigateur CLI (w3m, léger et présent
partout) adaptée à l'OS hôte : apt (Ubuntu/Debian), dnf (Fedora), pacman
(Arch). None si le gestionnaire est inconnu."""
apt = ["sudo", "apt-get", "install", "-y", "w3m"]
dnf = ["sudo", "dnf", "install", "-y", "w3m"]
pac = ["sudo", "pacman", "-S", "--needed", "--noconfirm", "w3m"]
by_id = {
"ubuntu": apt,
"debian": apt,
"linuxmint": apt,
"fedora": dnf,
"arch": pac,
}
cmd = by_id.get(_os_id())
if cmd:
return cmd
# Repli : d'après le gestionnaire de paquets présent.
if shutil.which("apt-get"):
return apt
if shutil.which("dnf"):
return dnf
if shutil.which("pacman"):
return pac
return None
def virsh_domstates() -> dict:
"""{nom: état} de tous les domaines libvirt (« virsh list --all »). Sert à
détecter une VM EN PAUSE ou EFFACÉE pendant le suivi. Un seul appel virsh
pour tout le parc (à interroger à intervalle LENT)."""
try:
res = subprocess.run(
["sudo", "virsh", "list", "--all"],
capture_output=True,
text=True,
timeout=15,
)
except (OSError, subprocess.SubprocessError):
return {}
states = {}
for line in res.stdout.splitlines():
parts = line.split()
# Ignore l'en-tête (« Id Name State ») et le séparateur (« ---- »,
# un seul token). Une VM éteinte a « - » en Id : à NE PAS ignorer.
if len(parts) < 3 or parts[0] == "Id":
continue
states[parts[1]] = " ".join(parts[2:])
return states
# --------------------------------------------------------------------------- #
# Dashboard Textual
# --------------------------------------------------------------------------- #
def run_monitor(manifest_path: str, run_app: bool = True):
"""Ouvre le dashboard Textual sur un manifeste d'installation. `run_app`
à False renvoie l'instance sans la lancer (tests headless)."""
from textual.app import App, ComposeResult
from textual.containers import Horizontal
from textual.widgets import DataTable, Footer, Header, RichLog, Static
manifest = json.loads(Path(manifest_path).read_text())
started = manifest.get("started", time.time())
vms = manifest["vms"]
ICON = {
"pending": "",
"running": "",
"done": "",
"failed": "",
}
class Monitor(App):
CSS = """
DataTable { width: 58; border: solid $accent; }
RichLog { border: solid $accent; }
#telemetry { height: 1; color: $text-muted; }
#sshbar { height: 2; color: $text-muted; }
"""
BINDINGS = [
("q", "quit", "Quitter (détaché)"),
("s", "ssh", "SSH"),
("w", "web", "Web (navigateur CLI)"),
("f", "follow", "Suivre"),
("c", "copy_log", "Copier log"),
]
def __init__(self):
super().__init__()
self._offsets = {vm["name"]: 0 for vm in vms}
self._selected = vms[0]["name"] if vms else None
self._follow = True
# Statuts TERMINAUX mémorisés : une VM finie n'est plus relue
# (réduit fortement l'I/O sur un gros parc). Valeur = (état, code,
# durée à la complétion).
self._final = {}
# Cache des cellules AFFICHÉES : on n'appelle update_cell (donc on
# ne re-render) que si la valeur CHANGE -> plus de churn de rendu.
self._cells = {}
# Historique de durées (ETA) + dossier disque à surveiller.
self._stats = load_stats()
self._disk_dir = os.path.dirname(vm_disk_path(vms[0])) if vms else "/"
# État libvirt (running/paused/gone), rafraîchi à intervalle LENT.
self._domstate = {}
@staticmethod
def _fmt(secs):
mm, ss = divmod(int(secs), 60)
return f"{mm:02d}:{ss:02d}"
def _set_cell(self, table, name, col, value):
key = (name, col)
if self._cells.get(key) == value:
return
self._cells[key] = value
table.update_cell(name, col, value)
def compose(self) -> ComposeResult:
yield Header(show_clock=True)
table = DataTable(id="vms", cursor_type="row")
# Clés de colonnes explicites : update_cell() les référence.
table.add_column("VM", key="vm")
table.add_column("État", key="state")
table.add_column("Durée", key="elapsed")
table.add_column("Disque", key="disk")
for vm in vms:
table.add_row(vm["name"], "", "--:--", "-", key=vm["name"])
# max_lines borne la mémoire/rendu (un install verbeux × 30 VM).
self._log = RichLog(
id="log", highlight=False, markup=False, max_lines=5000
)
with Horizontal():
yield table
yield self._log
# Barre de télémétrie hôte (CPU, disque, ETA parc).
yield Static("", id="telemetry")
yield Static("", id="sshbar")
yield Footer()
def on_mount(self) -> None:
# Le NOMBRE de VM figure dans le titre ; le sous-titre suit la
# progression (terminées / total + durée globale).
self.title = f"ERPLibre — {t('install monitoring')} ({len(vms)} VM)"
self.sub_title = f"0/{len(vms)} {t('completed')}"
self._refresh_ssh()
self._load_selected_log(reset=True)
# Table toutes les 2 s (30 lectures de fin de log), suivi du log
# sélectionné toutes les 1 s (une seule lecture incrémentale).
self.set_interval(2.0, self._tick_table)
self.set_interval(1.0, self._tick_log)
# État libvirt (pause / effacée) : check LENT (appel virsh) toutes
# les 10 s ; le tableau applique le cache à chaque tick (2 s).
self.set_interval(10.0, self._tick_domstate)
# Premier relevé domstate immédiat (async -> via worker).
self.run_worker(self._tick_domstate(), exclusive=False)
# -- helpers -------------------------------------------------------- #
def _vm_by_name(self, name):
return next((v for v in vms if v["name"] == name), None)
def _refresh_ssh(self):
vm = self._vm_by_name(self._selected)
bar = self.query_one("#sshbar", Static)
if vm:
bar.update(
f" {vm['ssh']} (s = SSH · w = web :8069 · "
"c = copier le log · Maj+glisser = sélectionner)\n"
f" Log : {vm['log']}"
)
def _load_selected_log(self, reset=False):
# Lecture SYNCHRONE (mount / changement de ligne). Le suivi périodique
# passe par _tick_log (async, I/O en thread).
vm = self._vm_by_name(self._selected)
if not vm:
return
name = vm["name"]
if reset:
self._log.clear()
self._offsets[name] = 0
new, off = _read_new(vm["log"], self._offsets.get(name, 0))
self._offsets[name] = off
for line in new.splitlines():
if EXIT_MARKER not in line:
self._log.write(line)
def _collect_tele(self):
"""(THREAD) chaîne de télémétrie hôte : CPU % + disque des images."""
try:
ncpu = os.cpu_count() or 1
load1 = os.getloadavg()[0]
du = shutil.disk_usage(self._disk_dir)
used_pct = int(du.used / du.total * 100) if du.total else 0
return (
f" ⚙ CPU {min(999, int(load1 / ncpu * 100))}% "
f"(charge {load1:.1f}/{ncpu}) "
f"💽 {self._disk_dir}: {_fmt_size(du.used)}/"
f"{_fmt_size(du.total)} ({used_pct}%) · "
f"libre {_fmt_size(du.free)}"
)
except Exception:
return ""
def _collect_table(self):
"""(THREAD) statut + taille disque de chaque VM + télémétrie. AUCUNE
mise à jour d'UI ici : uniquement des I/O bloquantes déportées."""
disks, status = {}, {}
for vm in vms:
name = vm["name"]
disks[name] = _fmt_size(disk_actual_size(vm_disk_path(vm)))
if (
name not in self._final
and self._domstate.get(name) != "gone"
):
status[name] = read_status(vm["log"])
return disks, status, self._collect_tele()
async def _tick_table(self):
# I/O (lectures de logs, stat disque, /proc) DÉPORTÉES en thread ->
# la boucle d'événements Textual reste fluide même sous forte
# charge ou disque lent. Les mises à jour d'UI restent sur la boucle.
try:
disks, status, tele = await asyncio.to_thread(
self._collect_table
)
except Exception:
return
try:
table = self.query_one("#vms", DataTable)
now = time.time()
remaining = []
for vm in vms:
name = vm["name"]
self._set_cell(table, name, "disk", disks.get(name, "-"))
if name in self._final:
continue
ds = self._domstate.get(name)
if ds == "gone":
self._final[name] = ("deleted", None, now - started)
self._set_cell(
table, name, "state", f"{t('deleted')}"
)
continue
st = status.get(name)
if st is None:
continue
state, code = st
if state in ("done", "failed"):
elapsed = now - started
self._final[name] = (state, code, elapsed)
if state == "done":
record_duration(vm.get("arch"), elapsed)
self._stats = load_stats()
lbl = "" if state == "done" else f"❌ ({code})"
self._set_cell(table, name, "state", lbl)
self._set_cell(
table, name, "elapsed", self._fmt(elapsed)
)
elif ds == "paused":
self._set_cell(
table, name, "state", f"{t('paused')}"
)
else:
self._set_cell(table, name, "state", ICON[state])
ref = eta_reference(self._stats, vm.get("arch"))
if ref is not None:
remaining.append(max(0, ref - (now - started)))
done = len(self._final)
eta = (
f" · ETA ~{_fmt_secs(max(remaining))}" if remaining else ""
)
self.sub_title = (
f"{done}/{len(vms)} {t('completed')} · "
f"{self._fmt(now - started)}{eta}"
)
if tele:
self.query_one("#telemetry", Static).update(tele)
except Exception:
pass
async def _tick_log(self):
if not self._follow:
return
vm = self._vm_by_name(self._selected)
if not vm:
return
name = vm["name"]
try:
new, off = await asyncio.to_thread(
_read_new, vm["log"], self._offsets.get(name, 0)
)
except Exception:
return
self._offsets[name] = off
for line in new.splitlines():
if EXIT_MARKER not in line:
self._log.write(line)
async def _tick_domstate(self):
# Relevé LENT (subprocess virsh) DÉPORTÉ en thread : ne bloque plus
# la boucle. Une VM absente de « virsh list » est EFFACÉE (« gone »).
try:
states = await asyncio.to_thread(virsh_domstates)
except Exception:
return
for vm in vms:
self._domstate[vm["name"]] = states.get(vm["name"], "gone")
# -- events --------------------------------------------------------- #
def on_data_table_row_highlighted(self, event) -> None:
name = event.row_key.value
if name and name != self._selected:
self._selected = name
self._refresh_ssh()
self._load_selected_log(reset=True)
def action_follow(self) -> None:
self._follow = not self._follow
def action_ssh(self) -> None:
vm = self._vm_by_name(self._selected)
if not vm:
return
with self.suspend():
os.system(f"ssh {SSH_OPTS} erplibre@{vm['ip']} || true")
def action_web(self) -> None:
"""Ouvre l'UI web de la VM (Odoo :8069) dans un navigateur CLI
(browsh/carbonyl/w3m/links/elinks/lynx). Surtout utile une fois
l'installation TERMINÉE."""
vm = self._vm_by_name(self._selected)
if not vm or not vm.get("ip"):
self.notify("Pas d'IP pour cette VM.", title="Web")
return
browser = cli_browser()
if not browser:
# Aucun navigateur CLI : proposer de l'INSTALLER (commande
# adaptée à l'OS, affichée + validée avant exécution).
install = browser_install_command()
with self.suspend():
if not install:
print(
"Aucun navigateur CLI et gestionnaire de paquets "
"inconnu. Installez w3m/lynx/links/elinks/browsh."
)
input("Entrée pour revenir…")
else:
printable = " ".join(install)
print(
"Aucun navigateur CLI trouvé. Commande "
"d'installation proposée :\n"
f" {printable}"
)
ans = input(
"Installer maintenant ? (o/N) : "
).strip().lower()
if ans in ("o", "oui", "y", "yes"):
os.system(printable + " || true")
browser = cli_browser()
if not browser:
self.notify(
"Toujours aucun navigateur CLI disponible.",
title="Web",
severity="warning",
)
return
url = f"http://{vm['ip']}:8069"
with self.suspend():
os.system(f"{browser} {shlex.quote(url)} || true")
def action_copy_log(self) -> None:
"""Copie le log complet de la VM sélectionnée dans le
presse-papiers (OSC 52 ; marche aussi à travers SSH)."""
vm = self._vm_by_name(self._selected)
if not vm:
return
try:
text = Path(vm["log"]).read_text(errors="replace")
except OSError:
return
self.copy_to_clipboard(text)
self.notify(
f"Log de {vm['name']} copié ({len(text)} car.)",
title="Presse-papiers",
)
app = Monitor()
if run_app:
app.run()
return app