# QEMU Manual # https://qemu.weilnetz.de/doc/qemu-doc.html # For QEMU Monitor Protocol Commands Information, See # https://qemu.weilnetz.de/doc/qemu-doc.html#pcsys_005fmonitor import os import random import subprocess as sp import tempfile import time from functools import wraps from os.path import join as join_path from string import Template from typing import Union import bitmath import sshtunnel from common.helpers import get_ipv6_address from common.request import RequestEntry, RequestType from common.vm import VMEntry, VMStatus from config import etcd_client, request_pool, running_vms, vm_pool, env_vars, image_storage_handler from . import qmp from host import logger class VM: def __init__(self, key, handle, vnc_socket_file): self.key = key # type: str self.handle = handle # type: qmp.QEMUMachine self.vnc_socket_file = vnc_socket_file # type: tempfile.NamedTemporaryFile def __repr__(self): return "VM({})".format(self.key) def create_dev(script, _id, dev, ip=None): command = [script, _id, dev] if ip: command.append(ip) try: output = sp.check_output(command, stderr=sp.PIPE) except Exception as e: print(e.stderr) return None else: return output.decode("utf-8").strip() def create_vxlan_br_tap(_id, _dev, ip=None): network_script_base = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'network') vxlan = create_dev(script=os.path.join(network_script_base, 'create-vxlan.sh'), _id=_id, dev=_dev) if vxlan: bridge = create_dev(script=os.path.join(network_script_base, 'create-bridge.sh'), _id=_id, dev=vxlan, ip=ip) if bridge: tap = create_dev(script=os.path.join(network_script_base, 'create-tap.sh'), _id=str(random.randint(1, 100000)), dev=bridge) if tap: return tap def random_bytes(num=6): return [random.randrange(256) for _ in range(num)] def generate_mac(uaa=False, multicast=False, oui=None, separator=':', byte_fmt='%02x'): mac = random_bytes() if oui: if type(oui) == str: oui = [int(chunk) for chunk in oui.split(separator)] mac = oui + random_bytes(num=6 - len(oui)) else: if multicast: mac[0] |= 1 # set bit 0 else: mac[0] &= ~1 # clear bit 0 if uaa: mac[0] &= ~(1 << 1) # clear bit 1 else: mac[0] |= 1 << 1 # set bit 1 return separator.join(byte_fmt % b for b in mac) def update_radvd_conf(etcd_client): network_script_base = os.path.join(os.path.dirname(os.path.dirname(__file__)), 'network') networks = { net.value['ipv6']: net.value['id'] for net in etcd_client.get_prefix('/v1/network/', value_in_json=True) if net.value.get('ipv6') } radvd_template = open(os.path.join(network_script_base, 'radvd-template.conf'), 'r').read() radvd_template = Template(radvd_template) content = [radvd_template.safe_substitute(bridge='br{}'.format(networks[net]), prefix=net) for net in networks if networks.get(net)] with open('/etc/radvd.conf', 'w') as radvd_conf: radvd_conf.writelines(content) sp.check_output(['systemctl', 'restart', 'radvd']) def get_start_command_args(vm_entry, vnc_sock_filename: str, migration=False, migration_port=None): threads_per_core = 1 vm_memory = int(bitmath.parse_string_unsafe(vm_entry.specs["ram"]).to_MB()) vm_cpus = int(vm_entry.specs["cpu"]) vm_uuid = vm_entry.uuid vm_networks = vm_entry.network command = "-name {}_{}".format(vm_entry.owner, vm_entry.name) command += " -drive file={},format=raw,if=virtio,cache=none".format( image_storage_handler.qemu_path_string(vm_uuid) ) command += " -device virtio-rng-pci -vnc unix:{}".format(vnc_sock_filename) command += " -m {} -smp cores={},threads={}".format( vm_memory, vm_cpus, threads_per_core ) if migration: command += " -incoming tcp:[::]:{}".format(migration_port) tap = None for network_and_mac in vm_networks: network_name, mac = network_and_mac _key = os.path.join(env_vars.get('NETWORK_PREFIX'), vm_entry.owner, network_name) network = etcd_client.get(_key, value_in_json=True) network_type = network.value["type"] network_id = str(network.value["id"]) network_ipv6 = network.value["ipv6"] if network_type == "vxlan": tap = create_vxlan_br_tap(network_id, env_vars.get("VXLAN_PHY_DEV"), network_ipv6) update_radvd_conf(etcd_client) command += " -netdev tap,id=vmnet{net_id},ifname={tap},script=no,downscript=no" \ " -device virtio-net-pci,netdev=vmnet{net_id},mac={mac}" \ .format(tap=tap, net_id=network_id, mac=mac) return command.split(" ") def create_vm_object(vm_entry, migration=False, migration_port=None): # NOTE: If migration suddenly stop working, having different # VNC unix filename on source and destination host can # be a possible cause of it. # REQUIREMENT: Use Unix Socket instead of TCP Port for VNC vnc_sock_file = tempfile.NamedTemporaryFile() qemu_args = get_start_command_args( vm_entry=vm_entry, vnc_sock_filename=vnc_sock_file.name, migration=migration, migration_port=migration_port, ) qemu_machine = qmp.QEMUMachine("/usr/bin/qemu-system-x86_64", args=qemu_args) return VM(vm_entry.key, qemu_machine, vnc_sock_file) def get_vm(vm_list: list, vm_key) -> Union[VM, None]: return next((vm for vm in vm_list if vm.key == vm_key), None) def need_running_vm(func): @wraps(func) def wrapper(e): vm = get_vm(running_vms, e.key) if vm: try: status = vm.handle.command("query-status") logger.debug("VM Status Check - %s", status) except Exception as exception: logger.info("%s failed - VM %s %s", func.__name__, e, exception) else: return func(e) return None else: logger.info("%s failed because VM %s is not running", func.__name__, e.key) return None return wrapper def create(vm_entry: VMEntry): if image_storage_handler.is_vm_image_exists(vm_entry.uuid): # File Already exists. No Problem Continue logger.debug("Image for vm %s exists", vm_entry.uuid) else: vm_hdd = int(bitmath.parse_string_unsafe(vm_entry.specs["os-ssd"]).to_MB()) if image_storage_handler.make_vm_image(src=vm_entry.image_uuid, dest=vm_entry.uuid): if not image_storage_handler.resize_vm_image(path=vm_entry.uuid, size=vm_hdd): vm_entry.status = VMStatus.error else: logger.info("New VM Created") def start(vm_entry: VMEntry, destination_host_key=None, migration_port=None): _vm = get_vm(running_vms, vm_entry.key) # VM already running. No need to proceed further. if _vm: logger.info("VM %s already running" % vm_entry.uuid) return else: logger.info("Trying to start %s" % vm_entry.uuid) if destination_host_key: launch_vm(vm_entry, migration=True, migration_port=migration_port, destination_host_key=destination_host_key) else: create(vm_entry) launch_vm(vm_entry) @need_running_vm def stop(vm_entry): vm = get_vm(running_vms, vm_entry.key) vm.handle.shutdown() if not vm.handle.is_running(): vm_entry.add_log("Shutdown successfully") vm_entry.declare_stopped() vm_pool.put(vm_entry) running_vms.remove(vm) def delete(vm_entry): logger.info("Deleting VM | %s", vm_entry) stop(vm_entry) r_status = image_storage_handler.delete_vm_image(vm_entry.uuid) if r_status: etcd_client.client.delete(vm_entry.key) def transfer(request_event): # This function would run on source host i.e host on which the vm # is running initially. This host would be responsible for transferring # vm state to destination host. _host, _port = request_event.parameters["host"], request_event.parameters["port"] _uuid = request_event.uuid _destination = request_event.destination_host_key vm = get_vm(running_vms, join_path(env_vars.get('VM_PREFIX'), _uuid)) if vm: tunnel = sshtunnel.SSHTunnelForwarder( _host, ssh_username=env_vars.get("ssh_username"), ssh_pkey=env_vars.get("ssh_pkey"), remote_bind_address=("127.0.0.1", _port), ssh_proxy_enabled=True, ssh_proxy=(_host, 22) ) try: tunnel.start() except sshtunnel.BaseSSHTunnelForwarderError: logger.exception("Couldn't establish connection to (%s, 22)", _host) else: vm.handle.command( "migrate", uri="tcp:0.0.0.0:{}".format(tunnel.local_bind_port) ) status = vm.handle.command("query-migrate")["status"] while status not in ["failed", "completed"]: time.sleep(2) status = vm.handle.command("query-migrate")["status"] with vm_pool.get_put(request_event.uuid) as source_vm: if status == "failed": source_vm.add_log("Migration Failed") elif status == "completed": # If VM is successfully migrated then shutdown the VM # on this host and update hostname to destination host key source_vm.add_log("Successfully migrated") source_vm.hostname = _destination running_vms.remove(vm) vm.handle.shutdown() source_vm.in_migration = False # VM transfer finished finally: tunnel.close() def launch_vm(vm_entry, migration=False, migration_port=None, destination_host_key=None): logger.info("Starting %s" % vm_entry.key) vm = create_vm_object(vm_entry, migration=migration, migration_port=migration_port) try: vm.handle.launch() except Exception: logger.exception("Error Occured while starting VM") vm.handle.shutdown() if migration: # We don't care whether MachineError or any other error occurred pass else: # Error during typical launch of a vm vm.handle.shutdown() vm_entry.declare_killed() vm_pool.put(vm_entry) else: vm_entry.vnc_socket = vm.vnc_socket_file.name running_vms.append(vm) if migration: vm_entry.in_migration = True r = RequestEntry.from_scratch( type=RequestType.TransferVM, hostname=vm_entry.hostname, parameters={"host": get_ipv6_address(), "port": migration_port}, uuid=vm_entry.uuid, destination_host_key=destination_host_key, request_prefix=env_vars.get("REQUEST_PREFIX") ) request_pool.put(r) else: # Typical launching of a vm vm_entry.status = VMStatus.running vm_entry.add_log("Started successfully") vm_pool.put(vm_entry)