X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_openvim%2Fhost_thread.py;h=ed6551fb1efc4a8f189ba149598148ed53af278b;hb=9cead2a0262deabadc900d580ef312fbc468efc0;hp=bf7a6da09c2b3816e59f4b07ac14a6951990f914;hpb=e3cf1f716be59fe0f3271a44589c1f31b10567ec;p=osm%2Fopenvim.git diff --git a/osm_openvim/host_thread.py b/osm_openvim/host_thread.py index bf7a6da..ed6551f 100644 --- a/osm_openvim/host_thread.py +++ b/osm_openvim/host_thread.py @@ -1,7 +1,7 @@ # -*- coding: utf-8 -*- ## -# Copyright 2015 Telefónica Investigación y Desarrollo, S.A.U. +# Copyright 2015 Telefonica Investigacion y Desarrollo, S.A.U. # This file is part of openvim # All Rights Reserved. # @@ -34,35 +34,40 @@ import threading import time import Queue import paramiko -from jsonschema import validate as js_v, exceptions as js_e -#import libvirt +import subprocess +# import libvirt import imp -from vim_schema import localinfo_schema, hostinfo_schema import random import os import logging +from jsonschema import validate as js_v, exceptions as js_e +from vim_schema import localinfo_schema, hostinfo_schema +class RunCommandException(Exception): + pass class host_thread(threading.Thread): lvirt_module = None - def __init__(self, name, host, user, db, db_lock, test, image_path, host_id, version, develop_mode, - develop_bridge_iface, logger_name=None, debug=None): - '''Init a thread. - Arguments: - 'id' number of thead - 'name' name of thread - 'host','user': host ip or name to manage and user - 'db', 'db_lock': database class and lock to use it in exclusion - ''' + def __init__(self, name, host, user, db, test, image_path, host_id, version, develop_mode, + develop_bridge_iface, password=None, keyfile = None, logger_name=None, debug=None, hypervisors=None): + """Init a thread to communicate with compute node or ovs_controller. + :param host_id: host identity + :param name: name of the thread + :param host: host ip or name to manage and user + :param user, password, keyfile: user and credentials to connect to host + :param db: database class, threading safe + """ threading.Thread.__init__(self) self.name = name self.host = host self.user = user self.db = db - self.db_lock = db_lock self.test = test + self.password = password + self.keyfile = keyfile self.localinfo_dirty = False + self.connectivity = True if not test and not host_thread.lvirt_module: try: @@ -82,64 +87,153 @@ class host_thread(threading.Thread): self.develop_mode = develop_mode self.develop_bridge_iface = develop_bridge_iface self.image_path = image_path + self.empty_image_path = image_path self.host_id = host_id self.version = version self.xml_level = 0 - #self.pending ={} + # self.pending ={} self.server_status = {} #dictionary with pairs server_uuid:server_status self.pending_terminate_server =[] #list with pairs (time,server_uuid) time to send a terminate for a server being destroyed self.next_update_server_status = 0 #time when must be check servers status +####### self.hypervisor = "kvm" #hypervisor flag (default: kvm) + if hypervisors: + self.hypervisors = hypervisors + else: + self.hypervisors = "kvm" + + self.xen_hyp = True if "xen" in self.hypervisors else False + self.hostinfo = None self.queueLock = threading.Lock() self.taskQueue = Queue.Queue(2000) self.ssh_conn = None + self.run_command_session = None + self.error = None + self.localhost = True if host == 'localhost' else False + + if self.xen_hyp: + self.lvirt_conn_uri = "xen+ssh://{user}@{host}/?no_tty=1&no_verify=1".format( + user=self.user, host=self.host) + else: + self.lvirt_conn_uri = "qemu+ssh://{user}@{host}/system?no_tty=1&no_verify=1".format( + user=self.user, host=self.host) + if keyfile: + self.lvirt_conn_uri += "&keyfile=" + keyfile + + self.remote_ip = None + self.local_ip = None + + def run_command(self, command, keep_session=False, ignore_exit_status=False): + """Run a command passed as a str on a localhost or at remote machine. + :param command: text with the command to execute. + :param keep_session: if True it returns a for sending input with '.write("text\n")'. + A command with keep_session=True MUST be followed by a command with keep_session=False in order to + close the session and get the output + :param ignore_exit_status: Return stdout and not raise an exepction in case of error. + :return: the output of the command if 'keep_session=False' or the object if 'keep_session=True' + :raises: RunCommandException if command fails + """ + if self.run_command_session and keep_session: + raise RunCommandException("Internal error. A command with keep_session=True must be followed by another " + "command with keep_session=False to close session") + try: + if self.localhost: + if self.run_command_session: + p = self.run_command_session + self.run_command_session = None + (output, outerror) = p.communicate() + returncode = p.returncode + p.stdin.close() + elif keep_session: + p = subprocess.Popen(('bash', "-c", command), stdin=subprocess.PIPE, stdout=subprocess.PIPE, + stderr=subprocess.PIPE) + self.run_command_session = p + return p.stdin + else: + if not ignore_exit_status: + output = subprocess.check_output(('bash', "-c", command)) + returncode = 0 + else: + out = None + p = subprocess.Popen(('bash', "-c", command), stdout=subprocess.PIPE) + out, err = p.communicate() + return out + else: + if self.run_command_session: + (i, o, e) = self.run_command_session + self.run_command_session = None + i.channel.shutdown_write() + else: + if not self.ssh_conn: + self.ssh_connect() + (i, o, e) = self.ssh_conn.exec_command(command, timeout=10) + if keep_session: + self.run_command_session = (i, o, e) + return i + returncode = o.channel.recv_exit_status() + output = o.read() + outerror = e.read() + if returncode != 0 and not ignore_exit_status: + text = "run_command='{}' Error='{}'".format(command, outerror) + self.logger.error(text) + raise RunCommandException(text) + + self.logger.debug("run_command='{}' result='{}'".format(command, output)) + return output + + except RunCommandException: + raise + except subprocess.CalledProcessError as e: + text = "run_command Exception '{}' '{}'".format(str(e), e.output) + except (paramiko.ssh_exception.SSHException, Exception) as e: + text = "run_command='{}' Exception='{}'".format(command, str(e)) + self.ssh_conn = None + self.run_command_session = None + raise RunCommandException(text) def ssh_connect(self): try: - #Connect SSH + # Connect SSH self.ssh_conn = paramiko.SSHClient() self.ssh_conn.set_missing_host_key_policy(paramiko.AutoAddPolicy()) self.ssh_conn.load_system_host_keys() - self.ssh_conn.connect(self.host, username=self.user, timeout=10) #, None) - except paramiko.ssh_exception.SSHException as e: - text = e.args[0] - self.logger.error("ssh_connect ssh Exception: " + text) - + self.ssh_conn.connect(self.host, username=self.user, password=self.password, key_filename=self.keyfile, + timeout=10) # auth_timeout=10) + self.remote_ip = self.ssh_conn.get_transport().sock.getpeername()[0] + self.local_ip = self.ssh_conn.get_transport().sock.getsockname()[0] + except (paramiko.ssh_exception.SSHException, Exception) as e: + text = 'ssh connect Exception: {}'.format(e) + self.ssh_conn = None + self.error = text + raise + + def check_connectivity(self): + if not self.test: + try: + command = 'sudo brctl show' + self.run_command(command) + except RunCommandException as e: + self.connectivity = False + self.logger.error("check_connectivity Exception: " + str(e)) + def load_localinfo(self): if not self.test: try: - #Connect SSH - self.ssh_connect() - - command = 'mkdir -p ' + self.image_path - #print self.name, ': command:', command - (_, stdout, stderr) = self.ssh_conn.exec_command(command) - content = stderr.read() - if len(content) > 0: - self.logger.error("command: '%s' stderr: '%s'", command, content) - - command = 'cat ' + self.image_path + '/.openvim.yaml' - #print self.name, ': command:', command - (_, stdout, stderr) = self.ssh_conn.exec_command(command) - content = stdout.read() - if len(content) == 0: - self.logger.error("command: '%s' stderr='%s'", command, stderr.read()) - raise paramiko.ssh_exception.SSHException("Error empty file, command: '{}'".format(command)) - self.localinfo = yaml.load(content) + self.run_command('sudo mkdir -p ' + self.image_path) + result = self.run_command('cat {}/.openvim.yaml'.format(self.image_path)) + self.localinfo = yaml.load(result) js_v(self.localinfo, localinfo_schema) self.localinfo_dirty = False if 'server_files' not in self.localinfo: self.localinfo['server_files'] = {} - self.logger.debug("localinfo load from host") + self.logger.debug("localinfo loaded from host") return - - except paramiko.ssh_exception.SSHException as e: - text = e.args[0] - self.logger.error("load_localinfo ssh Exception: " + text) + except RunCommandException as e: + self.logger.error("load_localinfo Exception: " + str(e)) except host_thread.lvirt_module.libvirtError as e: text = e.get_error_message() self.logger.error("load_localinfo libvirt Exception: " + text) @@ -157,34 +251,22 @@ class host_thread(threading.Thread): text = str(e) self.logger.error("load_localinfo Exception: " + text) - #not loaded, insert a default data and force saving by activating dirty flag + # not loaded, insert a default data and force saving by activating dirty flag self.localinfo = {'files':{}, 'server_files':{} } - #self.localinfo_dirty=True + # self.localinfo_dirty=True self.localinfo_dirty=False def load_hostinfo(self): if self.test: - return; + return try: - #Connect SSH - self.ssh_connect() - - - command = 'cat ' + self.image_path + '/hostinfo.yaml' - #print self.name, ': command:', command - (_, stdout, stderr) = self.ssh_conn.exec_command(command) - content = stdout.read() - if len(content) == 0: - self.logger.error("command: '%s' stderr: '%s'", command, stderr.read()) - raise paramiko.ssh_exception.SSHException("Error empty file ") - self.hostinfo = yaml.load(content) + result = self.run_command('cat {}/hostinfo.yaml'.format(self.image_path)) + self.hostinfo = yaml.load(result) js_v(self.hostinfo, hostinfo_schema) - self.logger.debug("hostlinfo load from host " + str(self.hostinfo)) + self.logger.debug("hostinfo load from host " + str(self.hostinfo)) return - - except paramiko.ssh_exception.SSHException as e: - text = e.args[0] - self.logger.error("load_hostinfo ssh Exception: " + text) + except RunCommandException as e: + self.logger.error("load_hostinfo ssh Exception: " + str(e)) except host_thread.lvirt_module.libvirtError as e: text = e.get_error_message() self.logger.error("load_hostinfo libvirt Exception: " + text) @@ -197,7 +279,7 @@ class host_thread(threading.Thread): except js_e.ValidationError as e: text = "" if len(e.path)>0: text=" at '" + ":".join(map(str, e.path))+"'" - self.logger.error("load_hostinfo format Exception: %s %s", text, e.message) + self.logger.error("load_hostinfo format Exception: %s %s", text, str(e)) except Exception as e: text = str(e) self.logger.error("load_hostinfo Exception: " + text) @@ -214,18 +296,17 @@ class host_thread(threading.Thread): tries-=1 try: - command = 'cat > ' + self.image_path + '/.openvim.yaml' - self.logger.debug("command:" + command) - (stdin, _, _) = self.ssh_conn.exec_command(command) - yaml.safe_dump(self.localinfo, stdin, explicit_start=True, indent=4, default_flow_style=False, tags=False, encoding='utf-8', allow_unicode=True) + command = 'cat > {}/.openvim.yaml'.format(self.image_path) + in_stream = self.run_command(command, keep_session=True) + yaml.safe_dump(self.localinfo, in_stream, explicit_start=True, indent=4, default_flow_style=False, + tags=False, encoding='utf-8', allow_unicode=True) + result = self.run_command(command, keep_session=False) # to end session + self.localinfo_dirty = False break #while tries - - except paramiko.ssh_exception.SSHException as e: - text = e.args[0] - self.logger.error("save_localinfo ssh Exception: " + text) - if "SSH session not active" in text: - self.ssh_connect() + + except RunCommandException as e: + self.logger.error("save_localinfo ssh Exception: " + str(e)) except host_thread.lvirt_module.libvirtError as e: text = e.get_error_message() self.logger.error("save_localinfo libvirt Exception: " + text) @@ -240,9 +321,7 @@ class host_thread(threading.Thread): self.logger.error("save_localinfo Exception: " + text) def load_servers_from_db(self): - self.db_lock.acquire() r,c = self.db.get_table(SELECT=('uuid','status', 'image_id'), FROM='instances', WHERE={'host_id': self.host_id}) - self.db_lock.release() self.server_status = {} if r<0: @@ -275,7 +354,7 @@ class host_thread(threading.Thread): try: self.logger.debug("deleting file '%s' of unused server '%s'", localfile['source file'], uuid) self.delete_file(localfile['source file']) - except paramiko.ssh_exception.SSHException as e: + except RunCommandException as e: self.logger.error("Exception deleting file '%s': %s", localfile['source file'], str(e)) del self.localinfo['server_files'][uuid] self.localinfo_dirty = True @@ -420,15 +499,19 @@ class host_thread(threading.Thread): bus_ide = True if bus=='ide' else False self.xml_level = 0 + hypervisor = server.get('hypervisor', 'kvm') + os_type_img = server.get('os_image_type', 'other') - text = "" + if hypervisor[:3] == 'xen': + text = "" + else: + text = "" #get topology topo = server_metadata.get('topology', None) if topo == None and 'metadata' in dev_list[0]: topo = dev_list[0]['metadata'].get('topology', None) #name - name = server.get('name','') + "_" + server['uuid'] - name = name[:58] #qemu impose a length limit of 59 chars or not start. Using 58 + name = server.get('name', '')[:28] + "_" + server['uuid'][:28] #qemu impose a length limit of 59 chars or not start. Using 58 text += self.inc_tab() + "" + name+ "" #uuid text += self.tab() + "" + server['uuid'] + "" @@ -496,12 +579,26 @@ class host_thread(threading.Thread): if dev['type']=='cdrom' : boot_cdrom = True break - text += self.tab()+ '' + \ - self.inc_tab() + "hvm" - if boot_cdrom: - text += self.tab() + "" - text += self.tab() + "" + \ - self.dec_tab()+'' + if hypervisor == 'xenhvm': + text += self.tab()+ '' + \ + self.inc_tab() + "hvm" + text += self.tab() + "/usr/lib/xen/boot/hvmloader" + if boot_cdrom: + text += self.tab() + "" + text += self.tab() + "" + \ + self.dec_tab()+'' + elif hypervisor == 'xen-unik': + text += self.tab()+ '' + \ + self.inc_tab() + "xen" + text += self.tab() + "" + str(dev_list[0]['source file']) + "" + \ + self.dec_tab()+'' + else: + text += self.tab()+ '' + \ + self.inc_tab() + "hvm" + if boot_cdrom: + text += self.tab() + "" + text += self.tab() + "" + \ + self.dec_tab()+'' #features text += self.tab()+''+\ self.inc_tab()+'' +\ @@ -511,7 +608,7 @@ class host_thread(threading.Thread): if topo == "oneSocket:hyperthreading": if vcpus % 2 != 0: return -1, 'Cannot expose hyperthreading with an odd number of vcpus' - text += self.tab() + " " % vcpus/2 + text += self.tab() + " " % (vcpus/2) elif windows_os or topo == "oneSocket": text += self.tab() + " " % vcpus else: @@ -520,14 +617,29 @@ class host_thread(threading.Thread): self.tab() + "preserve" + \ self.tab() + "restart" + \ self.tab() + "restart" - text += self.tab() + "" + \ - self.inc_tab() + "/usr/libexec/qemu-kvm" + \ - self.tab() + "" +\ - self.inc_tab() + "" + \ - self.dec_tab() + "" +\ - self.tab() + "" + \ - self.inc_tab()+ "" + \ - self.dec_tab()+'' + if hypervisor == 'xenhvm': + text += self.tab() + "" + \ + self.inc_tab() + "/usr/bin/qemu-system-i386" + \ + self.tab() + "" +\ + self.inc_tab() + "" + \ + self.dec_tab() + "" +\ + self.tab() + "" + \ + self.inc_tab()+ "" + \ + self.dec_tab()+'' #In some libvirt version may be: /usr/lib64/xen/bin/qemu-dm (depends on distro) + elif hypervisor == 'xen-unik': + text += self.tab() + "" + \ + self.tab() + "" + \ + self.inc_tab()+ "" + \ + self.dec_tab()+'' + else: + text += self.tab() + "" + \ + self.inc_tab() + "/usr/libexec/qemu-kvm" + \ + self.tab() + "" +\ + self.inc_tab() + "" + \ + self.dec_tab() + "" +\ + self.tab() + "" + \ + self.inc_tab()+ "" + \ + self.dec_tab()+'' if windows_os: text += self.tab() + "" + \ self.tab() + "" + \ @@ -538,6 +650,15 @@ class host_thread(threading.Thread): self.dec_tab() + "" + \ self.tab() + "" + \ self.tab() + "" #TODO revisar + elif hypervisor == 'xen-unik': + pass + else: + text += self.tab() + "" + \ + self.tab() + "" + \ + self.tab() + "" + \ + self.tab() + "" #> self.tab()+'\n' +\ #> self.dec_tab()+'\n' +\ @@ -554,7 +675,7 @@ class host_thread(threading.Thread): vd_index = 'a' for dev in dev_list: bus_ide_dev = bus_ide - if dev['type']=='cdrom' or dev['type']=='disk': + if (dev['type']=='cdrom' or dev['type']=='disk') and hypervisor != 'xen-unik': if dev['type']=='cdrom': bus_ide_dev = True text += self.tab() + "" @@ -567,7 +688,7 @@ class host_thread(threading.Thread): #else: # return -1, 'Unknown disk type ' + v['type'] vpci = dev.get('vpci',None) - if vpci == None: + if vpci == None and 'metadata' in dev: vpci = dev['metadata'].get('vpci',None) text += self.pci2xml(vpci) @@ -589,6 +710,8 @@ class host_thread(threading.Thread): dev_text = dev_text.replace('__dev__', vd_index) vd_index = chr(ord(vd_index)+1) text += dev_text + elif hypervisor == 'xen-unik': + pass else: return -1, 'Unknown device type ' + dev['type'] @@ -596,9 +719,7 @@ class host_thread(threading.Thread): bridge_interfaces = server.get('networks', []) for v in bridge_interfaces: #Get the brifge name - self.db_lock.acquire() result, content = self.db.get_table(FROM='nets', SELECT=('provider',),WHERE={'uuid':v['net_id']} ) - self.db_lock.release() if result <= 0: self.logger.error("create_xml_server ERROR %d getting nets %s", result, content) return -1, content @@ -629,6 +750,8 @@ class host_thread(threading.Thread): vlan = content[0]['provider'].replace('OVS:', '') text += self.tab() + "" + \ self.inc_tab() + "" + if hypervisor == 'xenhvm' or hypervisor == 'xen-unik': + text += self.tab() + "