#!/usr/bin/env python3 import argparse import logging import os import time import yaml from dotenv import load_dotenv from proxmoxer import ProxmoxAPI, ResourceException from proxmoxer.tools import Tasks from functions import clone_template, delete_vm, get_vm_ip def tasks_wait_completion( proxmox: ProxmoxAPI, upids: list[str], polling_interval: int = 30 ) -> None: for upid in upids: Tasks.blocking_status(proxmox, upid, polling_interval=polling_interval) def parse_args() -> argparse.Namespace: parser = argparse.ArgumentParser() parser.add_argument( "branch", help="used to pick the appropriate VM template to clone", choices=["sisyphus", "c10f2", "p10"], ) return parser.parse_args() def main() -> None: load_dotenv() if "PROXMOX_HOST" not in os.environ: raise RuntimeError("PROXMOX_HOST environment variable is not supplied") if "PROXMOX_USER" not in os.environ: raise RuntimeError("PROXMOX_USER environment variable is not supplied") if "PROXMOX_USER_FULL" not in os.environ: raise RuntimeError("PROXMOX_USER_FULL environment variable is not supplied") if "PROXMOX_PASSWORD" not in os.environ: raise RuntimeError("PROXMOX_PASSWORD environment variable is not supplied") PROXMOX_HOST = os.environ["PROXMOX_HOST"] PROXMOX_USER = os.environ["PROXMOX_USER"] PROXMOX_USER_FULL = os.environ["PROXMOX_USER_FULL"] PROXMOX_PASSWORD = os.environ["PROXMOX_PASSWORD"] args = parse_args() FORMAT = "%(asctime)s %(levelname)s\t%(name)s\t%(message)s" logging.basicConfig(format=FORMAT) log = logging.getLogger(__name__) log.setLevel(logging.DEBUG) proxmox = ProxmoxAPI( PROXMOX_HOST, user=PROXMOX_USER_FULL, password=PROXMOX_PASSWORD, verify_ssl=False, timeout=5, ) node = "pve02" match args.branch: case "sisyphus": template_id: int = 374 case "p10": template_id: int = 520 case "c10f2": template_id: int = 530 case _: raise RuntimeError(f"Unknown branch: {args.branch!r}") log.info(f"branch: {args.branch!r}, vm template id: {template_id}") prefix: str = PROXMOX_USER + "-test-k8s" vm_names: dict[str, str] = { "master": f"{prefix}-master", "worker1": f"{prefix}-node1", "worker2": f"{prefix}-node2", } vm_ids: dict[str, int] = { "master": 510, "worker1": 511, "worker2": 512, } delete: str = os.environ["DELETE_VMS"] if delete not in ("0", "1"): raise RuntimeError( f"Invalid value for DELETE_VMS environment variable: {delete!r}, expected '0' or '1'" ) if delete != "0": log.info("Stopping VMs %s...", vm_ids) tasks_wait_completion( proxmox, [ proxmox.nodes(node).qemu(vm_ids["master"]).status.stop.post(), proxmox.nodes(node).qemu(vm_ids["worker1"]).status.stop.post(), proxmox.nodes(node).qemu(vm_ids["worker2"]).status.stop.post(), ], ) log.info("Deleting VMs %s...", vm_ids) tasks_wait_completion( proxmox, [ delete_vm(proxmox, node, vm_ids["master"]), delete_vm(proxmox, node, vm_ids["worker1"]), delete_vm(proxmox, node, vm_ids["worker2"]), ], ) vm_id_master, upid_master = clone_template( proxmox, node, template_id, newid=vm_ids["master"], name=vm_names["master"], ) assert vm_id_master == vm_ids["master"] vm_id_worker1, upid_worker1 = clone_template( proxmox, node, template_id, newid=vm_ids["worker1"], name=vm_names["worker1"], ) assert vm_id_worker1 == vm_ids["worker1"] vm_id_worker2, upid_worker2 = clone_template( proxmox, node, template_id, newid=vm_ids["worker2"], name=vm_names["worker2"], ) assert vm_id_worker2 == vm_ids["worker2"] log.info("Waiting for cloning to complete...") tasks_wait_completion( proxmox, [ upid_master, upid_worker1, upid_worker2, ], ) log.info("Cloning completed!") log.info("Disabling VM delete-protection...") proxmox.nodes(node).qemu(vm_ids["master"]).config.put(protection=0) proxmox.nodes(node).qemu(vm_ids["worker1"]).config.put(protection=0) proxmox.nodes(node).qemu(vm_ids["worker2"]).config.put(protection=0) log.info("Disabled VM delete-protection") log.info("Starting VMs...") tasks_wait_completion( proxmox, [ proxmox.nodes(node).qemu(vm_ids["master"]).status.start.post(), proxmox.nodes(node).qemu(vm_ids["worker1"]).status.start.post(), proxmox.nodes(node).qemu(vm_ids["worker2"]).status.start.post(), ], ) log.info("VMs are running!") log.info("Waiting for QEMU guest agent to start") done = False while not done: try: _ = proxmox.nodes(node).qemu(vm_ids["master"]).agent.ping.post() _ = proxmox.nodes(node).qemu(vm_ids["worker1"]).agent.ping.post() _ = proxmox.nodes(node).qemu(vm_ids["worker2"]).agent.ping.post() except ResourceException: time.sleep(1) except Exception as e: raise RuntimeError("Failed to ping QEMU agent") from e else: done = True log.info("QEMU guest agent is running") vm_ip_master: str = get_vm_ip(proxmox, node, vm_ids["master"]) vm_ip_worker1: str = get_vm_ip(proxmox, node, vm_ids["worker1"]) vm_ip_worker2: str = get_vm_ip(proxmox, node, vm_ids["worker2"]) tmp_path: str = "./tmp" if not os.path.exists(tmp_path): os.makedirs(tmp_path) with open(f"{tmp_path}/vm_ids", "w") as ofile: vm_ids_file = f'{vm_ids["master"]}\t{vm_names["master"]}\n' vm_ids_file += f'{vm_ids["worker1"]}\t{vm_names["worker1"]}\n' vm_ids_file += f'{vm_ids["worker2"]}\t{vm_names["worker2"]}\n' log.info(vm_ids_file) ofile.write(vm_ids_file) with open(f"{tmp_path}/hosts", "w") as ofile: hosts = f'{vm_ip_master}\t{vm_names["master"]}\n' hosts += f'{vm_ip_worker1}\t{vm_names["worker1"]}\n' hosts += f'{vm_ip_worker2}\t{vm_names["worker2"]}\n' log.info(hosts) ofile.write(hosts) inventory = { "workers": { "hosts": { "worker1": { "ansible_host": vm_ip_worker1, }, "worker2": { "ansible_host": vm_ip_worker2, }, }, }, "all_vms": { "hosts": { "master": { "ansible_host": vm_ip_master, }, }, "children": {"workers": None}, "vars": { "ansible_user": "root", "ansible_python_interpreter": "/usr/bin/python3", }, }, } with open(f"{tmp_path}/generated_inventory.yaml", "w") as ofile: yaml.dump(inventory, ofile) if __name__ == "__main__": main()