X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=vim_thread.py;h=1b518d3e033398b161f021760db4a111edeaeaab;hb=639520f575673d48cfb8599bfb737aa559d1e7d5;hp=d89c4d141af69984ed3838c01aa734682968a855;hpb=3fbff9b6666ea4981e953c98d236b54fb86c583b;p=osm%2FRO.git diff --git a/vim_thread.py b/vim_thread.py index d89c4d14..1b518d3e 100644 --- a/vim_thread.py +++ b/vim_thread.py @@ -34,7 +34,7 @@ import Queue import logging import vimconn from db_base import db_base_Exception -from openvim.ovim import ovimException +from ovim import ovimException # from logging import Logger @@ -77,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: @@ -99,54 +332,59 @@ 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 - - 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 + 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() + + 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): - if len(error_text) >= 1024: - return error_text[:516] + " ... " + error_text[-500:] + 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): @@ -164,17 +402,26 @@ class vim_thread(threading.Thread): network = {"name": net_name, "type": net_type} vim_net = self.vim.get_network(net_id) - if vim_net.get('network_type') != 'vlan': - raise vimconn.vimconnException(net_name + "defined as type " + net_type + " but the created network in vim is " + vim_net['provider:network_type']) - + 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') - - sdn_net_id = None - with self.db_lock: - if network: + sdn_net_id = None + try: sdn_net_id = self.ovim.new_network(network) - self.db.update_rows("instance_nets", UPDATE={"vim_net_id": net_id, "sdn_net_id": sdn_net_id}, WHERE={"vim_net_id": task_id}) - + 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, "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)) @@ -189,9 +436,9 @@ class vim_thread(threading.Thread): 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) + #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: @@ -199,27 +446,48 @@ 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) + 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: - self.db.update_rows("instance_vms", UPDATE={"vim_vm_id": vm_id}, WHERE={"vim_vm_id": task_id}) + 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)) @@ -228,12 +496,13 @@ class vim_thread(threading.Thread): 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 e: - self.logger.error("Error updating database %s", str(e)) + 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] @@ -241,11 +510,21 @@ 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) @@ -265,6 +544,7 @@ class vim_thread(threading.Thread): except Exception as e: return False, "Error trying to get task_id='{}':".format(net_id, str(e)) try: + self._remove_refresh("get-net", net_id) result = self.vim.delete_network(net_id) if sdn_net_id: with self.db_lock: