X-Git-Url: https://osm.etsi.org/gitweb/?a=blobdiff_plain;f=NG-RO%2Fosm_ng_ro%2Fns_thread.py;h=8bb8d4bca25032747c28c8e747014b5143f5fed4;hb=refs%2Fchanges%2F05%2F12005%2F13;hp=d728d7024659d425cfe98c471f459f47bbb0348c;hpb=79ac6df793d0c47de98f6b3a57026aef6bdbb1f3;p=osm%2FRO.git diff --git a/NG-RO/osm_ng_ro/ns_thread.py b/NG-RO/osm_ng_ro/ns_thread.py index d728d702..8bb8d4bc 100644 --- a/NG-RO/osm_ng_ro/ns_thread.py +++ b/NG-RO/osm_ng_ro/ns_thread.py @@ -32,6 +32,7 @@ import queue from shutil import rmtree import threading import time +import traceback from unittest.mock import Mock from importlib_metadata import entry_points @@ -390,6 +391,11 @@ class VimInteractionVdu(VimInteractionBase): vim_vm_id, created_items = target_vim.new_vminstance(**params_copy) interfaces = [iface["vim_id"] for iface in params_copy["net_list"]] + # add to created items previous_created_volumes (healing) + if task.get("previous_created_volumes"): + for k, v in task["previous_created_volumes"].items(): + created_items[k] = v + ro_vim_item_update = { "vim_id": vim_vm_id, "vim_status": "BUILD", @@ -399,6 +405,7 @@ class VimInteractionVdu(VimInteractionBase): "vim_message": None, "interfaces_vim_ids": interfaces, "interfaces": [], + "interfaces_backup": [], } self.logger.debug( "task={} {} new-vm={} created={}".format( @@ -604,6 +611,7 @@ class VimInteractionVdu(VimInteractionBase): except (vimconn.VimConnException, NsWorkerException) as e: retries += 1 + self.logger.debug(traceback.format_exc()) if retries < self.max_retries_inject_ssh_key: return ( "BUILD", @@ -879,6 +887,48 @@ class VimInteractionAffinityGroup(VimInteractionBase): return "FAILED", ro_vim_item_update +class VimInteractionUpdateVdu(VimInteractionBase): + def exec(self, ro_task, task_index, task_depends): + task = ro_task["tasks"][task_index] + task_id = task["task_id"] + db_task_update = {"retries": 0} + created = False + created_items = {} + target_vim = self.my_vims[ro_task["target_id"]] + + try: + if task.get("params"): + vim_vm_id = task["params"].get("vim_vm_id") + action = task["params"].get("action") + context = {action: action} + target_vim.action_vminstance(vim_vm_id, context) + # created = True + ro_vim_item_update = { + "vim_id": vim_vm_id, + "vim_status": "DONE", + "created": created, + "created_items": created_items, + "vim_details": None, + "vim_message": None, + } + self.logger.debug( + "task={} {} vm-migration done".format(task_id, ro_task["target_id"]) + ) + return "DONE", ro_vim_item_update, db_task_update + except (vimconn.VimConnException, NsWorkerException) as e: + self.logger.error( + "task={} vim={} VM Migration:" + " {}".format(task_id, ro_task["target_id"], e) + ) + ro_vim_item_update = { + "vim_status": "VIM_ERROR", + "created": created, + "vim_message": str(e), + } + + return "FAILED", ro_vim_item_update, db_task_update + + class VimInteractionSdnNet(VimInteractionBase): @staticmethod def _match_pci(port_pci, mapping): @@ -1271,6 +1321,156 @@ class VimInteractionSdnNet(VimInteractionBase): return "DONE", ro_vim_item_update_ok +class VimInteractionMigration(VimInteractionBase): + def exec(self, ro_task, task_index, task_depends): + task = ro_task["tasks"][task_index] + task_id = task["task_id"] + db_task_update = {"retries": 0} + target_vim = self.my_vims[ro_task["target_id"]] + vim_interfaces = [] + created = False + created_items = {} + refreshed_vim_info = {} + + try: + if task.get("params"): + vim_vm_id = task["params"].get("vim_vm_id") + migrate_host = task["params"].get("migrate_host") + _, migrated_compute_node = target_vim.migrate_instance( + vim_vm_id, migrate_host + ) + + if migrated_compute_node: + # When VM is migrated, vdu["vim_info"] needs to be updated + vdu_old_vim_info = task["params"]["vdu_vim_info"].get( + ro_task["target_id"] + ) + + # Refresh VM to get new vim_info + vm_to_refresh_list = [vim_vm_id] + vim_dict = target_vim.refresh_vms_status(vm_to_refresh_list) + refreshed_vim_info = vim_dict[vim_vm_id] + + if refreshed_vim_info.get("interfaces"): + for old_iface in vdu_old_vim_info.get("interfaces"): + iface = next( + ( + iface + for iface in refreshed_vim_info["interfaces"] + if old_iface["vim_interface_id"] + == iface["vim_interface_id"] + ), + None, + ) + vim_interfaces.append(iface) + + ro_vim_item_update = { + "vim_id": vim_vm_id, + "vim_status": "ACTIVE", + "created": created, + "created_items": created_items, + "vim_details": None, + "vim_message": None, + } + + if refreshed_vim_info and refreshed_vim_info.get("status") not in ( + "ERROR", + "VIM_ERROR", + ): + ro_vim_item_update["vim_details"] = refreshed_vim_info["vim_info"] + + if vim_interfaces: + ro_vim_item_update["interfaces"] = vim_interfaces + + self.logger.debug( + "task={} {} vm-migration done".format(task_id, ro_task["target_id"]) + ) + + return "DONE", ro_vim_item_update, db_task_update + + except (vimconn.VimConnException, NsWorkerException) as e: + self.logger.error( + "task={} vim={} VM Migration:" + " {}".format(task_id, ro_task["target_id"], e) + ) + ro_vim_item_update = { + "vim_status": "VIM_ERROR", + "created": created, + "vim_message": str(e), + } + + return "FAILED", ro_vim_item_update, db_task_update + + +class VimInteractionResize(VimInteractionBase): + def exec(self, ro_task, task_index, task_depends): + task = ro_task["tasks"][task_index] + task_id = task["task_id"] + db_task_update = {"retries": 0} + created = False + target_flavor_uuid = None + created_items = {} + refreshed_vim_info = {} + target_vim = self.my_vims[ro_task["target_id"]] + + try: + if task.get("params"): + vim_vm_id = task["params"].get("vim_vm_id") + flavor_dict = task["params"].get("flavor_dict") + self.logger.info("flavor_dict %s", flavor_dict) + + try: + target_flavor_uuid = target_vim.get_flavor_id_from_data(flavor_dict) + except Exception as e: + self.logger.info("Cannot find any flavor matching %s.", str(e)) + try: + target_flavor_uuid = target_vim.new_flavor(flavor_dict) + except Exception as e: + self.logger.error("Error creating flavor at VIM %s.", str(e)) + + if target_flavor_uuid is not None: + resized_status = target_vim.resize_instance( + vim_vm_id, target_flavor_uuid + ) + + if resized_status: + # Refresh VM to get new vim_info + vm_to_refresh_list = [vim_vm_id] + vim_dict = target_vim.refresh_vms_status(vm_to_refresh_list) + refreshed_vim_info = vim_dict[vim_vm_id] + + ro_vim_item_update = { + "vim_id": vim_vm_id, + "vim_status": "DONE", + "created": created, + "created_items": created_items, + "vim_details": None, + "vim_message": None, + } + + if refreshed_vim_info and refreshed_vim_info.get("status") not in ( + "ERROR", + "VIM_ERROR", + ): + ro_vim_item_update["vim_details"] = refreshed_vim_info["vim_info"] + + self.logger.debug( + "task={} {} resize done".format(task_id, ro_task["target_id"]) + ) + return "DONE", ro_vim_item_update, db_task_update + except (vimconn.VimConnException, NsWorkerException) as e: + self.logger.error( + "task={} vim={} Resize:" " {}".format(task_id, ro_task["target_id"], e) + ) + ro_vim_item_update = { + "vim_status": "VIM_ERROR", + "created": created, + "vim_message": str(e), + } + + return "FAILED", ro_vim_item_update, db_task_update + + class NsWorker(threading.Thread): REFRESH_BUILD = 5 # 5 seconds REFRESH_ACTIVE = 60 # 1 minute @@ -1315,9 +1515,18 @@ class NsWorker(threading.Thread): "sdn_net": VimInteractionSdnNet( self.db, self.my_vims, self.db_vims, self.logger ), + "update": VimInteractionUpdateVdu( + self.db, self.my_vims, self.db_vims, self.logger + ), "affinity-or-anti-affinity-group": VimInteractionAffinityGroup( self.db, self.my_vims, self.db_vims, self.logger ), + "migrate": VimInteractionMigration( + self.db, self.my_vims, self.db_vims, self.logger + ), + "verticalscale": VimInteractionResize( + self.db, self.my_vims, self.db_vims, self.logger + ), } self.time_last_task_processed = None # lists of tasks to delete because nsrs or vnfrs has been deleted from db @@ -2324,6 +2533,7 @@ class NsWorker(threading.Thread): "vim_name", "vim_status", "interfaces", + "interfaces_backup", ) } @@ -2377,6 +2587,21 @@ class NsWorker(threading.Thread): ).split(";")[0] self.db.set_one(table, q_filter={"_id": _id}, update_dict=update_dict) + + # If interfaces exists, it backups VDU interfaces in the DB for healing operations + if ro_vim_item_update.get("interfaces"): + search_key = path_vim_status + ".interfaces" + if update_dict.get(search_key): + interfaces_backup_update = { + path_vim_status + ".interfaces_backup": update_dict[search_key] + } + + self.db.set_one( + table, + q_filter={"_id": _id}, + update_dict=interfaces_backup_update, + ) + else: update_dict = {path_item + ".status": "DELETED"} self.db.set_one(