fredrick/dynamic-infra-tooling
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor: implement dynamic storage discovery and add unit tests

Browse source
This commit is contained in:
Fredrick Amnehagen 2026-02-05 22:31:05 +01:00
parent 4231253dc7
commit 5d61ef8116
2 changed files with 58 additions and 24 deletions
Download patch file Download diff file Expand all files Collapse all files

49
infra_cli/proxmox.py
View file

@ -2,11 +2,12 @@ from .ssh import SSHClient
class ProxmoxManager:
def __init__(self, config, node_name=None):
node = config.get_node(node_name)
self.config = config
self.node_name = node_name or 'la-vmh-11'
node = config.get_node(self.node_name)
if not node:
raise ValueError(f"Node '{node_name}' not found in config")
raise ValueError(f"Node '{self.node_name}' not found in configuration")
self.node_name = node_name
self.host = node['host']
self.password = node.get('pass')
self.user = config.get('proxmox.user', 'root')
@ -17,10 +18,6 @@ class ProxmoxManager:
res = self.client.run("pct list")
return res.stdout
def list_vms(self):
res = self.client.run("qm list")
return res.stdout
def status_lxc(self, vmid):
res = self.client.run(f"pct status {vmid}")
return res.stdout
@ -28,39 +25,43 @@ class ProxmoxManager:
def resolve_template(self, alias):
"""Resolves a template alias (e.g. 'debian-13') to the latest full path on the node"""
if alias == "debian-13":
# Search for debian-13 templates in all storages
# Search for debian-13 templates in all active storages
res = self.client.run("pvesm list --content vztmpl | grep 'debian-13-standard' | sort -V | tail -n 1")
if res.returncode == 0 and res.stdout.strip():
# Extract first column (Volid)
return res.stdout.split()[0]
return alias # Fallback to original if no resolution
return alias
def create_lxc(self, vmid, template, hostname, ip, gateway, bridge="vmbr0", storage="local-zfs", password=None):
# Resolve template if it's an alias
def discover_storage(self):
"""Finds the best storage for container rootfs (prefers ZFS or LVM-thin)"""
res = self.client.run("pvesm status | grep -E 'zfspool|lvmthin' | grep ' active' | head -n 1")
if res.returncode == 0 and res.stdout.strip():
return res.stdout.split()[0]
return "local" # Absolute fallback
def create_lxc(self, vmid, template, hostname, ip, gateway, bridge="vmbr0", storage=None, password=None):
full_template = self.resolve_template(template)
target_storage = storage or self.discover_storage()
# Professional creation command with sane defaults
cmd = f"pct create {vmid} {full_template} --hostname {hostname} " \
f"--net0 name=eth0,bridge={bridge},ip={ip},gw={gateway} " \
f"--storage {storage} --onboot 1"
f"--storage {target_storage} --onboot 1"
if password:
cmd += f" --password {password}"
res = self.client.run(cmd)
if res.returncode != 0:
raise RuntimeError(f"Failed to create LXC {vmid}: {res.stderr}")
raise RuntimeError(f"Failed to create LXC {vmid} on {self.node_name}: {res.stderr}")
# Start the container
self.client.run(f"pct start {vmid}")
return res.stdout
def start_lxc(self, vmid):
return self.client.run(f"pct start {vmid}")
def stop_lxc(self, vmid):
return self.client.run(f"pct stop {vmid}")
def delete_lxc(self, vmid):
res = self.client.run(f"pct destroy {vmid}")
# Stop first
self.client.run(f"pct stop {vmid}")
res = self.client.run(f"pct destroy {vmid} --purge")
if res.returncode != 0:
raise RuntimeError(f"Failed to destroy LXC {vmid}: {res.stderr}")
return res.stdout
return res.stdout