import uuid
from pathlib import Path
from typing import Optional, Iterable
import libvirt
from libvirt import virConnect, virDomain, virStoragePool
from cloud_config import gen_cloud_config, _gen_cloudinit_iso_image
def _gen_libvirt_volume_config(name: str, path: Path, backing_image: Path, capacity: int = 1000,
image_format: str = 'qcow2') -> str:
"""
creates the XML config for a volume that will hold the data of a vm.
:param name: volume name
:param path: path to save volume to
:param backing_image: path to a backing image to base this one on
:param capacity: capacity in MiB
:param image_format: qcow2, iso, raw or alike.
A list of valid values can be found here: https://libvirt.org/storage.html#StorageBackendFS
:return: XML config as string
"""
xml_config = f"""
{name}.{image_format}{capacity}0{path}{backing_image}
"""
return xml_config
def _gen_libvirt_storage_pool_config(name: str, path: Path) -> str:
"""
create the libvirt XML config for a storage pool with directory backend.
:param name: name of the storage pool
:param path: the path to save images in the pool to
:return: XML config as string
"""
xml_config = f"""
{name}{path}
"""
return xml_config
def _gen_libvirt_config(name: str, hdd_path: Path, cdrom_path: Path, cpu: int = 1, mem: int = 512) -> str:
"""
return libvirt XML config for a virtual machine
:param name: name of the machine
:param hdd_path: path to the image to use as primary disk
:param cdrom_path: path to the cloud-init image
:param cpu: number of cpus
:param mem: amount of ram in MiB
:return: XML config as string
"""
vm_uuid = uuid.uuid4()
xml_config = f"""
{name}{vm_uuid}{mem}{mem}{cpu}hvmdestroyrestartdestroy/usr/bin/qemu-system-x86_64/dev/urandom
"""
return xml_config
def spawn_vm(connection: virConnect,
name: str,
storage_pool_path: Path,
wireguard_config: str,
ssh_authorized_keys: Optional[Iterable[str]] = None,
storage_pool_name: str = 'wireguard_test'
) -> virDomain:
"""
spawm a VM provisioned via cloud-init that is a mesh node
:param connection: libvirt connection object
:param name: name of the VM
:param storage_pool_path: path storage pool objects are saved in
:param wireguard_config: wireguard config file as string
:param ssh_authorized_keys: ssh public keys as list to add to the image
:param storage_pool_name: name of the storage pool (will be created if nonexistent)
:return: a libvirt virDomain object representing a virtual machine
"""
# if storage pool does not exist, create it. Otherwise, fetch it.
storage_pools = (p.name() for p in connection.listAllStoragePools())
if storage_pool_name not in storage_pools:
storage_pool_path.mkdir(exist_ok=True)
storage_pool_xml_config = _gen_libvirt_storage_pool_config(storage_pool_name, storage_pool_path)
storage_pool: virStoragePool = connection.storagePoolCreateXML(storage_pool_xml_config)
else:
storage_pool = connection.storagePoolLookupByName(storage_pool_name)
image_format = 'qcow2'
volume_path = storage_pool_path / f'{name}.{image_format}'
# if image exists, fetch and wipe it. Otherwise, create it.
if str(volume_path) in (v.path() for v in storage_pool.listAllVolumes()):
volume = storage_pool.storageVolLookupByName(volume_path.name)
volume.wipe()
else:
backing_file_path = Path('/tmp/focal-server-cloudimg-amd64.qcow2')
volume_xml_config = _gen_libvirt_volume_config(name, volume_path, backing_file_path, image_format=image_format)
# volume_create_flags = libvirt.VIR_STORAGE_VOL_CREATE_PREALLOC_METADATA
storage_pool.createXML(volume_xml_config)
# if the cloud-init images exist, delete them.
cloud_init_volume_name = f'cloudinit-{name}.iso'
try:
cloud_init_volume = storage_pool.storageVolLookupByName(cloud_init_volume_name)
cloud_init_volume.delete()
except libvirt.libvirtError:
pass
# recreate cloud-init images
cloud_init_iso_path = storage_pool_path / cloud_init_volume_name
user_data_path, meta_data_path = gen_cloud_config(name, Path('/tmp/'), wireguard_config, ssh_authorized_keys)
_gen_cloudinit_iso_image(cloud_init_iso_path, user_data_path, meta_data_path)
# generate VM XML config
vm_xml_config = _gen_libvirt_config(
name,
volume_path,
cloud_init_iso_path,
cpu=1,
mem=512
)
vm_create_flags = libvirt.VIR_DOMAIN_START_PAUSED
# create the virtual machine
# might use defineXML for persistence. VMs disappear after shutdown
vm = connection.createXML(vm_xml_config, flags=vm_create_flags)
return vm