agent de pod : shell pilote par le Hub

agent_pod.py

@@ -0,0 +1,239 @@
+"""Agent de pod : un shell pilote par le Hub.
+
+POURQUOI. Ce pod n'expose ni IP publique ni port TCP -- seulement 8080 en HTTP
+-- et son `PUBLIC_KEY` est vide. L'image `vllm/vllm-openai` ne contient pas de
+`sshd`, et notre `dockerStartCmd` remplace l'entrypoint : rien n'ecoute sur 22.
+Aucun SSH n'est donc possible. Le seul canal disponible est sortant : HTTPS
+vers le Hub. Cet agent en fait un shell.
+
+  - il relit `cmd/<pod>.sh` toutes les 8 s ; si l'empreinte a change, il
+    l'execute dans un fil separe et publie la sortie dans `etat/out-<pod>.log` ;
+  - il publie un etat machine toutes les 30 s dans `etat/vie-<pod>.log` :
+    GPU, RAM DU CGROUP, disque, avancement du telechargement, queue des
+    journaux en cours ;
+  - il lance le telechargement des poids des le demarrage, sans attendre une
+    premiere commande : 125,9 Go ne s'attendent pas.
+
+PIEGE MESURE UNE FOIS. `free -g` dans un conteneur lit /proc/meminfo, donc la
+RAM de l'HOTE (1 133 Go ici) et non la limite du conteneur (125 Go). Toute
+conclusion sur la faisabilite du dechargement PLE tiree de `free` est fausse.
+On lit donc le cgroup, v2 puis v1.
+
+PIEGE DE LA VIE DU POD. `volumeInGb = 0` : le disque conteneur est efface a
+chaque arret. Les 125,9 Go ne sont telecharges qu'une fois par vie du pod, donc
+toute l'iteration doit tenir dans une seule session -- d'ou ce shell.
+"""
+import hashlib
+import os
+import shutil
+import subprocess
+import threading
+import time
+import urllib.request
+
+DEPOT = "patdev/k3-a40-bootstrap"
+POD = os.environ.get("RUNPOD_POD_ID", "inconnu")
+JETON = os.environ.get("HF_TOKEN")
+TRAVAIL = "/travail"
+MODELE = os.environ.get("BANC_MODEL", "RadixArk/Qwen3.8-Flash-Next-NVFP4")
+# Taille annoncee du depot, pour transformer les octets en pourcentage.
+ATTENDU_GO = float(os.environ.get("BANC_TAILLE_GO", "125.9"))
+
+os.makedirs(TRAVAIL, exist_ok=True)
+os.makedirs(TRAVAIL + "/sorties", exist_ok=True)
+
+UA = ("Mozilla/5.0 (Windows NT 10.0; Win64; x64) AppleWebKit/537.36 "
+      "(KHTML, like Gecko) Chrome/126.0.0.0 Safari/537.36")
+
+COURANTES = []
+
+
+def sh(cmd, timeout=25):
+    try:
+        r = subprocess.run(["bash", "-lc", cmd], capture_output=True,
+                           text=True, timeout=timeout)
+        return (r.stdout + r.stderr).strip()
+    except Exception as e:
+        return "%s: %s" % (type(e).__name__, str(e)[:120])
+
+
+def publier(chemin_local, chemin_depot):
+    from huggingface_hub import HfApi
+    for essai in range(3):
+        try:
+            HfApi().upload_file(path_or_fileobj=chemin_local,
+                                path_in_repo=chemin_depot,
+                                repo_id=DEPOT, token=JETON)
+            return True
+        except Exception as e:
+            if essai == 2:
+                print("publication %s : %s" % (chemin_depot, e), flush=True)
+        time.sleep(3)
+    return False
+
+
+def lire_hub(chemin):
+    """Lit un fichier du depot en contournant le cache du CDN."""
+    url = ("https://huggingface.co/%s/resolve/main/%s?t=%d"
+           % (DEPOT, chemin, int(time.time())))
+    entetes = {"User-Agent": UA, "Cache-Control": "no-cache"}
+    if JETON:
+        entetes["Authorization"] = "Bearer " + JETON
+    try:
+        req = urllib.request.Request(url, headers=entetes)
+        with urllib.request.urlopen(req, timeout=25) as r:
+            return r.read().decode("utf-8", "replace")
+    except Exception:
+        return None
+
+
+# --------------------------------------------------------------- etat machine
+def octets_repertoire(d):
+    total = 0
+    for racine, _, fichiers in os.walk(d):
+        for f in fichiers:
+            try:
+                total += os.stat(os.path.join(racine, f),
+                                 follow_symlinks=False).st_size
+            except OSError:
+                pass
+    return total
+
+
+def ram_cgroup():
+    """La limite REELLE du conteneur, pas celle de l'hote."""
+    paires = (("/sys/fs/cgroup/memory.max", "/sys/fs/cgroup/memory.current"),
+              ("/sys/fs/cgroup/memory/memory.limit_in_bytes",
+               "/sys/fs/cgroup/memory/memory.usage_in_bytes"))
+    for lim, use in paires:
+        try:
+            t = open(lim).read().strip()
+            u = int(open(use).read().strip())
+            if t == "max":
+                return None, u
+            t = int(t)
+            if t > (1 << 50):          # « pas de limite » deguise en immense
+                return None, u
+            return t, u
+        except Exception:
+            continue
+    return None, None
+
+
+CACHE = os.environ.get("HF_HOME") or os.path.expanduser("~/.cache/huggingface")
+
+
+def etat():
+    l = []
+    l.append("[VIE] %s pod=%s" % (time.strftime("%H:%M:%S", time.gmtime()), POD))
+    l.append(sh("nvidia-smi --query-gpu=name,memory.used,memory.total,"
+                "utilization.gpu --format=csv,noheader"))
+    lim, use = ram_cgroup()
+    if lim:
+        l.append("RAM conteneur : %.1f / %.1f Go  (cgroup)"
+                 % (use / 1e9, lim / 1e9))
+    elif use is not None:
+        l.append("RAM conteneur : %.1f Go utilises, aucune limite cgroup lisible"
+                 % (use / 1e9))
+    try:
+        du = shutil.disk_usage("/")
+        l.append("disque / : %.0f Go libres sur %.0f" % (du.free / 1e9,
+                                                         du.total / 1e9))
+    except Exception:
+        pass
+    try:
+        go = octets_repertoire(CACHE) / 1e9
+        l.append("poids telecharges : %.1f / %.1f Go  (%.0f %%)"
+                 % (go, ATTENDU_GO, 100 * go / ATTENDU_GO))
+    except Exception:
+        pass
+    for nom in ("dl.log", "vllm.log", "banc.log"):
+        p = os.path.join(TRAVAIL, nom)
+        if os.path.exists(p):
+            q = sh("tail -c 1500 %s | tr '\\r' '\\n' | grep -v '^$' | tail -4" % p)
+            if q:
+                l.append("--- %s ---\n%s" % (nom, q))
+    vivantes = [c for c in COURANTES if c["fil"].is_alive()]
+    for c in vivantes:
+        l.append("commande en cours : %s (depuis %d s)"
+                 % (c["nom"], time.time() - c["t0"]))
+    return "\n".join(x for x in l if x)
+
+
+# ------------------------------------------------------------- telechargement
+CODE_DL = """
+import os, time
+from huggingface_hub import snapshot_download
+t0 = time.time()
+for essai in range(6):
+    try:
+        p = snapshot_download(%r, max_workers=8,
+                              token=os.environ.get('HF_TOKEN'),
+                              ignore_patterns=['*.pth', 'original/*'])
+        print('TELECHARGE en %%.0f s -> %%s' %% (time.time() - t0, p), flush=True)
+        break
+    except Exception as e:
+        print('essai', essai, type(e).__name__, str(e)[:200], flush=True)
+        time.sleep(10)
+else:
+    print('TELECHARGEMENT ECHOUE', flush=True)
+"""
+
+
+def telecharger():
+    chemin = os.path.join(TRAVAIL, "dl.py")
+    with open(chemin, "w") as f:
+        f.write(CODE_DL % MODELE)
+    with open(os.path.join(TRAVAIL, "dl.log"), "w") as f:
+        subprocess.run(["python3", chemin], stdout=f, stderr=subprocess.STDOUT)
+
+
+# ------------------------------------------------------------------ commandes
+def executer(nom, script):
+    chemin = os.path.join(TRAVAIL, "sorties", nom + ".log")
+    with open(chemin, "w") as f:
+        f.write("$ %s\n%s\n%s\n%s\n" % (nom, "-" * 70, script, "-" * 70))
+        f.flush()
+        p = subprocess.Popen(["bash", "-lc", script], stdout=f,
+                             stderr=subprocess.STDOUT, cwd=TRAVAIL)
+        code = p.wait()
+        f.write("\n%s\n[FIN] code=%d\n" % ("-" * 70, code))
+    publier(chemin, "etat/out-%s.log" % POD)
+    print("commande %s terminee (code %d)" % (nom, code), flush=True)
+
+
+def boucle_commandes():
+    vue = None
+    n = 0
+    while True:
+        texte = lire_hub("cmd/%s.sh" % POD)
+        if texte is not None and texte.strip():
+            h = hashlib.sha256(texte.encode()).hexdigest()[:12]
+            if h != vue:
+                vue = h
+                n += 1
+                nom = "%03d-%s" % (n, h)
+                print("commande recue %s" % nom, flush=True)
+                fil = threading.Thread(target=executer, args=(nom, texte),
+                                       daemon=True)
+                fil.start()
+                COURANTES.append({"nom": nom, "fil": fil, "t0": time.time()})
+        time.sleep(8)
+
+
+def boucle_etat():
+    p = os.path.join(TRAVAIL, "vie.log")
+    while True:
+        try:
+            with open(p, "w") as f:
+                f.write(etat() + "\n")
+            publier(p, "etat/vie-%s.log" % POD)
+        except Exception as e:
+            print("etat:", e, flush=True)
+        time.sleep(30)
+
+
+print("[AGENT] pod=%s modele=%s" % (POD, MODELE), flush=True)
+threading.Thread(target=telecharger, daemon=True).start()
+threading.Thread(target=boucle_commandes, daemon=True).start()
+boucle_etat()