X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=osm_ro%2Fvim_thread.py;h=3c35d6c905bfb9de98f7d3a932c17e1c5dec6f3c;hb=refs%2Fchanges%2F07%2F1607%2F3;hp=42279a23628f4e69cf12bc69f887f6efd8de30fc;hpb=4b6216b9d00195bf2e3772bb9278faf18ee9fd46;p=osm%2FRO.git diff --git a/osm_ro/vim_thread.py b/osm_ro/vim_thread.py index 42279a23..3c35d6c9 100644 --- a/osm_ro/vim_thread.py +++ b/osm_ro/vim_thread.py @@ -25,7 +25,7 @@ This is thread that interact with the host and the libvirt to manage VM One thread will be launched per host ''' -__author__ = "Alfonso Tierno" +__author__ = "Alfonso Tierno, Pablo Montes" __date__ = "$10-feb-2017 12:07:15$" import threading @@ -34,6 +34,7 @@ import Queue import logging import vimconn from db_base import db_base_Exception +from ovim import ovimException # from logging import Logger @@ -46,7 +47,7 @@ def is_task_id(id): class vim_thread(threading.Thread): - def __init__(self, vimconn, task_lock, name=None, datacenter_name=None, datacenter_tenant_id=None, db=None, db_lock=None): + def __init__(self, vimconn, task_lock, name=None, datacenter_name=None, datacenter_tenant_id=None, db=None, db_lock=None, ovim=None): """Init a thread. Arguments: 'id' number of thead @@ -64,6 +65,7 @@ class vim_thread(threading.Thread): self.vim = vimconn self.datacenter_name = datacenter_name self.datacenter_tenant_id = datacenter_tenant_id + self.ovim = ovim if not name: self.name = vimconn["id"] + "." + vimconn["config"]["datacenter_tenant_id"] else: @@ -75,6 +77,239 @@ class vim_thread(threading.Thread): self.task_lock = task_lock self.task_queue = Queue.Queue(2000) + self.refresh_list = [] + """Contains time ordered task list for refreshing the status of VIM VMs and nets""" + + def _refres_elements(self): + """Call VIM to get VMs and networks status until 10 elements""" + now = time.time() + vm_to_refresh_list = [] + net_to_refresh_list = [] + vm_to_refresh_dict = {} + net_to_refresh_dict = {} + items_to_refresh = 0 + while self.refresh_list: + task = self.refresh_list[0] + with self.task_lock: + if task['status'] == 'deleted': + self.refresh_list.pop(0) + continue + if task['time'] > now: + break + task["status"] = "processing" + self.refresh_list.pop(0) + if task["name"] == 'get-vm': + vm_to_refresh_list.append(task["vim_id"]) + vm_to_refresh_dict[task["vim_id"]] = task + elif task["name"] == 'get-net': + net_to_refresh_list.append(task["vim_id"]) + net_to_refresh_dict[task["vim_id"]] = task + else: + error_text = "unknown task {}".format(task["name"]) + self.logger.error(error_text) + items_to_refresh += 1 + if items_to_refresh == 10: + break + + if vm_to_refresh_list: + try: + vim_dict = self.vim.refresh_vms_status(vm_to_refresh_list) + for vim_id, vim_info in vim_dict.items(): + #look for task + task = vm_to_refresh_dict[vim_id] + self.logger.debug("get-vm vm_id=%s result=%s", task["vim_id"], str(vim_info)) + + # update database + if vim_info.get("error_msg"): + vim_info["error_msg"] = self._format_vim_error_msg(vim_info["error_msg"]) + if task["vim_info"].get("status") != vim_info["status"] or \ + task["vim_info"].get("error_msg") != vim_info.get("error_msg") or \ + task["vim_info"].get("vim_info") != vim_info["vim_info"]: + with self.db_lock: + temp_dict = {"status": vim_info["status"], + "error_msg": vim_info.get("error_msg"), + "vim_info": vim_info["vim_info"]} + self.db.update_rows('instance_vms', UPDATE=temp_dict, WHERE={"vim_vm_id": vim_id}) + for interface in vim_info["interfaces"]: + for task_interface in task["vim_info"]["interfaces"]: + if task_interface["vim_net_id"] == interface["vim_net_id"]: + break + else: + task_interface = {"vim_net_id": interface["vim_net_id"]} + task["vim_info"]["interfaces"].append(task_interface) + if task_interface != interface: + #delete old port + if task_interface.get("sdn_port_id"): + try: + self.ovim.delete_port(task_interface["sdn_port_id"]) + task_interface["sdn_port_id"] = None + except ovimException as e: + self.logger.error("ovimException deleting external_port={} ".format( + task_interface["sdn_port_id"]) + str(e), exc_info=True) + # TODO Set error_msg at instance_nets + vim_net_id = interface.pop("vim_net_id") + sdn_net_id = None + sdn_port_name = None + with self.db_lock: + where_= {'iv.vim_vm_id': vim_id, "ine.vim_net_id": vim_net_id, + 'ine.datacenter_tenant_id': self.datacenter_tenant_id} + # TODO check why vim_interface_id is not present at database + # if interface.get("vim_interface_id"): + # where_["vim_interface_id"] = interface["vim_interface_id"] + db_ifaces = self.db.get_rows( + FROM="instance_interfaces as ii left join instance_nets as ine on " + "ii.instance_net_id=ine.uuid left join instance_vms as iv on " + "ii.instance_vm_id=iv.uuid", + SELECT=("ii.uuid as iface_id", "ine.uuid as net_id", "iv.uuid as vm_id", "sdn_net_id"), + WHERE=where_) + if len(db_ifaces)>1: + self.logger.critical("Refresing interfaces. " + "Found more than one interface at database for '{}'".format(where_)) + elif len(db_ifaces)==0: + self.logger.critical("Refresing interfaces. " + "Not found any interface at database for '{}'".format(where_)) + continue + else: + db_iface = db_ifaces[0] + if db_iface.get("sdn_net_id") and interface.get("compute_node") and interface.get("pci"): + sdn_net_id = db_iface["sdn_net_id"] + sdn_port_name = sdn_net_id + "." + db_iface["vm_id"] + sdn_port_name = sdn_port_name[:63] + try: + sdn_port_id = self.ovim.new_external_port( + {"compute_node": interface["compute_node"], + "pci": interface["pci"], + "vlan": interface.get("vlan"), + "net_id": sdn_net_id, + "region": self.vim["config"]["datacenter_id"], + "name": sdn_port_name, + "mac": interface.get("mac_address")}) + interface["sdn_port_id"] = sdn_port_id + except (ovimException, Exception) as e: + self.logger.error( + "ovimException creating new_external_port compute_node={} " \ + "pci={} vlan={} ".format( + interface["compute_node"], + interface["pci"], + interface.get("vlan")) + str(e), + exc_info=True) + # TODO Set error_msg at instance_nets + with self.db_lock: + self.db.update_rows('instance_interfaces', UPDATE=interface, + WHERE={'uuid': db_iface["iface_id"]}) + # TODO insert instance_id + interface["vim_net_id"] = vim_net_id + + task["vim_info"] = vim_info + if task["vim_info"]["status"] == "BUILD": + self._insert_refresh(task, now+5) # 5seconds + else: + self._insert_refresh(task, now+300) # 5minutes + except vimconn.vimconnException as e: + self.logger.error("vimconnException Exception when trying to refresh vms " + str(e)) + self._insert_refresh(task, now + 300) # 5minutes + + if net_to_refresh_list: + try: + vim_dict = self.vim.refresh_nets_status(net_to_refresh_list) + for vim_id, vim_info in vim_dict.items(): + #look for task + task = net_to_refresh_dict[vim_id] + self.logger.debug("get-net net_id=%s result=%s", task["vim_id"], str(vim_info)) + + #get database info + where_ = {"vim_net_id": vim_id, 'datacenter_tenant_id': self.datacenter_tenant_id} + with self.db_lock: + db_nets = self.db.get_rows( + FROM="instance_nets", + SELECT=("uuid as net_id", "sdn_net_id"), + WHERE=where_) + if len(db_nets) > 1: + self.logger.critical("Refresing networks. " + "Found more than one instance-networks at database for '{}'".format(where_)) + elif len(db_nets) == 0: + self.logger.critical("Refresing networks. " + "Not found any instance-network at database for '{}'".format(where_)) + continue + else: + db_net = db_nets[0] + if db_net.get("sdn_net_id"): + # get ovim status + try: + sdn_net = self.ovim.show_network(db_net["sdn_net_id"]) + if sdn_net["status"] == "ERROR": + if not vim_info.get("error_msg"): + vim_info["error_msg"] = sdn_net["error_msg"] + else: + vim_info["error_msg"] = "VIM_ERROR: {} && SDN_ERROR: {}".format( + self._format_vim_error_msg(vim_info["error_msg"], 1024//2-14), + self._format_vim_error_msg(sdn_net["error_msg"], 1024//2-14)) + if vim_info["status"] == "VIM_ERROR": + vim_info["status"] = "VIM_SDN_ERROR" + else: + vim_info["status"] = "SDN_ERROR" + + except (ovimException, Exception) as e: + self.logger.error( + "ovimException getting network infor snd_net_id={}".format(db_net["sdn_net_id"]), + exc_info=True) + # TODO Set error_msg at instance_nets + + # update database + if vim_info.get("error_msg"): + vim_info["error_msg"] = self._format_vim_error_msg(vim_info["error_msg"]) + if task["vim_info"].get("status") != vim_info["status"] or \ + task["vim_info"].get("error_msg") != vim_info.get("error_msg") or \ + task["vim_info"].get("vim_info") != vim_info["vim_info"]: + with self.db_lock: + temp_dict = {"status": vim_info["status"], + "error_msg": vim_info.get("error_msg"), + "vim_info": vim_info["vim_info"]} + self.db.update_rows('instance_nets', UPDATE=temp_dict, WHERE={"vim_net_id": vim_id}) + + task["vim_info"] = vim_info + if task["vim_info"]["status"] == "BUILD": + self._insert_refresh(task, now+5) # 5seconds + else: + self._insert_refresh(task, now+300) # 5minutes + except vimconn.vimconnException as e: + self.logger.error("vimconnException Exception when trying to refresh nets " + str(e)) + self._insert_refresh(task, now + 300) # 5minutes + + if not items_to_refresh: + time.sleep(1) + + def _insert_refresh(self, task, threshold_time): + """Insert a task at list of refreshing elements. The refreshing list is ordered by threshold_time (task['time'] + It is assumed that this is called inside this thread + """ + task["time"] = threshold_time + for index in range(0, len(self.refresh_list)): + if self.refresh_list[index]["time"] > threshold_time: + self.refresh_list.insert(index, task) + break + else: + index = len(self.refresh_list) + self.refresh_list.append(task) + self.logger.debug("new refresh task={} name={}, time={} index={}".format( + task["id"], task["name"], task["time"], index)) + + def _remove_refresh(self, task_name, vim_id): + """Remove a task with this name and vim_id from the list of refreshing elements. + It is assumed that this is called inside this thread outside _refres_elements method + Return True if self.refresh_list is modified, task is found + Return False if not found + """ + index_to_delete = None + for index in range(0, len(self.refresh_list)): + if self.refresh_list[index]["name"] == task_name and self.refresh_list[index]["vim_id"] == vim_id: + index_to_delete = index + break + else: + return False + if index_to_delete != None: + del self.refresh_list[index_to_delete] + return True def insert_task(self, task): try: @@ -97,64 +332,113 @@ class vim_thread(threading.Thread): while True: #TODO reload service while True: - if not self.task_queue.empty(): - task = self.task_queue.get() - self.task_lock.acquire() - if task["status"] == "deleted": + try: + if not self.task_queue.empty(): + task = self.task_queue.get() + self.task_lock.acquire() + if task["status"] == "deleted": + self.task_lock.release() + continue + task["status"] = "processing" self.task_lock.release() + else: + self._refres_elements() continue - task["status"] == "processing" - self.task_lock.release() - else: - now=time.time() - time.sleep(1) - continue - self.logger.debug("processing task id={} name={} params={}".format(task["id"], task["name"], - str(task["params"]))) - if task["name"] == 'exit' or task["name"] == 'reload': - result, content = self.terminate(task) - elif task["name"] == 'new-vm': - result, content = self.new_vm(task) - elif task["name"] == 'del-vm': - result, content = self.del_vm(task) - elif task["name"] == 'new-net': - result, content = self.new_net(task) - elif task["name"] == 'del-net': - result, content = self.del_net(task) - else: - error_text = "unknown task {}".format(task["name"]) - self.logger.error(error_text) - result = False - content = error_text + self.logger.debug("processing task id={} name={} params={}".format(task["id"], task["name"], + str(task["params"]))) + if task["name"] == 'exit' or task["name"] == 'reload': + result, content = self.terminate(task) + elif task["name"] == 'new-vm': + result, content = self.new_vm(task) + elif task["name"] == 'del-vm': + result, content = self.del_vm(task) + elif task["name"] == 'new-net': + result, content = self.new_net(task) + elif task["name"] == 'del-net': + result, content = self.del_net(task) + else: + error_text = "unknown task {}".format(task["name"]) + self.logger.error(error_text) + result = False + content = error_text + self.logger.debug("task id={} name={} result={}:{} params={}".format(task["id"], task["name"], + result, content, + str(task["params"]))) - with self.task_lock: - task["status"] = "done" if result else "error" - task["result"] = content - self.task_queue.task_done() + with self.task_lock: + task["status"] = "done" if result else "error" + task["result"] = content + self.task_queue.task_done() - if task["name"] == 'exit': - return 0 - elif task["name"] == 'reload': - break + if task["name"] == 'exit': + return 0 + elif task["name"] == 'reload': + break + except Exception as e: + self.logger.critical("Unexpected exception at run: " + str(e), exc_info=True) self.logger.debug("Finishing") def terminate(self, task): return True, None + def _format_vim_error_msg(self, error_text, max_length=1024): + if error_text and len(error_text) >= max_length: + return error_text[:max_length//2-3] + " ... " + error_text[-max_length//2+3:] + return error_text + def new_net(self, task): try: task_id = task["id"] params = task["params"] net_id = self.vim.new_network(*params) + + net_name = params[0] + net_type = params[1] + + network = None + sdn_net_id = None + sdn_controller = self.vim.config.get('sdn-controller') + if sdn_controller and (net_type == "data" or net_type == "ptp"): + network = {"name": net_name, "type": net_type} + + vim_net = self.vim.get_network(net_id) + if vim_net.get('encapsulation') != 'vlan': + raise vimconn.vimconnException( + "net '{}' defined as type '{}' has not vlan encapsulation '{}'".format( + net_name, net_type, vim_net['encapsulation'])) + network["vlan"] = vim_net.get('segmentation_id') + try: + sdn_net_id = self.ovim.new_network(network) + except (ovimException, Exception) as e: + self.logger.error("task=%s cannot create SDN network vim_net_id=%s input='%s' ovimException='%s'", + str(task_id), net_id, str(network), str(e)) with self.db_lock: - self.db.update_rows("instance_nets", UPDATE={"vim_net_id": net_id}, WHERE={"vim_net_id": task_id}) + self.db.update_rows("instance_nets", UPDATE={"vim_net_id": net_id, "sdn_net_id": sdn_net_id}, + WHERE={"vim_net_id": task_id}) + new_refresh_task = {"status": "enqueued", + "id": task_id, + "name": "get-net", + "vim_id": net_id, + "vim_info": {} } + self._insert_refresh(new_refresh_task, time.time()) return True, net_id except db_base_Exception as e: self.logger.error("Error updating database %s", str(e)) return True, net_id except vimconn.vimconnException as e: + self.logger.error("Error creating NET, task=%s: %s", str(task_id), str(e)) + try: + with self.db_lock: + self.db.update_rows("instance_nets", + UPDATE={"error_msg": self._format_vim_error_msg(str(e)), "status": "VIM_ERROR"}, + WHERE={"vim_net_id": task_id}) + except db_base_Exception as e: + self.logger.error("Error updating database %s", str(e)) return False, str(e) + #except ovimException as e: + # self.logger.error("Error creating NET in ovim, task=%s: %s", str(task_id), str(e)) + # return False, str(e) def new_vm(self, task): try: @@ -162,33 +446,63 @@ class vim_thread(threading.Thread): task_id = task["id"] depends = task.get("depends") net_list = params[5] + error_text = "" for net in net_list: - if is_task_id(net["net_id"]): # change task_id into network_id + if "net_id" in net and is_task_id(net["net_id"]): # change task_id into network_id try: task_net = depends[net["net_id"]] with self.task_lock: if task_net["status"] == "error": - return False, "Cannot create VM because depends on a network that cannot be created: " + \ + error_text = "Cannot create VM because depends on a network that cannot be created: " +\ str(task_net["result"]) + break elif task_net["status"] == "enqueued" or task_net["status"] == "processing": - return False, "Cannot create VM because depends on a network still not created" + error_text = "Cannot create VM because depends on a network still not created" + break network_id = task_net["result"] net["net_id"] = network_id except Exception as e: - return False, "Error trying to map from task_id={} to task result: {}".format(net["net_id"], - str(e)) - vm_id = self.vim.new_vminstance(*params) - with self.db_lock: - self.db.update_rows("instance_vms", UPDATE={"vim_vm_id": vm_id}, WHERE={"vim_vm_id": task_id}) - return True, vm_id - except db_base_Exception as e: - self.logger.error("Error updtaing database %s", str(e)) + error_text = "Error trying to map from task_id={} to task result: {}".format( + net["net_id"],str(e)) + break + if not error_text: + vm_id = self.vim.new_vminstance(*params) + try: + with self.db_lock: + if error_text: + update = self.db.update_rows("instance_vms", + UPDATE={"status": "VIM_ERROR", "error_msg": error_text}, + WHERE={"vim_vm_id": task_id}) + else: + update = self.db.update_rows("instance_vms", UPDATE={"vim_vm_id": vm_id}, WHERE={"vim_vm_id": task_id}) + if not update: + self.logger.error("task id={} name={} database not updated vim_vm_id={}".format( + task["id"], task["name"], vm_id)) + except db_base_Exception as e: + self.logger.error("Error updating database %s", str(e)) + if error_text: + return False, error_text + new_refresh_task = {"status": "enqueued", + "id": task_id, + "name": "get-vm", + "vim_id": vm_id, + "vim_info": {"interfaces":[]} } + self._insert_refresh(new_refresh_task, time.time()) return True, vm_id except vimconn.vimconnException as e: + self.logger.error("Error creating VM, task=%s: %s", str(task_id), str(e)) + try: + with self.db_lock: + self.db.update_rows("instance_vms", + UPDATE={"error_msg": self._format_vim_error_msg(str(e)), "status": "VIM_ERROR"}, + WHERE={"vim_vm_id": task_id}) + except db_base_Exception as edb: + self.logger.error("Error updating database %s", str(edb)) return False, str(e) def del_vm(self, task): - vm_id = task["params"] + vm_id = task["params"][0] + interfaces = task["params"][1] if is_task_id(vm_id): try: task_create = task["depends"][vm_id] @@ -196,17 +510,28 @@ class vim_thread(threading.Thread): if task_create["status"] == "error": return True, "VM was not created. It has error: " + str(task_create["result"]) elif task_create["status"] == "enqueued" or task_create["status"] == "processing": - return False, "Cannot delete VM because still creating" + return False, "Cannot delete VM vim_id={} because still creating".format(vm_id) vm_id = task_create["result"] except Exception as e: return False, "Error trying to get task_id='{}':".format(vm_id, str(e)) try: + self._remove_refresh("get-vm", vm_id) + for iface in interfaces: + if iface.get("sdn_port_id"): + try: + self.ovim.delete_port(iface["sdn_port_id"]) + except ovimException as e: + self.logger.error("ovimException deleting external_port={} at VM vim_id={} deletion ".format( + iface["sdn_port_id"], vm_id) + str(e), exc_info=True) + # TODO Set error_msg at instance_nets + return True, self.vim.delete_vminstance(vm_id) except vimconn.vimconnException as e: return False, str(e) def del_net(self, task): - net_id = task["params"] + net_id = task["params"][0] + sdn_net_id = task["params"][1] if is_task_id(net_id): try: task_create = task["depends"][net_id] @@ -219,8 +544,16 @@ class vim_thread(threading.Thread): except Exception as e: return False, "Error trying to get task_id='{}':".format(net_id, str(e)) try: - return True, self.vim.delete_network(net_id) + self._remove_refresh("get-net", net_id) + result = self.vim.delete_network(net_id) + if sdn_net_id: + with self.db_lock: + self.ovim.delete_network(sdn_net_id) + return True, result except vimconn.vimconnException as e: return False, str(e) + except ovimException as e: + logging.error("Error deleting network from ovim. net_id: {}, sdn_net_id: {}".format(net_id, sdn_net_id)) + return False, str(e)