uncloud/ucloud/host/virtualmachine.py

# 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 subprocess as sp
import ipaddress

from string import Template
from os.path import join as join_path

from ucloud.common.request import RequestEntry, RequestType
from ucloud.common.vm import VMStatus, declare_stopped
from ucloud.common.network import create_dev, delete_network_interface
from ucloud.common.schemas import VMSchema, NetworkSchema
from ucloud.host import logger
from ucloud.shared import shared
from ucloud.settings import settings
from ucloud.vmm import VMM

from marshmallow import ValidationError


class VM:
    def __init__(self, vm_entry):
        self.schema = VMSchema()
        self.vmm = VMM()
        self.key = vm_entry.key
        try:
            self.vm = self.schema.loads(vm_entry.value)
        except ValidationError:
            logger.exception(
                "Couldn't validate VM Entry", vm_entry.value
            )
            self.vm = None
        else:
            self.uuid = vm_entry.key.split("/")[-1]
            self.host_key = self.vm["hostname"]

    def get_qemu_args(self):
        command = (
            "-name {owner}_{name}"
            " -drive file={file},format=raw,if=virtio,cache=none"
            " -device virtio-rng-pci"
            " -m {memory} -smp cores={cores},threads={threads}"
        ).format(
            owner=self.vm["owner"],
            name=self.vm["name"],
            memory=int(self.vm["specs"]["ram"].to_MB()),
            cores=self.vm["specs"]["cpu"],
            threads=1,
            file=shared.storage_handler.qemu_path_string(self.uuid),
        )

        return command.split(" ")

    def start(self, destination_host_key=None):
        migration = False
        if destination_host_key:
            migration = True

        self.create()
        try:
            network_args = self.create_network_dev()
        except Exception as err:
            declare_stopped(self.vm)
            self.vm["log"].append("Cannot Setup Network Properly")
            logger.error(
                "Cannot Setup Network Properly for vm %s",
                self.uuid,
                exc_info=err,
            )
        else:
            self.vmm.start(
                uuid=self.uuid,
                migration=migration,
                *self.get_qemu_args(),
                *network_args
            )

            status = self.vmm.get_status(self.uuid)
            if status == "running":
                self.vm["status"] = VMStatus.running
                self.vm["vnc_socket"] = self.vmm.get_vnc(self.uuid)
            elif status == "inmigrate":
                r = RequestEntry.from_scratch(
                    type=RequestType.TransferVM,  # Transfer VM
                    hostname=self.host_key,  # Which VM should get this request. It is source host
                    uuid=self.uuid,  # uuid of VM
                    destination_sock_path=join_path(
                        self.vmm.socket_dir, self.uuid
                    ),
                    destination_host_key=destination_host_key,  # Where source host transfer VM
                    request_prefix=settings["etcd"]["request_prefix"],
                )
                shared.request_pool.put(r)
            else:
                self.stop()
                declare_stopped(self.vm)

        self.sync()

    def stop(self):
        self.vmm.stop(self.uuid)
        self.delete_network_dev()
        declare_stopped(self.vm)
        self.sync()

    def migrate(self, destination_host, destination_sock_path):
        self.vmm.transfer(
            src_uuid=self.uuid,
            destination_sock_path=destination_sock_path,
            host=destination_host,
        )

    def create_network_dev(self):
        command = ""
        for network_mac_and_tap in self.vm["network"]:
            network_name, mac, tap = network_mac_and_tap

            _key = os.path.join(
                settings["etcd"]["network_prefix"],
                self.vm["owner"],
                network_name,
            )
            network = shared.etcd_client.get(_key, value_in_json=True)
            network_schema = NetworkSchema()
            try:
                network = network_schema.load(network.value)
            except ValidationError:
                continue

            if network["type"] == "vxlan":
                tap = create_vxlan_br_tap(
                    _id=network["id"],
                    _dev=settings["network"]["vxlan_phy_dev"],
                    tap_id=tap,
                    ip=network["ipv6"],
                )

                all_networks = shared.etcd_client.get_prefix(
                    settings["etcd"]["network_prefix"],
                    value_in_json=True,
                )

                if ipaddress.ip_network(network["ipv6"]).is_global:
                    update_radvd_conf(all_networks)

            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 delete_network_dev(self):
        try:
            for network in self.vm["network"]:
                network_name = network[0]
                _ = network[1]  # tap_mac
                tap_id = network[2]

                delete_network_interface("tap{}".format(tap_id))

                owners_vms = shared.vm_pool.by_owner(self.vm["owner"])
                owners_running_vms = shared.vm_pool.by_status(
                    VMStatus.running, _vms=owners_vms
                )

                networks = map(
                    lambda n: n[0],
                    map(lambda vm: vm.network, owners_running_vms),
                )
                networks_in_use_by_user_vms = [vm[0] for vm in networks]
                if network_name not in networks_in_use_by_user_vms:
                    network_entry = resolve_network(
                        network[0], self.vm["owner"]
                    )
                    if network_entry:
                        network_type = network_entry.value["type"]
                        network_id = network_entry.value["id"]
                        if network_type == "vxlan":
                            delete_network_interface(
                                "br{}".format(network_id)
                            )
                            delete_network_interface(
                                "vxlan{}".format(network_id)
                            )
        except Exception:
            logger.exception("Exception in network interface deletion")

    def create(self):
        if shared.storage_handler.is_vm_image_exists(self.uuid):
            # File Already exists. No Problem Continue
            logger.debug("Image for vm %s exists", self.uuid)
        else:
            if shared.storage_handler.make_vm_image(
                src=self.vm["image_uuid"], dest=self.uuid
            ):
                if not shared.storage_handler.resize_vm_image(
                    path=self.uuid,
                    size=int(self.vm["specs"]["os-ssd"].to_MB()),
                ):
                    self.vm["status"] = VMStatus.error
                else:
                    logger.info("New VM Created")

    def sync(self):
        shared.etcd_client.put(
            self.key, self.schema.dump(self.vm), value_in_json=True
        )

    def delete(self):
        self.stop()

        if shared.storage_handler.is_vm_image_exists(self.uuid):
            r_status = shared.storage_handler.delete_vm_image(self.uuid)
            if r_status:
                shared.etcd_client.client.delete(self.key)
        else:
            shared.etcd_client.client.delete(self.key)


def resolve_network(network_name, network_owner):
    network = shared.etcd_client.get(
        join_path(
            settings["etcd"]["network_prefix"],
            network_owner,
            network_name,
        ),
        value_in_json=True,
    )
    return network


def create_vxlan_br_tap(_id, _dev, tap_id, 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(tap_id),
                dev=bridge,
            )
            if tap:
                return tap


def update_radvd_conf(all_networks):
    network_script_base = os.path.join(
        os.path.dirname(os.path.dirname(__file__)), "network"
    )

    networks = {
        net.value["ipv6"]: net.value["id"]
        for net in all_networks
        if net.value.get("ipv6")
        and ipaddress.ip_network(net.value.get("ipv6")).is_global
    }
    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)
    try:
        sp.check_output(["systemctl", "restart", "radvd"])
    except sp.CalledProcessError:
        try:
            sp.check_output(["service", "radvd", "restart"])
        except sp.CalledProcessError as err:
            raise err.__class__(
                "Cannot start/restart radvd service", err.cmd
            ) from err