X-Git-Url: https://osm.etsi.org/gitweb/?p=osm%2FRO.git;a=blobdiff_plain;f=RO%2Fosm_ro%2Fvim_thread.py;h=622cccb57d3b930be01a20bc7ffb7dadf28d7340;hp=8d397c2364d0a25a2f306213e344920bbc58f37e;hb=79870eac1909757c72047fc0b7df1444ce566fc5;hpb=57dadcfd0fcf3c8e933602e2fb57f32658d9c845 diff --git a/RO/osm_ro/vim_thread.py b/RO/osm_ro/vim_thread.py index 8d397c23..622cccb5 100644 --- a/RO/osm_ro/vim_thread.py +++ b/RO/osm_ro/vim_thread.py @@ -293,9 +293,6 @@ class vim_thread(threading.Thread): }) continue - # task of creation must be the first in the list of related_task - assert(related_tasks[0]["action"] in ("CREATE", "FIND")) - task["params"] = None if task["extra"]: extra = yaml.load(task["extra"], Loader=yaml.Loader) @@ -323,6 +320,8 @@ class vim_thread(threading.Thread): copy_to["sdn_net_id"] = copy_from["sdn_net_id"] if copy_from.get("interfaces"): copy_to["interfaces"] = copy_from["interfaces"] + if copy_from.get("sdn-ports"): + copy_to["sdn-ports"] = copy_from["sdn-ports"] if copy_from.get("created_items"): if not copy_to.get("created_items"): copy_to["created_items"] = {} @@ -330,11 +329,13 @@ class vim_thread(threading.Thread): task_create = None dependency_task = None - deletion_needed = False + deletion_needed = task["extra"].get("created", False) if task["status"] == "FAILED": return # TODO need to be retry?? try: - # get all related tasks + # get all related tasks. task of creation must be the first in the list of related_task, + # unless the deletion fails and it is pendingit fails + # TODO this should be removed, passing related_tasks related_tasks = self.db.get_rows(FROM="vim_wim_actions", WHERE={self.target_k: self.target_v, "status": ['SCHEDULED', 'BUILD', 'DONE', 'FAILED'], @@ -358,13 +359,14 @@ class vim_thread(threading.Thread): break # mark task_create as FINISHED - self.db.update_rows("vim_wim_actions", UPDATE={"status": "FINISHED"}, - WHERE={self.target_k: self.target_v, - "instance_action_id": task_create["instance_action_id"], - "task_index": task_create["task_index"] - }) + if task_create: + self.db.update_rows("vim_wim_actions", UPDATE={"status": "FINISHED"}, + WHERE={self.target_k: self.target_v, + "instance_action_id": task_create["instance_action_id"], + "task_index": task_create["task_index"] + }) if not deletion_needed: - return + return False elif dependency_task: # move create information from task_create to relate_task extra_new_created = yaml.load(dependency_task["extra"], Loader=yaml.Loader) or {} @@ -380,10 +382,19 @@ class vim_thread(threading.Thread): "task_index": dependency_task["task_index"] }) return False - else: + elif task_create: task["vim_id"] = task_create["vim_id"] copy_extra_created(copy_to=task["extra"], copy_from=task_create["extra"]) + # Ensure this task extra information is stored at database + self.db.update_rows("vim_wim_actions", + UPDATE={"extra": yaml.safe_dump(task["extra"], default_flow_style=True, + width=256)}, + WHERE={self.target_k: self.target_v, + "instance_action_id": task["instance_action_id"], + "task_index": task["task_index"], + }) return True + return deletion_needed except Exception as e: self.logger.critical("Unexpected exception at _delete_task: " + str(e), exc_info=True) @@ -548,6 +559,7 @@ class vim_thread(threading.Thread): old_task_status = task["status"] create_or_find = False # if as result of processing this task something is created or found next_refresh = 0 + task_id = task["instance_action_id"] + "." + str(task["task_index"]) try: if task["status"] == "SCHEDULED": @@ -606,9 +618,10 @@ class vim_thread(threading.Thread): elif not self.vim and not self.sdnconnector: task["status"] = "FAILED" task["error_msg"] = self.error_status - database_update = {"status": "VIM_ERROR", "error_msg": task["error_msg"]} + database_update = {"status": "VIM_ERROR" if self.datacenter_tenant_id else "WIM_ERROR", + "error_msg": task["error_msg"]} elif task["item_id"] != related_tasks[0]["item_id"] and task["action"] in ("FIND", "CREATE"): - # Do nothing, just copy values from one to another and updata database + # Do nothing, just copy values from one to another and update database task["status"] = related_tasks[0]["status"] task["error_msg"] = related_tasks[0]["error_msg"] task["vim_id"] = related_tasks[0]["vim_id"] @@ -659,7 +672,10 @@ class vim_thread(threading.Thread): else: raise vimconn.vimconnException(self.name + "unknown task action {}".format(task["action"])) elif task["item"] == 'instance_sfis': - if task["action"] == "CREATE": + if task["status"] in ('BUILD', 'DONE') and task["action"] in ("FIND", "CREATE"): + database_update = self._refres_sfis(task) + create_or_find = True + elif task["action"] == "CREATE": create_or_find = True database_update = self.new_sfi(task) elif task["action"] == "DELETE": @@ -667,7 +683,10 @@ class vim_thread(threading.Thread): else: raise vimconn.vimconnException(self.name + "unknown task action {}".format(task["action"])) elif task["item"] == 'instance_sfs': - if task["action"] == "CREATE": + if task["status"] in ('BUILD', 'DONE') and task["action"] in ("FIND", "CREATE"): + database_update = self._refres_sfs(task) + create_or_find = True + elif task["action"] == "CREATE": create_or_find = True database_update = self.new_sf(task) elif task["action"] == "DELETE": @@ -675,7 +694,10 @@ class vim_thread(threading.Thread): else: raise vimconn.vimconnException(self.name + "unknown task action {}".format(task["action"])) elif task["item"] == 'instance_classifications': - if task["action"] == "CREATE": + if task["status"] in ('BUILD', 'DONE') and task["action"] in ("FIND", "CREATE"): + database_update = self._refres_classifications(task) + create_or_find = True + elif task["action"] == "CREATE": create_or_find = True database_update = self.new_classification(task) elif task["action"] == "DELETE": @@ -683,7 +705,10 @@ class vim_thread(threading.Thread): else: raise vimconn.vimconnException(self.name + "unknown task action {}".format(task["action"])) elif task["item"] == 'instance_sfps': - if task["action"] == "CREATE": + if task["status"] in ('BUILD', 'DONE') and task["action"] in ("FIND", "CREATE"): + database_update = self._refres_sfps(task) + create_or_find = True + elif task["action"] == "CREATE": create_or_find = True database_update = self.new_sfp(task) elif task["action"] == "DELETE": @@ -693,16 +718,17 @@ class vim_thread(threading.Thread): else: raise vimconn.vimconnException(self.name + "unknown task item {}".format(task["item"])) # TODO - except VimThreadException as e: + except Exception as e: + if not isinstance(e, VimThreadException): + self.logger.error("Error executing task={}: {}".format(task_id, e), exc_info=True) task["error_msg"] = str(e) task["status"] = "FAILED" database_update = {"status": "VIM_ERROR", "error_msg": task["error_msg"]} - if task["item"] == 'instance_vms': - database_update["vim_vm_id"] = None - elif task["item"] == 'instance_nets': - database_update["vim_net_id"] = None + # if task["item"] == 'instance_vms': + # database_update["vim_vm_id"] = None + # elif task["item"] == 'instance_nets': + # database_update["vim_net_id"] = None - task_id = task["instance_action_id"] + "." + str(task["task_index"]) self.logger.debug("task={} item={} action={} result={}:'{}' params={}".format( task_id, task["item"], task["action"], task["status"], task["vim_id"] if task["status"] == "DONE" else task.get("error_msg"), task["params"])) @@ -712,7 +738,7 @@ class vim_thread(threading.Thread): next_refresh = time.time() if task["extra"].get("vim_status") == "BUILD": next_refresh += self.REFRESH_BUILD - elif task["extra"].get("vim_status") in ("ERROR", "VIM_ERROR"): + elif task["extra"].get("vim_status") in ("ERROR", "VIM_ERROR", "WIM_ERROR"): next_refresh += self.REFRESH_ERROR elif task["extra"].get("vim_status") == "DELETED": next_refresh += self.REFRESH_DELETE @@ -760,7 +786,7 @@ class vim_thread(threading.Thread): if database_update: where_filter = {"related": task["related"]} if task["item"] == "instance_nets" and task["datacenter_vim_id"]: - where_filter["datacenter_tenant_id"] = task["datacenter_vim_id"] + where_filter["datacenter_tenant_id"] = task["datacenter_vim_id"] self.db.update_rows(table=task["item"], UPDATE=database_update, WHERE=where_filter) @@ -1106,7 +1132,7 @@ class vim_thread(threading.Thread): except Exception as e: if isinstance(e, SdnConnectorError) and e.http_error == HTTPStatus.NOT_FOUND.value: pass - else: + else: self._proccess_sdn_exception(e) params = task["params"] @@ -1114,19 +1140,21 @@ class vim_thread(threading.Thread): # look for ports sdn_ports = [] pending_ports = 0 + vlan_used = None ports = self.db.get_rows(FROM='instance_interfaces', WHERE={'instance_wim_net_id': task["item_id"]}) sdn_need_update = False for port in ports: + vlan_used = port.get("vlan") or vlan_used # TODO. Do not connect if already done if port.get("compute_node") and port.get("pci"): - for map in self.port_mappings: - if map.get("device_id") == port["compute_node"] and \ - map.get("device_interface_id") == port["pci"]: + for pmap in self.port_mappings: + if pmap.get("device_id") == port["compute_node"] and \ + pmap.get("device_interface_id") == port["pci"]: break else: if self.sdnconn_config.get("mapping_not_needed"): - map = { + pmap = { "service_endpoint_id": "{}:{}".format(port["compute_node"], port["pci"]), "service_endpoint_encapsulation_info": { "vlan": port["vlan"], @@ -1136,25 +1164,25 @@ class vim_thread(threading.Thread): } } else: - map = None + pmap = None error_list.append("Port mapping not found for compute_node={} pci={}".format( port["compute_node"], port["pci"])) - if map: - if port["uuid"] not in connected_ports or port["modified_at"] > last_update: + if pmap: + if port["modified_at"] > last_update: sdn_need_update = True new_connected_ports.append(port["uuid"]) sdn_ports.append({ - "service_endpoint_id": map["service_endpoint_id"], + "service_endpoint_id": pmap["service_endpoint_id"], "service_endpoint_encapsulation_type": "dot1q" if port["model"] == "SR-IOV" else None, "service_endpoint_encapsulation_info": { "vlan": port["vlan"], "mac": port["mac_address"], - "device_id": map.get("device_id"), - "device_interface_id": map.get("device_interface_id"), - "switch_dpid": map.get("switch_dpid"), - "switch_port": map.get("switch_port"), - "service_mapping_info": map.get("service_mapping_info"), + "device_id": pmap.get("device_id"), + "device_interface_id": pmap.get("device_interface_id"), + "switch_dpid": pmap.get("switch_dpid"), + "switch_port": pmap.get("switch_port"), + "service_mapping_info": pmap.get("service_mapping_info"), } }) @@ -1163,8 +1191,28 @@ class vim_thread(threading.Thread): if pending_ports: error_list.append("Waiting for getting interfaces location from VIM. Obtained '{}' of {}" .format(len(ports)-pending_ports, len(ports))) + + # connect external ports + for index, external_port in enumerate(task["extra"].get("sdn-ports") or ()): + external_port_id = external_port.get("service_endpoint_id") or str(index) + sdn_ports.append({ + "service_endpoint_id": external_port_id, + "service_endpoint_encapsulation_type": external_port.get("service_endpoint_encapsulation_type", + "dot1q"), + "service_endpoint_encapsulation_info": { + "vlan": external_port.get("vlan") or vlan_used, + "mac": external_port.get("mac_address"), + "device_id": external_port.get("device_id"), + "device_interface_id": external_port.get("device_interface_id"), + "switch_dpid": external_port.get("switch_dpid") or external_port.get("switch_id"), + "switch_port": external_port.get("switch_port"), + "service_mapping_info": external_port.get("service_mapping_info"), + }}) + new_connected_ports.append(external_port_id) + # if there are more ports to connect or they have been modified, call create/update - if sdn_need_update and len(sdn_ports) >= 2: + if (set(connected_ports) != set(new_connected_ports) or sdn_need_update) and len(sdn_ports) >= 2: + last_update = time.time() if not wimconn_net_id: if params[0] == "data": net_type = "ELAN" @@ -1177,10 +1225,9 @@ class vim_thread(threading.Thread): else: created_items = self.sdnconnector.edit_connectivity_service(wimconn_net_id, conn_info=created_items, connection_points=sdn_ports) - last_update = time.time() connected_ports = new_connected_ports elif wimconn_net_id: - try: + try: wim_status_dict = self.sdnconnector.get_connectivity_service_status(wimconn_net_id, conn_info=created_items) sdn_status = wim_status_dict["sdn_status"] @@ -1194,7 +1241,7 @@ class vim_thread(threading.Thread): task["status"] = "DONE" task["extra"]["vim_info"] = {} # task["extra"]["sdn_net_id"] = sdn_net_id - task["extra"]["vim_status"] = "BUILD" + task["extra"]["vim_status"] = sdn_status task["extra"]["created"] = True task["extra"]["created_items"] = created_items task["extra"]["connected_ports"] = connected_ports @@ -1282,7 +1329,7 @@ class vim_thread(threading.Thread): task["extra"]["created"] = True task["extra"]["vim_status"] = "ACTIVE" task["error_msg"] = None - task["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing + task["status"] = "DONE" task["vim_id"] = vim_sfi_id instance_element_update = {"status": "ACTIVE", "vim_sfi_id": vim_sfi_id, "error_msg": None} return instance_element_update @@ -1331,7 +1378,7 @@ class vim_thread(threading.Thread): task["extra"]["created"] = True task["extra"]["vim_status"] = "ACTIVE" task["error_msg"] = None - task["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing + task["status"] = "DONE" task["vim_id"] = vim_sf_id instance_element_update = {"status": "ACTIVE", "vim_sf_id": vim_sf_id, "error_msg": None} return instance_element_update @@ -1369,12 +1416,27 @@ class vim_thread(threading.Thread): task_id = task["instance_action_id"] + "." + str(task["task_index"]) dep_id = "TASK-" + str(task["extra"]["depends_on"][0]) error_text = "" - interfaces = task.get("depends").get(dep_id).get("extra").get("interfaces").keys() + interfaces = task.get("depends").get(dep_id).get("extra").get("interfaces") # Bear in mind that different VIM connectors might support Classifications differently. # In the case of OpenStack, only the first VNF attached to the classifier will be used # to create the Classification(s) (the "logical source port" of the "Flow Classifier"). # Since the VNFFG classifier match lacks the ethertype, classification defaults to # using the IPv4 flow classifier. + logical_source_port_vim_id = None + logical_source_port_id = params.get("logical_source_port") + for vim_interface, interface_data in interfaces.items(): + if interface_data.get("interface_id") == logical_source_port_id: + logical_source_port_vim_id = vim_interface + break + if not logical_source_port_vim_id: + error_text = "Error creating Flow Classifier, Logical Source Port id {}".format( + logical_source_port_id) + self.logger.error(error_text) + task["error_msg"] = error_text + task["status"] = "FAILED" + task["vim_id"] = None + return None + name = "c-{}".format(task["item_id"][:8]) # if not CIDR is given for the IP addresses, add /32: ip_proto = int(params.get("ip_proto")) @@ -1382,7 +1444,7 @@ class vim_thread(threading.Thread): destination_ip = params.get("destination_ip") source_port = params.get("source_port") destination_port = params.get("destination_port") - definition = {"logical_source_port": interfaces[0]} + definition = {"logical_source_port": logical_source_port_vim_id} if ip_proto: if ip_proto == 1: ip_proto = 'icmp' @@ -1412,7 +1474,7 @@ class vim_thread(threading.Thread): task["extra"]["created"] = True task["extra"]["vim_status"] = "ACTIVE" task["error_msg"] = None - task["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing + task["status"] = "DONE" task["vim_id"] = vim_classification_id instance_element_update = {"status": "ACTIVE", "vim_classification_id": vim_classification_id, "error_msg": None} @@ -1468,7 +1530,7 @@ class vim_thread(threading.Thread): task["extra"]["created"] = True task["extra"]["vim_status"] = "ACTIVE" task["error_msg"] = None - task["status"] = "FINISHED" # with FINISHED instead of DONE it will not be refreshing + task["status"] = "DONE" task["vim_id"] = vim_sfp_id instance_element_update = {"status": "ACTIVE", "vim_sfp_id": vim_sfp_id, "error_msg": None} return instance_element_update @@ -1498,3 +1560,149 @@ class vim_thread(threading.Thread): return None task["status"] = "FAILED" return None + + def _refres_sfps(self, task): + """Call VIM to get SFPs status""" + database_update = None + + vim_id = task["vim_id"] + sfp_to_refresh_list = [vim_id] + task_id = task["instance_action_id"] + "." + str(task["task_index"]) + try: + vim_dict = self.vim.refresh_sfps_status(sfp_to_refresh_list) + vim_info = vim_dict[vim_id] + except vimconn.vimconnException as e: + # Mark all tasks at VIM_ERROR status + self.logger.error("task={} get-sfp: vimconnException when trying to refresh sfps {}".format(task_id, e)) + vim_info = {"status": "VIM_ERROR", "error_msg": str(e)} + + self.logger.debug("task={} get-sfp: vim_sfp_id={} result={}".format(task_id, task["vim_id"], vim_info)) + #TODO: Revise this part + vim_info_error_msg = None + if vim_info.get("error_msg"): + vim_info_error_msg = self._format_vim_error_msg(vim_info["error_msg"]) + task_vim_info = task["extra"].get("vim_info") + task_error_msg = task.get("error_msg") + task_vim_status = task["extra"].get("vim_status") + if task_vim_status != vim_info["status"] or task_error_msg != vim_info_error_msg or \ + (vim_info.get("vim_info") and task_vim_info != vim_info["vim_info"]): + database_update = {"status": vim_info["status"], "error_msg": vim_info_error_msg} + if vim_info.get("vim_info"): + database_update["vim_info"] = vim_info["vim_info"] + + task["extra"]["vim_status"] = vim_info["status"] + task["error_msg"] = vim_info_error_msg + if vim_info.get("vim_info"): + task["extra"]["vim_info"] = vim_info["vim_info"] + + return database_update + + def _refres_sfis(self, task): + """Call VIM to get sfis status""" + database_update = None + + vim_id = task["vim_id"] + sfi_to_refresh_list = [vim_id] + task_id = task["instance_action_id"] + "." + str(task["task_index"]) + try: + vim_dict = self.vim.refresh_sfis_status(sfi_to_refresh_list) + vim_info = vim_dict[vim_id] + except vimconn.vimconnException as e: + # Mark all tasks at VIM_ERROR status + self.logger.error("task={} get-sfi: vimconnException when trying to refresh sfis {}".format(task_id, e)) + vim_info = {"status": "VIM_ERROR", "error_msg": str(e)} + + self.logger.debug("task={} get-sfi: vim_sfi_id={} result={}".format(task_id, task["vim_id"], vim_info)) + #TODO: Revise this part + vim_info_error_msg = None + if vim_info.get("error_msg"): + vim_info_error_msg = self._format_vim_error_msg(vim_info["error_msg"]) + task_vim_info = task["extra"].get("vim_info") + task_error_msg = task.get("error_msg") + task_vim_status = task["extra"].get("vim_status") + if task_vim_status != vim_info["status"] or task_error_msg != vim_info_error_msg or \ + (vim_info.get("vim_info") and task_vim_info != vim_info["vim_info"]): + database_update = {"status": vim_info["status"], "error_msg": vim_info_error_msg} + if vim_info.get("vim_info"): + database_update["vim_info"] = vim_info["vim_info"] + + task["extra"]["vim_status"] = vim_info["status"] + task["error_msg"] = vim_info_error_msg + if vim_info.get("vim_info"): + task["extra"]["vim_info"] = vim_info["vim_info"] + + return database_update + + def _refres_sfs(self, task): + """Call VIM to get sfs status""" + database_update = None + + vim_id = task["vim_id"] + sf_to_refresh_list = [vim_id] + task_id = task["instance_action_id"] + "." + str(task["task_index"]) + try: + vim_dict = self.vim.refresh_sfs_status(sf_to_refresh_list) + vim_info = vim_dict[vim_id] + except vimconn.vimconnException as e: + # Mark all tasks at VIM_ERROR status + self.logger.error("task={} get-sf: vimconnException when trying to refresh sfs {}".format(task_id, e)) + vim_info = {"status": "VIM_ERROR", "error_msg": str(e)} + + self.logger.debug("task={} get-sf: vim_sf_id={} result={}".format(task_id, task["vim_id"], vim_info)) + #TODO: Revise this part + vim_info_error_msg = None + if vim_info.get("error_msg"): + vim_info_error_msg = self._format_vim_error_msg(vim_info["error_msg"]) + task_vim_info = task["extra"].get("vim_info") + task_error_msg = task.get("error_msg") + task_vim_status = task["extra"].get("vim_status") + if task_vim_status != vim_info["status"] or task_error_msg != vim_info_error_msg or \ + (vim_info.get("vim_info") and task_vim_info != vim_info["vim_info"]): + database_update = {"status": vim_info["status"], "error_msg": vim_info_error_msg} + if vim_info.get("vim_info"): + database_update["vim_info"] = vim_info["vim_info"] + + task["extra"]["vim_status"] = vim_info["status"] + task["error_msg"] = vim_info_error_msg + if vim_info.get("vim_info"): + task["extra"]["vim_info"] = vim_info["vim_info"] + + return database_update + + def _refres_classifications(self, task): + """Call VIM to get classifications status""" + database_update = None + + vim_id = task["vim_id"] + classification_to_refresh_list = [vim_id] + task_id = task["instance_action_id"] + "." + str(task["task_index"]) + try: + vim_dict = self.vim.refresh_classifications_status(classification_to_refresh_list) + vim_info = vim_dict[vim_id] + except vimconn.vimconnException as e: + # Mark all tasks at VIM_ERROR status + self.logger.error("task={} get-classification: vimconnException when trying to refresh classifications {}" + .format(task_id, e)) + vim_info = {"status": "VIM_ERROR", "error_msg": str(e)} + + self.logger.debug("task={} get-classification: vim_classification_id={} result={}".format(task_id, + task["vim_id"], vim_info)) + #TODO: Revise this part + vim_info_error_msg = None + if vim_info.get("error_msg"): + vim_info_error_msg = self._format_vim_error_msg(vim_info["error_msg"]) + task_vim_info = task["extra"].get("vim_info") + task_error_msg = task.get("error_msg") + task_vim_status = task["extra"].get("vim_status") + if task_vim_status != vim_info["status"] or task_error_msg != vim_info_error_msg or \ + (vim_info.get("vim_info") and task_vim_info != vim_info["vim_info"]): + database_update = {"status": vim_info["status"], "error_msg": vim_info_error_msg} + if vim_info.get("vim_info"): + database_update["vim_info"] = vim_info["vim_info"] + + task["extra"]["vim_status"] = vim_info["status"] + task["error_msg"] = vim_info_error_msg + if vim_info.get("vim_info"): + task["extra"]["vim_info"] = vim_info["vim_info"] + + return database_update