Christian ManivongandClaude Sonnet 4.6
b1ba991e6d
Implements set_hostname on ProxmoxSystemMixin: 1. Writes /etc/hostname (short name, base64-safe transfer) 2. Replaces old hostname in /etc/hosts via Python regex + base64 3. Updates /etc/mailname if present 4. Updates Postfix myhostname via postconf -e if installed 5. Applies hostname immediately at runtime via hostname(1) 6. Regenerates Proxmox node TLS certificate via pvecm updatecerts -f (falls back to pvenode cert create → pveproxy restart) Accepts bare hostname or FQDN. A reboot is required for the Proxmox node name to update in the web UI / cluster — the driver logs this. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
338 lines
14 KiB
Python
338 lines
14 KiB
Python
# 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)}
|