"""Обёртка над Proxmox REST API.""" import logging import random import re import string import time from proxmoxer import ProxmoxAPI from .config import settings logger = logging.getLogger(__name__) def _client() -> ProxmoxAPI: """Создаёт клиент Proxmox API, аутентифицированный по API-токену.""" host = settings.pve_host.replace("https://", "").replace("http://", "").split(":")[0] return ProxmoxAPI( host, user=settings.pve_token_name.split("!")[0], token_name=settings.pve_token_name.split("!")[1], token_value=settings.pve_token_value, verify_ssl=settings.pve_verify_ssl, ) def gen_password(length: int = 14) -> str: """Генерирует пароль из букв и цифр.""" alphabet = string.ascii_letters + string.digits return "".join(random.choice(alphabet) for _ in range(length)) def get_next_vmid() -> int: """Берёт следующий свободный VMID у Proxmox.""" return int(_client().cluster.nextid.get()) def clone_vm(source_vmid: int, new_vmid: int, name: str, node: str = None, storage: str = None) -> None: """Клонирует VM-шаблон в новую VM.""" node = node or settings.pve_node px = _client() params = {"newid": new_vmid, "name": name, "full": 1} if storage: params["storage"] = storage upid = px.nodes(node).qemu(source_vmid).clone.post(**params) _wait_task(px, node, upid) def resize_vm(vmid: int, cores: int, memory_mb: int, node: str = None) -> None: """Изменяет количество vCPU и RAM у VM.""" node = node or settings.pve_node _client().nodes(node).qemu(vmid).config.put(cores=cores, memory=memory_mb) def create_lxc( new_vmid: int, name: str, template_volid: str, cores: int, memory_mb: int, disk_gb: int, storage: str = "local-lvm", node: str = None, ) -> str: """Создаёт LXC из архивного шаблона и запускает его. Шаги (каждый — отдельный запрос к Proxmox API): 1. Создаёт LXC через ``ostemplate`` (без запуска). 2. Задаёт новый root-пароль через ``/passwd``. 3. Создаёт rootfs указанного размера через ``/resize``. 4. Запускает контейнер через ``/status/start``. LXC создаётся ТОЛЬКО из архивного шаблона (vztmpl на storage). Клонирование существующего LXC по VMID не поддерживается. """ node = node or settings.pve_node px = _client() password = gen_password() # 1. Создаём LXC через ostemplate. start=0 — не запускаем, чтобы успеть # задать пароль и диск нужного размера до старта. upid = px.nodes(node).lxc.post( vmid=new_vmid, hostname=name, ostemplate=template_volid, cores=cores, memory=memory_mb, swap=memory_mb, net0="name=eth0,bridge=vmbr0,ip=dhcp", unprivileged=1, password=password, start=0, ) _wait_task(px, node, upid) # 2. Задаём новый root-пароль через отдельный endpoint /passwd. # Поле password в POST /lxc на некоторых версиях Proxmox игнорируется, # поэтому дублируем через /passwd — это идемпотентная операция. px.nodes(node).lxc(new_vmid).passwd.post(password=password) # 3. Создаём rootfs указанного размера (storage:size_G). px.nodes(node).lxc(new_vmid).resize.put( disk="rootfs", size=f"{storage}:{disk_gb}", ) # 4. Запускаем контейнер. upid = px.nodes(node).lxc(new_vmid).status.start.post() if upid: _wait_task(px, node, upid) return password def _wait_task(px: ProxmoxAPI, node: str, upid: str, timeout: int = 1200, poll: float = 2.0) -> None: """Ждёт завершения асинхронной задачи Proxmox.""" start = time.time() delay = poll while time.time() - start < timeout: status = px.nodes(node).tasks(upid).status.get() if status.get("status") == "stopped": if status.get("exitstatus") != "OK": raise RuntimeError(f"Задача Proxmox завершилась с ошибкой: {status}") return time.sleep(delay) delay = min(delay * 1.5, 5.0) raise TimeoutError("Превышено время ожидания задачи Proxmox") _DISK_KEYS = ("scsi0", "virtio0", "ide0", "sata0") def _disk_size_from_config(cfg: dict) -> int: """Извлекает размер диска VM в ГБ.""" for key in _DISK_KEYS: value = cfg.get(key, "") match = re.search(r"size=(\d+)G", value) if match: return int(match.group(1)) return 10 def list_vm_templates(node: str = None) -> list: """Возвращает список VM-шаблонов.""" node = node or settings.pve_node px = _client() result = [] for vm in px.nodes(node).qemu.get(): if vm.get("template") == 1: cfg = px.nodes(node).qemu(vm["vmid"]).config.get() result.append({ "vmid": vm["vmid"], "name": vm.get("name", ""), "cores": int(cfg.get("cores", 1)), "memory_mb": int(cfg.get("memory", 1024)), "disk_gb": _disk_size_from_config(cfg), }) return result def list_lxc_templates(node: str = None) -> list: """Возвращает архивные LXC-шаблоны (vztmpl) из доступных хранилищ.""" node = node or settings.pve_node px = _client() result = [] for storage in px.nodes(node).storage.get(): if "vztmpl" not in storage.get("content", ""): continue try: for item in px.nodes(node).storage(storage["storage"]).content.get(): if item.get("content") != "vztmpl": continue name = item["volid"].split("/")[-1] for extension in (".tar.zst", ".tar.gz", ".tar.xz"): name = name.replace(extension, "") result.append({ "volid": item["volid"], "name": name, "storage": storage["storage"], "size_mb": round(item.get("size", 0) / (1024 ** 2), 1), }) except Exception as exc: logger.warning("Не удалось прочитать шаблоны с хранилища %s: %s", storage["storage"], exc) return result def resize_disk(vmid: int, disk_gb: int, node: str = None) -> None: """Увеличивает диск VM до указанного размера.""" node = node or settings.pve_node px = _client() config = px.nodes(node).qemu(vmid).config.get() for key in _DISK_KEYS: value = config.get(key, "") if not value: continue match = re.search(r"size=(\d+)G", value) current = int(match.group(1)) if match else 0 if disk_gb > current: px.nodes(node).qemu(vmid).resize.put(disk=key, size=f"+{disk_gb - current}G") return logger.warning("У VM %s не найден диск для resize", vmid) def configure_cloud_init(vmid: int, ciuser: str, cipassword: str, node: str = None) -> None: """Задаёт cloud-init user/password и сеть DHCP.""" node = node or settings.pve_node _client().nodes(node).qemu(vmid).config.put( ciuser=ciuser, cipassword=cipassword, ipconfig0="ip=dhcp", ) def get_instance_ip(vmid: int, node: str = None) -> str: """Возвращает основной IPv4 инстанса.""" node = node or settings.pve_node px = _client() try: ifaces = px.nodes(node).qemu(vmid).agent.get("network-get-interfaces").get("result", []) for iface in ifaces: if iface.get("name") == "lo": continue for address in iface.get("ip-addresses", []) or []: ip = address.get("ip-address", "") if address.get("ip-address-type") == "ipv4" and not ip.startswith("127."): return ip except Exception as exc: logger.debug("VM %s: guest-agent недоступен (%s)", vmid, exc) try: for iface in px.nodes(node).lxc(vmid).interfaces.get() or []: for ip_data in iface.get("ip-addresses", []) or []: ip = ip_data.get("ip-address", "") if not ip.startswith("127."): return ip except Exception as exc: logger.debug("LXC %s: интерфейсы недоступны (%s)", vmid, exc) return "не определён (агент недоступен)" def get_live_stats(vmid: int, node: str = None) -> dict: """Возвращает текущие показатели VM или LXC.""" node = node or settings.pve_node px = _client() def shape(status: dict) -> dict: return { "cpu": round(status.get("cpu", 0) * 100, 1), "mem_used": status.get("mem", 0), "mem_total": status.get("maxmem", 0), "uptime": status.get("uptime", 0), "status": status.get("status", "unknown"), } for endpoint in (px.nodes(node).qemu(vmid).status.current, px.nodes(node).lxc(vmid).status.current): try: return shape(endpoint.get()) except Exception: continue return {"cpu": 0, "mem_used": 0, "mem_total": 0, "uptime": 0, "status": "unknown"} _VALID_ACTIONS = {"start", "stop", "shutdown", "reboot"} def guest_action(guest_type: str, vmid: int, action: str, node: str = None) -> None: """Выполняет действие над VM/LXC.""" if action not in _VALID_ACTIONS: raise ValueError(f"Неизвестное действие: {action}") node = node or settings.pve_node px = _client() endpoint = px.nodes(node).qemu(vmid) if guest_type == "vm" else px.nodes(node).lxc(vmid) getattr(endpoint.status, action).post() def _wait_guest_stopped(px: ProxmoxAPI, guest_type: str, vmid: int, node: str, timeout: int = 180) -> None: """Ожидает, пока VM или LXC перейдёт в состояние stopped.""" endpoint = px.nodes(node).qemu(vmid) if guest_type == "vm" else px.nodes(node).lxc(vmid) deadline = time.time() + timeout while time.time() < deadline: status = endpoint.status.current.get().get("status") if status == "stopped": return time.sleep(2) raise TimeoutError(f"Инстанс {vmid} не остановился за {timeout} секунд") def delete_guest(guest_type: str, vmid: int, node: str = None) -> None: """Останавливает работающий инстанс и удаляет его.""" node = node or settings.pve_node px = _client() endpoint = px.nodes(node).qemu(vmid) if guest_type == "vm" else px.nodes(node).lxc(vmid) status = endpoint.status.current.get().get("status") if status == "running": try: shutdown_upid = endpoint.status.shutdown.post(timeout=60) if shutdown_upid: _wait_task(px, node, shutdown_upid, timeout=120) except Exception as exc: logger.warning("Мягкая остановка %s %s не удалась: %s", guest_type, vmid, exc) endpoint.status.stop.post() _wait_guest_stopped(px, guest_type, vmid, node) upid = endpoint.delete() if upid: _wait_task(px, node, upid) def get_status(guest_type: str, vmid: int, node: str = None) -> dict: """Возвращает текущий статус VM/LXC.""" node = node or settings.pve_node px = _client() if guest_type == "vm": return px.nodes(node).qemu(vmid).status.current.get() return px.nodes(node).lxc(vmid).status.current.get()