mravi/uncloud
1
0
Fork 0
Code Issues Pull requests Projects Releases Wiki Activity

Remove ucloud_common and put its files under ucloud.common subpackage.

Browse source 
Remove individual config.py used by every component and put them into single config.py ucloud/config.py
Use /etc/ucloud/ucloud.conf for Environment Variables
Refactoring and a lot of it
Make ucloud repo a package and different components of ucloud a subpackage for avoiding code duplication.
Improved logging.
This commit is contained in:
ahmadbilalkhalid 2019-11-18 22:39:57 +05:00
commit 6fa77bce4d
51 changed files with 889 additions and 566 deletions
Download patch file Download diff file Expand all files Collapse all files

3
host/__init__.py Normal file
View file

@ -0,0 +1,3 @@
import logging
logger = logging.getLogger(__name__)

36
host/config.py
View file

@ -1,36 +0,0 @@
import logging
from etcd3_wrapper import Etcd3Wrapper
from ucloud_common.vm import VmPool
from ucloud_common.host import HostPool
from ucloud_common.request import RequestPool
from decouple import config
WITHOUT_CEPH = config("WITHOUT_CEPH", False, cast=bool)
logging.basicConfig(
level=logging.DEBUG,
filename="log.txt",
filemode="a",
format="%(asctime)s: %(levelname)s - %(message)s",
datefmt="%d-%b-%y %H:%M:%S",
)
etcd_wrapper_args = ()
etcd_wrapper_kwargs = {"host": config("ETCD_URL")}
etcd_client = Etcd3Wrapper(*etcd_wrapper_args, **etcd_wrapper_kwargs)
HOST_PREFIX = config("HOST_PREFIX")
NETWORK_PREFIX = config("NETWORK_PREFIX")
VM_PREFIX = config("VM_PREFIX")
REQUEST_PREFIX = config("REQUEST_PREFIX")
VM_DIR = config("VM_DIR")
IMAGE_DIR = config("IMAGE_DIR")
host_pool = HostPool(etcd_client, HOST_PREFIX)
vm_pool = VmPool(etcd_client, VM_PREFIX)
request_pool = RequestPool(etcd_client, REQUEST_PREFIX)
running_vms = []

78
host/main.py
View file

