# Copyright 2025 The NetOrk Project Authors. All rights reserved. # # Licensed under the Apache License, Version 2.0 (the "License"); # you may not use this file except in compliance with the License. # You may obtain a copy of the License at # # http://www.apache.org/licenses/LICENSE-2.0 # # Unless required by applicable law or agreed to in writing, software # distributed under the License is distributed on an "AS IS" BASIS, # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. # See the License for the specific language governing permissions and # limitations under the License. """VM/container NAPALM getters for Proxmox VE nodes.""" from __future__ import annotations import logging import re from typing import Any from napalm_proxmox import utils logger = logging.getLogger(__name__) _JsonDict = dict[str, Any] class ProxmoxVMMixin: """Mixin providing VM and container NAPALM methods.""" def get_vm_interfaces( self, vmid: int, vm_type: str ) -> tuple[dict[str, _JsonDict], bool, bool]: """Return network interfaces for a single VM or LXC container. Returns a 3-tuple ``(interfaces, agent_running, agent_enabled)``: * ``interfaces`` – dict keyed by interface name, each with NAPALM-compatible fields plus ``ipv4``, ``bridge``, and ``tag``. * ``agent_running`` – True if the QEMU Guest Agent responded during this call (always False for LXC). * ``agent_enabled`` – True if the QEMU Guest Agent is enabled in the VM's Proxmox config (always False for LXC). LXC : uses ``/nodes/{node}/lxc/{vmid}/interfaces`` + LXC config QEMU : tries QEMU guest agent first, falls back to VM config parsing. In both paths the VM config is fetched to derive bridge/tag. """ _NET_MODELS = {"virtio", "e1000", "e1000e", "vmxnet3", "rtl8139", "ne2k_pci"} def _parse_net_entry(val_str: str) -> tuple[str, str, int | None]: """Parse a Proxmox net config value → (mac_upper, bridge, tag|None).""" mac = bridge = "" tag: int | None = None for part in str(val_str).split(","): if "=" not in part: continue k, v = part.split("=", 1) k = k.strip().lower() if k in _NET_MODELS: mac = v.strip() elif k == "bridge": bridge = v.strip() elif k == "tag": try: tag = int(v.strip()) except ValueError: pass return mac.upper() if mac else "", bridge, tag interfaces: dict[str, _JsonDict] = {} if vm_type == "container": # Build iface_name → (bridge, tag) from LXC config # LXC net entries look like: net0=name=eth0,bridge=vmbr40,tag=40,... lxc_net_map: dict[str, tuple[str, int | None]] = {} try: config = self._node_api().lxc(vmid).config.get() or {} net_re = re.compile(r"^net(\d+)$") for key, val in config.items(): if not net_re.match(key): continue iface_name = "" bridge = "" tag: int | None = None for part in str(val).split(","): if "=" not in part: continue k, v = part.split("=", 1) k = k.strip().lower() if k == "name": iface_name = v.strip() elif k == "bridge": bridge = v.strip() elif k == "tag": try: tag = int(v.strip()) except ValueError: pass if iface_name: lxc_net_map[iface_name] = (bridge, tag) except Exception as exc: logger.debug("get_vm_interfaces: LXC %s config failed: %s", vmid, exc) try: for iface in (self._node_api().lxc(vmid).interfaces.get() or []): name = iface.get("name", "") if not name or name == "lo": continue mac = iface.get("hwaddr", "") ipv4 = "" inet = iface.get("inet", "") if inet: ipv4 = inet.split("/")[0] bridge, tag = lxc_net_map.get(name, ("", None)) interfaces[name] = { "is_up": True, "is_enabled": True, "description": bridge, "mac_address": mac.upper() if mac else "", "speed": -1, "mtu": 1500, "last_flapped": -1.0, "ipv4": ipv4, "bridge": bridge, "tag": tag, } except Exception as exc: logger.debug("get_vm_interfaces: LXC %s ifaces failed: %s", vmid, exc) # LXC containers do not use QEMU Guest Agent return interfaces, False, False else: # QEMU: pre-fetch VM config to build MAC → (bridge, tag) map and # to check whether the QEMU Guest Agent is enabled. mac_to_net: dict[str, tuple[str, int | None]] = {} # mac_upper → (bridge, tag) net_idx_map: dict[str, tuple[str, str, int | None]] = {} # "netN" → (mac, bridge, tag) agent_enabled = False try: config = self._node_api().qemu(vmid).config.get() or {} # Proxmox stores the agent setting as agent=1, agent=0, or # agent=enabled=1[,fstrim_cloned_disks=1,...] raw_agent = str(config.get("agent", "0")) # Treat any truthy value ("1", "enabled=1", ...) as enabled agent_enabled = bool( raw_agent.strip() in ("1", "true") or raw_agent.startswith("enabled=1") or raw_agent.startswith("1,") ) net_re = re.compile(r"^net(\d+)$") for key, val in config.items(): m = net_re.match(key) if not m: continue mac, bridge, tag = _parse_net_entry(val) iface_key = f"net{m.group(1)}" net_idx_map[iface_key] = (mac, bridge, tag) if mac: mac_to_net[mac] = (bridge, tag) except Exception as exc: logger.debug("get_vm_interfaces: QEMU %s config fetch failed: %s", vmid, exc) # Try guest agent first agent_ok = False try: agent_result = self._node_api().qemu(vmid).agent("network-get-interfaces").get() for iface in (agent_result or {}).get("result", []): name = iface.get("name", "") if not name or name == "lo": continue mac = (iface.get("hardware-address", "") or "").upper() ipv4 = "" for addr in iface.get("ip-addresses", []): if addr.get("ip-address-type") == "ipv4": ipv4 = addr.get("ip-address", "") break bridge, tag = mac_to_net.get(mac, ("", None)) interfaces[name] = { "is_up": True, "is_enabled": True, "description": bridge, "mac_address": mac, "speed": -1, "mtu": 1500, "last_flapped": -1.0, "ipv4": ipv4, "bridge": bridge, "tag": tag, } agent_ok = bool(interfaces) except Exception as exc: logger.debug("QEMU guest agent network-get-interfaces failed: %s", exc) if not agent_ok: # Fall back to config-only (gives MAC + bridge + tag, no IP) for iface_key, (mac, bridge, tag) in net_idx_map.items(): interfaces[iface_key] = { "is_up": False, "is_enabled": True, "description": bridge, "mac_address": mac, "speed": -1, "mtu": 1500, "last_flapped": -1.0, "ipv4": "", "bridge": bridge, "tag": tag, } return interfaces, agent_ok, agent_enabled def get_vms(self) -> list[_JsonDict]: """Return all VMs (QEMU) and containers (LXC) on this node. Each entry contains: * vmid (int) - Proxmox VM/container ID * name (str) - display name * type (str) - ``"vm"`` or ``"container"`` * status (str) - ``"running"``, ``"stopped"``, etc. * vcpus (int) - allocated vCPUs * memory (int) - configured RAM in megabytes * cpu_usage (float) - current CPU utilisation 0.0–1.0 (from last stats cycle) * memory_usage (int)- current RSS in megabytes * uptime (int) - uptime in seconds (0 if stopped) * node (str) - cluster node name * interfaces (dict) - network interfaces (NAPALM format + ipv4 field) * ipv4 (str) - primary IPv4 address (empty string if unknown) """ result: list[_JsonDict] = [] # QEMU VMs try: for vm in (self._node_api().qemu.get() or []): vmid = int(vm.get("vmid", 0)) name = vm.get("name", f"vm-{vmid}") status = vm.get("status", "unknown") cpu_usage = float(vm.get("cpu", 0.0) or 0.0) uptime = int(vm.get("uptime", 0) or 0) # mem/maxmem are in bytes maxmem_bytes = int(vm.get("maxmem", 0) or 0) mem_bytes = int(vm.get("mem", 0) or 0) memory_mb = maxmem_bytes // (1024 * 1024) memory_usage_mb = mem_bytes // (1024 * 1024) # vcpus can be in "cpus" key for running VMs vcpus = int(vm.get("cpus", vm.get("vcpus", 0)) or 0) interfaces, agent_running, agent_enabled = self.get_vm_interfaces(vmid, "vm") ipv4 = next( (iface["ipv4"] for iface in interfaces.values() if iface.get("ipv4")), "", ) result.append({ "vmid": vmid, "name": name, "type": "vm", "status": status, "vcpus": vcpus, "memory": memory_mb, "cpu_usage": round(cpu_usage, 4), "memory_usage": memory_usage_mb, "uptime": uptime, "node": self._node_name, "interfaces": interfaces, "ipv4": ipv4, "agent_enabled": agent_enabled, "agent_running": agent_running, }) except Exception as exc: logger.warning("get_vms: failed to list QEMU VMs: %s", exc) # LXC containers try: for ct in (self._node_api().lxc.get() or []): vmid = int(ct.get("vmid", 0)) name = ct.get("name", f"ct-{vmid}") status = ct.get("status", "unknown") cpu_usage = float(ct.get("cpu", 0.0) or 0.0) uptime = int(ct.get("uptime", 0) or 0) maxmem_bytes = int(ct.get("maxmem", 0) or 0) mem_bytes = int(ct.get("mem", 0) or 0) memory_mb = maxmem_bytes // (1024 * 1024) memory_usage_mb = mem_bytes // (1024 * 1024) vcpus = int(ct.get("cpus", 0) or 0) interfaces, agent_running, agent_enabled = self.get_vm_interfaces(vmid, "container") ipv4 = next( (iface["ipv4"] for iface in interfaces.values() if iface.get("ipv4")), "", ) result.append({ "vmid": vmid, "name": name, "type": "container", "status": status, "vcpus": vcpus, "memory": memory_mb, "cpu_usage": round(cpu_usage, 4), "memory_usage": memory_usage_mb, "uptime": uptime, "node": self._node_name, "interfaces": interfaces, "ipv4": ipv4, "agent_enabled": agent_enabled, "agent_running": agent_running, }) except Exception as exc: logger.warning("get_vms: failed to list LXC containers: %s", exc) return sorted(result, key=lambda x: x["vmid"]) _POWER_ACTIONS_VM = {'start', 'stop', 'shutdown', 'reboot', 'reset'} _POWER_ACTIONS_CT = {'start', 'stop', 'shutdown', 'reboot'} def power_vm(self, vmid: int, vm_type: str, action: str) -> dict: """Send a power action to a VM or container on this node. Supported actions for VMs: start, stop, shutdown, reboot, reset Supported actions for containers: start, stop, shutdown, reboot """ allowed = self._POWER_ACTIONS_VM if vm_type == 'vm' else self._POWER_ACTIONS_CT if action not in allowed: return {"success": False, "error": f"Action '{action}' not supported for {vm_type} (allowed: {sorted(allowed)})"} try: vm_api = self._node_api().qemu(vmid) if vm_type == 'vm' else self._node_api().lxc(vmid) task_id = getattr(vm_api.status, action).post() return {"success": True, "task_id": task_id or ""} except Exception as exc: return {"success": False, "error": str(exc)}