configurar.py

#!/usr/bin/env python3
"""Crea en Semaphore el proyecto «Nodos del clúster» con sus plantillas (botones).

Lo lanza scripts/portal.sh; se puede repetir: lo que ya existe no se duplica.
Lee la URL y las credenciales de admin de las variables PORTAL_URL, PORTAL_USUARIO,
PORTAL_CLAVE y, si el usuario tiene 2FA, PORTAL_CODIGO (el código de 6 dígitos de la
app). Nunca de argumentos, para que no queden en el historial.
"""
import http.cookiejar
import json
import os
import sys
import urllib.request

URL = os.environ["PORTAL_URL"].rstrip("/")
PROYECTO = "Nodos del clúster"
REPO = "/portal/repo"  # copia de main montada en solo lectura

# Valores para elegir a qué nodos se aplica cada botón (variable `nodos`).
NODOS = ["all", "workers", "masters", "master-2", "master-3", "master-1",
         "worker-1", "worker-3", "worker-2"]

PLANTILLAS = [
    # (nombre, playbook, argumentos, descripción)
    ("1 · Revisar ahora", "revisar", [],
     "Revisa qué actualizaciones faltan en cada nodo. No instala nada."),
    ("2 · Simular actualización", "actualizar", ["--check"],
     "Dice qué se instalaría en cada nodo. No instala ni reinicia nada."),
    ("3 · Actualizar sistema", "actualizar", [],
     "Instala las actualizaciones nodo a nodo. Si uno pide reinicio: drain, reinicio, "
     "espera a que vuelva Ready y uncordon. El master-1 no se reinicia solo."),
    ("4 · Actualizar Tailscale", "tailscale", [], "Actualiza solo Tailscale, nodo a nodo."),
    ("5 · Reiniciar los que lo piden", "reiniciar", [],
     "Reinicia, de uno en uno y con drain, los nodos que lo piden."),
    ("6 · Actualizar k3s", "k3s", [],
     "Sube k3s a la versión indicada: primero los masters (de uno en uno, con snapshot de "
     "etcd) y después los workers. Parche: sin drain. Cambio de rama (como mucho una): con "
     "drain; antes, revisar compatibilidades (docs/operacion.md). Prueba antes con --check."),
]

# Variable extra de cada plantilla, además de «nodos».
VERSION_K3S = {
    "name": "version", "title": "Versión de k3s", "type": "", "required": True,
    "description": "Formato v1.X.Y+k3sN, p. ej. v1.34.12+k3s1 (releases de github.com/k3s-io/k3s)",
}
EXTRA = {"k3s": [VERSION_K3S]}

ENTORNO = {
    # La imagen es de solo lectura salvo /tmp.
    "ANSIBLE_CONFIG": f"{REPO}/ansible/ansible.cfg",
    "ANSIBLE_HOME": "/tmp/ansible-home",
    "ANSIBLE_LOCAL_TEMP": "/tmp/ansible-local",
    "ANSIBLE_FORCE_COLOR": "true",
}

jar = http.cookiejar.CookieJar()
abrir = urllib.request.build_opener(urllib.request.HTTPCookieProcessor(jar)).open


def api(metodo, ruta, datos=None):
    cuerpo = json.dumps(datos).encode() if datos is not None else None
    req = urllib.request.Request(URL + "/api" + ruta, data=cuerpo, method=metodo,
                                 headers={"Content-Type": "application/json", "Accept": "application/json"})
    try:
        with abrir(req, timeout=20) as r:
            texto = r.read().decode()
    except urllib.error.HTTPError as e:
        sys.exit(f"✘ {metodo} {ruta}: HTTP {e.code} {e.read().decode()[:300]}")
    return json.loads(texto) if texto.strip() else None


def buscar_o_crear(ruta, nombre, datos):
    for x in api("GET", ruta) or []:
        if x.get("name") == nombre:
            print(f"  · ya existe: {nombre}")
            return x
    print(f"  + {nombre}")
    return api("POST", ruta, datos)


def main():
    api("POST", "/auth/login", {"auth": os.environ["PORTAL_USUARIO"], "password": os.environ["PORTAL_CLAVE"]})
    codigo = os.environ.get("PORTAL_CODIGO", "").strip()
    if codigo:  # 2FA (TOTP): lo mismo que hace la web tras el login
        api("POST", "/auth/verify", {"passcode": codigo})

    p = buscar_o_crear("/projects", PROYECTO, {"name": PROYECTO, "alert": True, "max_parallel_tasks": 1})
    pid = p["id"]
    base = f"/project/{pid}"

    clave = buscar_o_crear(f"{base}/keys", "Sin clave (la pone ansible.cfg)",
                           {"name": "Sin clave (la pone ansible.cfg)", "type": "none", "project_id": pid})
    repo = buscar_o_crear(f"{base}/repositories", "k3s (main)",
                          {"name": "k3s (main)", "git_url": REPO, "git_branch": "",
                           "ssh_key_id": clave["id"], "project_id": pid})
    inv = buscar_o_crear(f"{base}/inventory", "Nodos (ansible/inventario.yml)",
                         {"name": "Nodos (ansible/inventario.yml)", "type": "file",
                          "inventory": "ansible/inventario.yml", "repository_id": repo["id"],
                          "ssh_key_id": clave["id"], "project_id": pid})
    env = buscar_o_crear(f"{base}/environment", "Portal",
                         {"name": "Portal", "json": "{}", "env": json.dumps(ENTORNO), "project_id": pid})

    encuesta = [{
        "name": "nodos", "title": "Nodos", "type": "enum", "required": True,
        "description": "all = todos, por grupos (workers, masters) o uno solo",
        "values": [{"name": n, "value": n} for n in NODOS],
    }]
    for nombre, playbook, args, desc in PLANTILLAS:
        survey = encuesta + EXTRA.get(playbook, [])
        buscar_o_crear(f"{base}/templates", nombre, {
            "name": nombre, "description": desc, "project_id": pid,
            "playbook": f"ansible/playbooks/{playbook}.yml", "app": "ansible", "type": "",
            "arguments": json.dumps(args), "repository_id": repo["id"], "inventory_id": inv["id"],
            "environment_id": env["id"], "survey_vars": survey, "allow_override_args_in_task": False,
        })
    print(f"✓ Proyecto «{PROYECTO}» listo: {URL}/project/{pid}/templates")


if __name__ == "__main__":
    main()