X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;ds=sidebyside;f=osm_ro%2Fvim_thread.py;h=f4849f48ac9abec3bd16314f4be5d6c91095ccdb;hb=HEAD;hp=a56dd64152ef10ccfa4f25798851bdf1c84e4e81;hpb=d573573c267f0db9ee21c454ed2a5a086967c81e;p=osm%2FRO.git diff --git a/osm_ro/vim_thread.py b/osm_ro/vim_thread.py deleted file mode 100644 index a56dd641..00000000 --- a/osm_ro/vim_thread.py +++ /dev/null @@ -1,559 +0,0 @@ -# -*- coding: utf-8 -*- - -## -# Copyright 2015 Telefónica Investigación y Desarrollo, S.A.U. -# This file is part of openvim -# All Rights Reserved. -# -# Licensed under the Apache License, Version 2.0 (the "License"); you may -# not use this file except in compliance with the License. You may obtain -# a copy of the License at -# -# http://www.apache.org/licenses/LICENSE-2.0 -# -# Unless required by applicable law or agreed to in writing, software -# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT -# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the -# License for the specific language governing permissions and limitations -# under the License. -# -# For those usages not covered by the Apache License, Version 2.0 please -# contact with: nfvlabs@tid.es -## - -''' -This is thread that interact with the host and the libvirt to manage VM -One thread will be launched per host -''' -__author__ = "Alfonso Tierno, Pablo Montes" -__date__ = "$10-feb-2017 12:07:15$" - -import threading -import time -import Queue -import logging -import vimconn -from db_base import db_base_Exception -from lib_osm_openvim.ovim import ovimException - - -# from logging import Logger -# import auxiliary_functions as af - - -def is_task_id(id): - return True if id[:5] == "TASK." else False - - -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, ovim=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 - """ - self.tasksResult = {} - """ It will contain a dictionary with - task_id: - status: enqueued,done,error,deleted,processing - result: VIM result, - """ - threading.Thread.__init__(self) - 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: - self.name = name - - self.logger = logging.getLogger('openmano.vim.'+self.name) - self.db = db - self.db_lock = db_lock - - 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: - self.task_queue.put(task, False) - return task["id"] - except Queue.Full: - raise vimconn.vimconnException(self.name + ": timeout inserting a task") - - def del_task(self, task): - with self.task_lock: - if task["status"] == "enqueued": - task["status"] == "deleted" - return True - else: # task["status"] == "processing" - self.task_lock.release() - return False - - def run(self): - self.logger.debug("Starting") - while True: - #TODO reload service - while True: - 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 - 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, 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, "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: - params = task["params"] - task_id = task["id"] - depends = task.get("depends") - net_list = params[5] - error_text = "" - for net in net_list: - 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": - 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": - 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: - 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"][0] - interfaces = task["params"][1] - if is_task_id(vm_id): - try: - task_create = task["depends"][vm_id] - with self.task_lock: - 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 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"][0] - sdn_net_id = task["params"][1] - if is_task_id(net_id): - try: - task_create = task["depends"][net_id] - with self.task_lock: - if task_create["status"] == "error": - return True, "net was not created. It has error: " + str(task_create["result"]) - elif task_create["status"] == "enqueued" or task_create["status"] == "processing": - return False, "Cannot delete net because still creating" - net_id = task_create["result"] - 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: - 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) - -