@ -1,31 +1,29 @@
import argparse
import time
import os
import sys
import virtualmachine
import multiprocessing as mp
import os
import time
from ucloud_common.host import HostEntry
from ucloud_common.request import RequestEntry, RequestType
from config import (vm_pool, host_pool, request_pool,
etcd_client, logging, running_vms,
etcd_wrapper_args, etcd_wrapper_kwargs,
REQUEST_PREFIX, HOST_PREFIX,
WITHOUT_CEPH, VM_DIR, HostPool)
from etcd3_wrapper import Etcd3Wrapper
import etcd3
from common.request import RequestEntry, RequestType
from config import (vm_pool, request_pool,
etcd_client, running_vms,
etcd_wrapper_args, etcd_wrapper_kwargs,
HostPool, env_vars)
from . import virtualmachine
from host import logger
def update_heartbeat(host):
client = Etcd3Wrapper(*etcd_wrapper_args, **etcd_wrapper_kwargs)
host_pool = HostPool(client, HOST_PREFIX)
host_pool = HostPool(client, env_vars.get('HOST_PREFIX'))
this_host = next(filter(lambda h: h.hostname == host, host_pool.hosts), None)
while True:
this_host.update_heartbeat()
host_pool.put(this_host)
time.sleep(10)
def maintenance(host):
# To capture vm running according to running_vms list
@ -65,31 +63,25 @@ def maintenance(host):
running_vms.remove(_vm)
def main():
argparser = argparse.ArgumentParser()
argparser.add_argument("hostname", help="Name of this host. e.g /v1/host/1")
args = argparser.parse_args()
def main(hostname):
assert env_vars.get('WITHOUT_CEPH') and os.path.isdir(env_vars.get('VM_DIR')), (
"You have set env_vars.get('WITHOUT_CEPH') to True. So, the vm directory mentioned"
" in .env file must exists. But, it don't.")
assert WITHOUT_CEPH and os.path.isdir(VM_DIR), (
"You have set WITHOUT_CEPH to True. So, the vm directory mentioned"
" in .env file must exists. But, it don't." )
mp.set_start_method('spawn')
heartbeat_updating_process = mp.Process(target=update_heartbeat, args=(args.hostname,))
heartbeat_updating_process = mp.Process(target=update_heartbeat, args=(hostname,))
host_pool = HostPool(etcd_client, HOST_PREFIX)
host = next(filter(lambda h: h.hostname == args.hostname, host_pool.hosts), None)
host_pool = HostPool(etcd_client, env_vars.get('HOST_PREFIX'))
host = next(filter(lambda h: h.hostname == hostname, host_pool.hosts), None)
assert host is not None, "No such host"
try:
heartbeat_updating_process.start()
except Exception as e:
logging.info("No Need To Go Further. Our heartbeat updating mechanism is not working")
logging.exception(e)
logger.info("No Need To Go Further. Our heartbeat updating mechanism is not working")
logger.exception(e)
exit(-1)
logging.info("%s Session Started %s", '*' * 5, '*' * 5)
logger.info("%s Session Started %s", '*' * 5, '*' * 5)
# It is seen that under heavy load, timeout event doesn't come
# in a predictive manner (which is intentional because we give
@ -99,22 +91,21 @@ def main():
# update the heart beat in a predictive manner we start Heart
# beat updating mechanism in separated thread
for events_iterator in [
etcd_client.get_prefix(REQUEST_PREFIX, value_in_json=True),
etcd_client.watch_prefix(REQUEST_PREFIX, timeout=10, value_in_json=True),
etcd_client.get_prefix(env_vars.get('REQUEST_PREFIX'), value_in_json=True),
etcd_client.watch_prefix(env_vars.get('REQUEST_PREFIX'), timeout=10, value_in_json=True),
]:
for request_event in events_iterator:
request_event = RequestEntry(request_event)
if request_event.type == "TIMEOUT":
logging.info("Timeout Event")
logger.info("Timeout Event")
maintenance(host)
continue
# If the event is directed toward me OR I am destination of a InitVMMigration
if (request_event.hostname == host.key or request_event.destination == host.key):
logging.debug("EVENT: %s", request_event)
if request_event.hostname == host.key or request_event.destination == host.key:
logger.debug("EVENT: %s", request_event)
request_pool.client.client.delete(request_event.key)
vm_entry = vm_pool.get(request_event.uuid)
@ -135,11 +126,14 @@ def main():
elif request_event.type == RequestType.TransferVM:
virtualmachine.transfer(request_event)
else:
logging.info("VM Entry missing")
logging.info("Running VMs %s", running_vms)
logger.info("VM Entry missing")
logger.info("Running VMs %s", running_vms)
if __name__ == "__main__":
main()
argparser = argparse.ArgumentParser()
argparser.add_argument("hostname", help="Name of this host. e.g /v1/host/1")
args = argparser.parse_args()
mp.set_start_method('spawn')
main(args.hostname)

16
host/qmp/__init__.py
View file

@ -15,24 +15,23 @@
import errno
import logging
import os
import subprocess
import re
import shutil
import socket
import subprocess
import tempfile
from . import qmp
LOG = logging.getLogger(__name__)
# Mapping host architecture to any additional architectures it can
# support which often includes its 32 bit cousin.
ADDITIONAL_ARCHES = {
"x86_64" : "i386",
"aarch64" : "armhf"
"x86_64": "i386",
"aarch64": "armhf"
}
def kvm_available(target_arch=None):
host_arch = os.uname()[4]
if target_arch and target_arch != host_arch:
@ -56,10 +55,12 @@ class QEMUMachineAddDeviceError(QEMUMachineError):
failures reported by the QEMU binary itself.
"""
class MonitorResponseError(qmp.QMPError):
"""
Represents erroneous QMP monitor reply
"""
def __init__(self, reply):
try:
desc = reply["error"]["desc"]
@ -108,7 +109,7 @@ class QEMUMachine(object):
self._qemu_log_file = None
self._popen = None
self._binary = binary
self._args = list(args) # Force copy args in case we modify them
self._args = list(args) # Force copy args in case we modify them
self._wrapper = wrapper
self._events = []
self._iolog = None
@ -453,10 +454,11 @@ class QEMUMachine(object):
See event_match for details.
timeout: QEMUMonitorProtocol.pull_event timeout parameter.
"""
def _match(event):
for name, match in events:
if (event['event'] == name and
self.event_match(event, match)):
self.event_match(event, match)):
return True
return False

BIN
host/qmp/__pycache__/__init__.cpython-37.pyc
View file

Binary file not shown.

BIN
host/qmp/__pycache__/qmp.cpython-37.pyc
View file

Binary file not shown.

5
host/qmp/qmp.py
View file

@ -8,10 +8,10 @@
# This work is licensed under the terms of the GNU GPL, version 2. See
# the COPYING file in the top-level directory.
import json
import errno
import socket
import json
import logging
import socket
class QMPError(Exception):
@ -31,7 +31,6 @@ class QMPTimeoutError(QMPError):
class QEMUMonitorProtocol(object):
#: Logger object for debugging messages
logger = logging.getLogger('QMP')
#: Socket's error class

126
host/virtualmachine.py
View file

@ -6,30 +6,24 @@
import errno
import os
import random
import subprocess as sp
import tempfile
import time
import random
import ipaddress
from functools import wraps
from os.path import join
from typing import Union
from string import Template
from typing import Union
import bitmath
import sshtunnel
import qmp
from decouple import config
from ucloud_common.helpers import get_ipv4_address
from ucloud_common.request import RequestEntry, RequestType
from ucloud_common.vm import VMEntry, VMStatus
from config import (WITHOUT_CEPH, VM_PREFIX, VM_DIR, IMAGE_DIR,
NETWORK_PREFIX, etcd_client, logging,
request_pool, running_vms, vm_pool)
from common.helpers import get_ipv4_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
from . import qmp
from host import logger
class VM:
def __init__(self, key, handle, vnc_socket_file):
@ -43,7 +37,7 @@ class VM:
def create_dev(script, _id, dev, ip=None):
command = [script, _id, dev]
if ip:
if ip:
command.append(ip)
try:
output = sp.check_output(command, stderr=sp.PIPE)
@ -57,13 +51,13 @@ def create_dev(script, _id, dev, ip=None):
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)
_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)
_id=str(random.randint(1, 100000)), dev=bridge)
if tap:
return tap
@ -77,35 +71,35 @@ def generate_mac(uaa=False, multicast=False, oui=None, separator=':', byte_fmt='
if oui:
if type(oui) == str:
oui = [int(chunk) for chunk in oui.split(separator)]
mac = oui + random_bytes(num=6-len(oui))
mac = oui + random_bytes(num=6 - len(oui))
else:
if multicast:
mac[0] |= 1 # set bit 0
mac[0] |= 1 # set bit 0
else:
mac[0] &= ~1 # clear bit 0
mac[0] &= ~1 # clear bit 0
if uaa:
mac[0] &= ~(1 << 1) # clear bit 1
mac[0] &= ~(1 << 1) # clear bit 1
else:
mac[0] |= 1 << 1 # set bit 1
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')
}
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)
prefix=net)
for net in networks if networks.get(net)]
with open('/etc/radvd.conf', 'w') as radvd_conf:
radvd_conf.writelines(content)
@ -113,7 +107,7 @@ def update_radvd_conf(etcd_client):
def get_start_command_args(
vm_entry, vnc_sock_filename: str, migration=False, migration_port=4444,
vm_entry, vnc_sock_filename: str, migration=False, migration_port=4444,
):
threads_per_core = 1
vm_memory = int(bitmath.parse_string(vm_entry.specs["ram"]).to_MB())
@ -121,9 +115,9 @@ def get_start_command_args(
vm_uuid = vm_entry.uuid
vm_networks = vm_entry.network
if WITHOUT_CEPH:
if env_vars.get('WITHOUT_CEPH'):
command = "-drive file={},format=raw,if=virtio,cache=none".format(
os.path.join(VM_DIR, vm_uuid)
os.path.join(env_vars.get('VM_DIR'), vm_uuid)
)
else:
command = "-drive file=rbd:uservms/{},format=raw,if=virtio,cache=none".format(
@ -138,24 +132,24 @@ def get_start_command_args(
if migration:
command += " -incoming tcp:0:{}".format(migration_port)
tap = None
for network_and_mac in vm_networks:
network_name, mac = network_and_mac
_key = os.path.join(NETWORK_PREFIX, vm_entry.owner, network_name)
_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, config("VXLAN_PHY_DEV"), network_ipv6)
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)
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(" ")
@ -189,15 +183,15 @@ def need_running_vm(func):
if vm:
try:
status = vm.handle.command("query-status")
logging.debug("VM Status Check - %s", status)
logger.debug("VM Status Check - %s", status)
except Exception as exception:
logging.info("%s failed - VM %s %s", func.__name__, e, exception)
logger.info("%s failed - VM %s %s", func.__name__, e, exception)
else:
return func(e)
return None
else:
logging.info("%s failed because VM %s is not running", func.__name__, e.key)
logger.info("%s failed because VM %s is not running", func.__name__, e.key)
return None
return wrapper
@ -206,18 +200,18 @@ def need_running_vm(func):
def create(vm_entry: VMEntry):
vm_hdd = int(bitmath.parse_string(vm_entry.specs["os-ssd"]).to_MB())
if WITHOUT_CEPH:
if env_vars.get('WITHOUT_CEPH'):
_command_to_create = [
"cp",
os.path.join(IMAGE_DIR, vm_entry.image_uuid),
os.path.join(VM_DIR, vm_entry.uuid),
os.path.join(env_vars.get('IMAGE_DIR'), vm_entry.image_uuid),
os.path.join(env_vars.get('VM_DIR'), vm_entry.uuid),
]
_command_to_extend = [
"qemu-img",
"resize",
"-f", "raw",
os.path.join(VM_DIR, vm_entry.uuid),
os.path.join(env_vars.get('VM_DIR'), vm_entry.uuid),
"{}M".format(vm_hdd),
]
else:
@ -240,22 +234,22 @@ def create(vm_entry: VMEntry):
sp.check_output(_command_to_create)
except sp.CalledProcessError as e:
if e.returncode == errno.EEXIST:
logging.debug("Image for vm %s exists", vm_entry.uuid)
logger.debug("Image for vm %s exists", vm_entry.uuid)
# File Already exists. No Problem Continue
return
# This exception catches all other exceptions
# i.e FileNotFound (BaseImage), pool Does Not Exists etc.
logging.exception(e)
logger.exception(e)
vm_entry.status = "ERROR"
else:
try:
sp.check_output(_command_to_extend)
except Exception as e:
logging.exception(e)
logger.exception(e)
else:
logging.info("New VM Created")
logger.info("New VM Created")
def start(vm_entry: VMEntry):
@ -263,7 +257,7 @@ def start(vm_entry: VMEntry):
# VM already running. No need to proceed further.
if _vm:
logging.info("VM %s already running", vm_entry.uuid)
logger.info("VM %s already running", vm_entry.uuid)
return
else:
create(vm_entry)
@ -282,19 +276,19 @@ def stop(vm_entry):
def delete(vm_entry):
logging.info("Deleting VM | %s", vm_entry)
logger.info("Deleting VM | %s", vm_entry)
stop(vm_entry)
path_without_protocol = vm_entry.path[vm_entry.path.find(":") + 1 :]
path_without_protocol = vm_entry.path[vm_entry.path.find(":") + 1:]
if WITHOUT_CEPH:
vm_deletion_command = ["rm", os.path.join(VM_DIR, vm_entry.uuid)]
if env_vars.get('WITHOUT_CEPH'):
vm_deletion_command = ["rm", os.path.join(env_vars.get('VM_DIR'), vm_entry.uuid)]
else:
vm_deletion_command = ["rbd", "rm", path_without_protocol]
try:
sp.check_output(vm_deletion_command)
except Exception as e:
logging.exception(e)
logger.exception(e)
else:
etcd_client.client.delete(vm_entry.key)
@ -307,20 +301,20 @@ def transfer(request_event):
_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(VM_PREFIX, _uuid))
vm = get_vm(running_vms, join(env_vars.get('VM_PREFIX'), _uuid))
if vm:
tunnel = sshtunnel.SSHTunnelForwarder(
(_host, 22),
ssh_username=config("ssh_username"),
ssh_pkey=config("ssh_pkey"),
ssh_private_key_password=config("ssh_private_key_password"),
ssh_username=env_vars.get("ssh_username"),
ssh_pkey=env_vars.get("ssh_pkey"),
ssh_private_key_password=env_vars.get("ssh_private_key_password"),
remote_bind_address=("127.0.0.1", _port),
)
try:
tunnel.start()
except sshtunnel.BaseSSHTunnelForwarderError:
logging.exception("Couldn't establish connection to (%s, 22)", _host)
logger.exception("Couldn't establish connection to (%s, 22)", _host)
else:
vm.handle.command(
"migrate", uri="tcp:{}:{}".format(_host, tunnel.local_bind_port)
@ -356,7 +350,7 @@ def init_migration(vm_entry, destination_host_key):
if _vm:
# VM already running. No need to proceed further.
logging.info("%s Already running", _vm.key)
logger.info("%s Already running", _vm.key)
return
launch_vm(vm_entry, migration=True, migration_port=4444,
@ -364,13 +358,13 @@ def init_migration(vm_entry, destination_host_key):
def launch_vm(vm_entry, migration=False, migration_port=None, destination_host_key=None):
logging.info("Starting %s", vm_entry.key)
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 as e:
logging.exception(e)
logger.exception(e)
if migration:
# We don't care whether MachineError or any other error occurred
@ -392,6 +386,7 @@ def launch_vm(vm_entry, migration=False, migration_port=None, destination_host_k
parameters={"host": get_ipv4_address(), "port": 4444},
uuid=vm_entry.uuid,
destination_host_key=destination_host_key,
request_prefix=env_vars.get("REQUEST_PREFIX")
)
request_pool.put(r)
else:
@ -400,4 +395,3 @@ def launch_vm(vm_entry, migration=False, migration_port=None, destination_host_k
vm_entry.add_log("Started successfully")
vm_pool.put(vm_entry)